From 72211e6ddafed40f42f973f1a6fa596079af4805 Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 5 Jun 2026 13:57:30 +0800 Subject: [PATCH] feat(captain): align streaming fallbacks --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 59 +++++++++---- .../handler/api/v1/captain_task_handler.go | 34 +++++--- .../api/v1/captain_task_handler_test.go | 82 +++++++++++++++++- internal/handler/api/v1/copilot_handler.go | 22 ++++- .../api/v1/copilot_thread_handler_test.go | 42 +++++++++ internal/handler/api/v1/sse_stream_handler.go | 5 ++ internal/service/captain_task_service.go | 86 +++++++++++++++---- .../captain_task_service_streaming_test.go | 66 +++++++++++--- 8 files changed, 329 insertions(+), 67 deletions(-) diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index e68293c1..e70c0c33 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md @@ -17,9 +17,9 @@ Build GoChat as a Go backend that can directly reuse the frontend from `referenc ## Current Baseline - Plan freeze checkpoint: 2026-06-05 docs-only tracker landing after `13cb750 feat(captain): align task payload persistence`. -- Latest implementation checkpoint: `13cb750 feat(captain): align task payload persistence`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(captain): align streaming fallbacks`. - Latest documentation checkpoint before this freeze: `3263ed9 docs: land copilot task execution plan`; this document now carries the active follow-up plan directly. -- Worktree status at this implementation checkpoint: B11.1a aligns Captain assistant CRUD/tools/inbox bindings; B11.1b aligns Captain scenarios and custom tools; B11.1c aligns Captain documents, assistant responses, bulk actions, and custom-tool test payloads; B11.2 aligns Copilot thread/message create/list/get/delete payloads, account/user scoping, and no-LLM fallback persistence; B11.3a aligns Captain preferences show/update payloads and account-level model/feature storage; B11.3b aligns Captain playground request/response payloads, account scoping, v2 history handling, and no-LLM fallback; B11.3c adds the fakeable Captain document sync backend gate with disabled, failed, and fake-success states; B11.3d aligns Captain task request/response payloads, no-provider disabled states, follow-up context, suggestion persistence, and Copilot message tool-call key validation. Next active implementation slice is B11.3e streaming/realtime compatibility, followed by B12 reused frontend smoke. +- Worktree status at this implementation checkpoint: B11.1a aligns Captain assistant CRUD/tools/inbox bindings; B11.1b aligns Captain scenarios and custom tools; B11.1c aligns Captain documents, assistant responses, bulk actions, and custom-tool test payloads; B11.2 aligns Copilot thread/message create/list/get/delete payloads, account/user scoping, and no-LLM fallback persistence; B11.3a aligns Captain preferences show/update payloads and account-level model/feature storage; B11.3b aligns Captain playground request/response payloads, account scoping, v2 history handling, and no-LLM fallback; B11.3c adds the fakeable Captain document sync backend gate with disabled, failed, and fake-success states; B11.3d aligns Captain task request/response payloads, no-provider disabled states, follow-up context, suggestion persistence, and Copilot message tool-call key validation; B11.3e aligns Captain stream DTOs/disabled SSE fallbacks and Copilot push-event payload shapes. Next active implementation slice is B12 reused frontend smoke. - `go test ./...` passes. - Route dump succeeds with `TOTAL: 830` after adding the Chatwoot-compatible applied-SLA index route. - Route parity artifacts now exist under `docs/parity/` and are generated by `cmd/route_parity`. @@ -45,17 +45,16 @@ Next ordered checkpoints: | Order | Slice | Required outcome | Primary verification | | --- | --- | --- | --- | -| 1 | B11.3e | Streaming/realtime paths have frontend-compatible non-streaming or disabled fallbacks until true streaming is wired. | Deterministic response tests and event/push payload tests. | -| 2 | B12 | Reused Chatwoot frontend smoke runs repeatably against GoChat. | Checked smoke command plus gap report under `docs/parity/`. | -| 3 | Phase 5 job parity | Any async behavior deferred from B7-B11 has a durable worker decision, fakeable delivery boundary, and replay/idempotency tests. | Focused worker/job tests plus a Phase 5 gap ledger update. | -| 4 | Phase 2/3 audit pass | Route/controller/serializer drift found by B12 or new reference inspection is captured as named slices, not free-form TODOs. | Regenerated route parity artifacts and fixture-backed serializer tests. | -| 5 | Phase 6 placeholder burn-down | Remaining account/contact/conversation/message/inbox placeholder handlers are either real Chatwoot-compatible flows or explicitly tracked as unsupported reference gaps. | `rg` placeholder audit, route smoke, and endpoint-family tests. | +| 1 | B12 | Reused Chatwoot frontend smoke runs repeatably against GoChat. | Checked smoke command plus gap report under `docs/parity/`. | +| 2 | Phase 5 job parity | Any async behavior deferred from B7-B11 has a durable worker decision, fakeable delivery boundary, and replay/idempotency tests. | Focused worker/job tests plus a Phase 5 gap ledger update. | +| 3 | Phase 2/3 audit pass | Route/controller/serializer drift found by B12 or new reference inspection is captured as named slices, not free-form TODOs. | Regenerated route parity artifacts and fixture-backed serializer tests. | +| 4 | Phase 6 placeholder burn-down | Remaining account/contact/conversation/message/inbox placeholder handlers are either real Chatwoot-compatible flows or explicitly tracked as unsupported reference gaps. | `rg` placeholder audit, route smoke, and endpoint-family tests. | ## Handoff Contract This checkpoint is intended to make the development plan complete enough to track without reading Hermes notes first. -- The only active implementation slice is B11.3e. Do not start B12 until Captain/Copilot stream and realtime disabled states are deterministic and tested. +- The next active implementation slice is B12. B11.3e has deterministic Captain stream disabled/error SSE behavior and Copilot push payload tests; provider-specific realtime delivery depth remains a Phase 5 integration concern if it needs durable dispatch infrastructure. - The Hermes search plan is fully represented by Phase 1/B6. Future search changes must be Meilisearch-first and must not reintroduce production DB fallback. - The Hermes automation/macro/CSAT plan is fully represented by B8/B9 and Phase 5. Durable delayed execution and channel-specific template delivery remain visible Phase 5 work, not hidden Done work. - Enterprise scope is fixed: SLA, Audit, CustomRole, AgentCapacity, Captain/Copilot, CSAT, InboxLimit, automation, macros, assignment policies, and related limits/workflows are in scope; SSO/SAML/LDAP/OIDC are out of scope. @@ -65,8 +64,6 @@ Open work after the current checkpoint: | Area | Next concrete action | Tracking location | Done boundary | | --- | --- | --- | --- | -| B11.3e Captain streams | Reuse non-stream task DTOs for stream routes, guard nil/non-streaming LLM provider, and emit stable SSE error/done events. | `B11.3e streaming/realtime landing contract` | Focused SSE tests cover no-provider, invalid operation, provider error, and success chunks. | -| B11.3e Copilot realtime | Add or verify push payload serializers that match Chatwoot `CopilotThread#push_event_data` and `CopilotMessage#push_event_data`. | `B11.3e streaming/realtime landing contract` | Event payload tests prove raw thread/message shapes and allowed message keys. | | B12 smoke | Create repeatable GoChat plus reused Chatwoot frontend boot and seed path. | `B12 reused frontend verification breakdown` | `docs/parity/frontend_smoke_report.md` exists and maps failures to slices. | | Phase 5 jobs | Decide durable worker mechanism and wire deferred SLA/automation/macro/export/template delivery jobs behind idempotent boundaries. | `Phase 5: Background Jobs And Integrations` | Worker tests prove enqueue, retry, idempotency, and fakeable external effects. | | Phase 2/3 drift | Expand tracked route/serializer fixtures when B12 exposes frontend-critical gaps. | `Phase 2`, `Phase 3`, `docs/parity/` | Route parity remains 0 missing for tracked frontend routes; serializers have reference fixtures. | @@ -80,7 +77,7 @@ Open work after the current checkpoint: | Phase 1 | Meilisearch search engine | Review | B6 payload parity, optional live gate, and DB-fallback hardening are implemented; an actual live Meilisearch run is optional and environment-dependent | | Phase 2 | Route and controller parity audit | Doing | Ruby/Bundler unavailable, so Chatwoot route extraction currently uses static `routes.rb` fallback | | Phase 3 | Data and serializer parity | Doing | JSON fixture coverage is partial and still endpoint-family based | -| Phase 4 | Enterprise feature completion | Doing | B7, B8, B9, and B10 are in Review; B11 Captain resources, preferences, playground, document sync gates, Copilot thread/message payloads, and Captain task payload persistence are landed, while streaming/realtime compatibility remains active | +| Phase 4 | Enterprise feature completion | Doing | B7, B8, B9, B10, and B11 are in Review; B12 reused frontend smoke is the next broad verification gate | | Phase 5 | Background jobs and integrations | Planned | durable worker choice and job parity are open | | Phase 6 | Core placeholder burn-down | Doing | account/contact/conversation/message/inbox placeholder groups remain broad | | Phase 7 | Verification harness | Planned | search live gate and reused-frontend smoke harness are not complete | @@ -103,6 +100,7 @@ This ledger records the committed parity checkpoints that future slices should b | Commit | Scope | Verification summary | Follow-up state | | --- | --- | --- | --- | +| `feat(captain): align streaming fallbacks` | Completed B11.3e streaming/realtime compatibility: Captain task stream routes now reuse non-stream request DTOs, resolve conversations by account-scoped display ID or legacy ID, guard missing LLM providers with deterministic SSE `error`/`done` events, validate rewrite operations, and keep stream errors frontend-readable. Copilot message nested thread payloads now use Chatwoot `push_event_data` shape and event payload helpers omit REST-only fields. | Focused CaptainTask/Captain/Copilot handler and service tests passed; Copilot/Captain repository tests passed; full verification recorded below. | B11 moves to Review; start B12 reused frontend smoke. | | `docs: land parity tracker handoff` | Froze the active tracker after B11.3d, copied the remaining Hermes-derived work into explicit B11.3e/B12/Phase 5/Phase 6 tracking rows, and clarified the commit/update rules for future checkpoints. | Documentation-only checkpoint; `git diff --check` is sufficient. | Start B11.3e streaming/realtime compatibility. | | `6aa62c6 docs: consolidate chatwoot parity roadmap` | Promoted Hermes-era plans into this master tracker; locked user decisions; added milestone, slice, enterprise, and webhook provider tracking. | Documentation-only checkpoint. | B1/P6.7 selected as next implementation slice. | | `9e3f561 feat(webhook): align chatwoot ingress routes` | Added Chatwoot public webhook paths for Twitter, Telegram, LINE, SMS/Twilio, WhatsApp, Instagram, TikTok, and Shopify; removed generic fallback success masking. | `go test ./...`; route dump regenerated with `TOTAL: 801`; route parity stayed `251 exact, 0 missing`. | Provider-specific lookup and verification moved to review. | @@ -189,7 +187,7 @@ Completed implementation slice: B11.3c now adds Captain document sync backend ga Completed documentation slice: the remaining development plan is now fully landed in this tracker. B11.3 is split into concrete Captain playground, document sync/indexing, Copilot task/tool-call, and streaming/realtime checkpoints. B12 has a repeatable smoke-report contract. B9.3 delayed/durable worker scheduling and B9.4 macro attachment depth remain named Phase 5 follow-ups. -Next implementation slice: start B11.3e streaming/realtime compatibility. B11.3d now covers rewrite/summarize/reply suggestion/label suggestion/follow-up request payloads, raw disabled states, persisted suggestion state, and reloadable Copilot message key validation. +Next implementation slice: start B12 reused frontend smoke. B11.3d covers rewrite/summarize/reply suggestion/label suggestion/follow-up request payloads, raw disabled states, persisted suggestion state, and reloadable Copilot message key validation; B11.3e now adds deterministic stream disabled states and Copilot push payload parity. | Step | Required result | Reference source | Verification | | --- | --- | --- | --- | @@ -217,7 +215,7 @@ Next implementation slice: start B11.3e streaming/realtime compatibility. B11.3d | N22 | Keep B11.3b Captain assistant playground payload parity as current playground baseline. | `reference/chatwoot/enterprise/app/controllers/api/v1/accounts/captain/assistants_controller.rb#playground`, dashboard `api/captain/assistant.js`, assistant playground component/store. | Done by `feat(captain): align playground fallback`; top-level `message_content` and `message_history` are accepted, account-scoped assistant lookup is enforced, legacy `{ content }` and v2 `{ response }` shapes are covered, and no-LLM fallback returns raw frontend-compatible JSON without 500. | | N23 | Keep B11.3c Captain document sync/indexing gates as current document-sync baseline. | Captain document sync service/jobs, document controller `sync`, existing Meilisearch engine, and any local embedding boundary. | Done by `feat(captain): gate document sync backend`; disabled config, fake successful indexing, failed sync metadata, fingerprint normalization, and account-scoped document lookup are covered without opening external network connections in default tests. | | N24 | Keep B11.3d Copilot task/tool-call persistence as current task baseline. | `resource :tasks` routes, Copilot/Captain task services, dashboard Copilot clients, current local `copilot_*` models. | Done by `feat(captain): align task payload persistence`; task routes accept frontend bodies, persist suggestions, return raw payloads/disabled states, and validate reloadable Copilot message keys. | -| N25 | Implement B11.3e streaming/realtime compatibility. | Chatwoot Copilot/Captain streaming, push/event payloads, current Go channel dispatcher. | If streaming is not implemented, the endpoint returns deterministic non-streaming payloads and records the fallback; no route returns placeholder success or local envelope. | +| N25 | Keep B11.3e streaming/realtime compatibility as current Captain/Copilot disabled-state baseline. | Chatwoot Copilot/Captain streaming, push/event payloads, current Go channel dispatcher. | Done by `feat(captain): align streaming fallbacks`; Captain stream routes have deterministic SSE error/done states and Copilot push payload helpers match reference shapes. | | N26 | Start B12 smoke harness after B11.3 has a tested disabled/external-provider story. | `reference/chatwoot` frontend boot scripts and GoChat dev/test boot flow. | A checked command and `docs/parity/frontend_smoke_report.md` record core and enterprise smoke status. | | N27 | Keep Hermes source plans mapped but inactive. | `.hermes/plans/2025-05-24-global-search-meilisearch.md`, `.hermes/plans/2026-05-24-automation-macro-csat.md`. | New work must update this tracker directly; Hermes files are read-only source notes unless the user asks otherwise. | | N28 | Update this tracker after every implementation checkpoint. | This document. | `git diff --check`; `go test ./...` for Go changes. | @@ -816,7 +814,7 @@ B11 Captain/Copilot breakdown: | B11.3b | Align Captain assistant playground request/response behavior and no-LLM fallback. | `assistants_controller.rb#playground`, `Captain::Llm::AssistantChatService`, `AgentRunnerService`, dashboard `api/captain/assistant.js`, playground component/store. | Handler/service tests cover top-level `message_content`, `message_history`, account-scoped assistant lookup, no-history payloads, no duplicate latest user message for v2 behavior if enabled, raw `{ content }` or `{ response }` response shape, and safe no-LLM fallback. | Done by `feat(captain): align playground fallback` | | B11.3c | Implement document sync/indexing gates for Meilisearch or the chosen embedding/search backend. | Captain document sync/indexing services, document controller `sync`, existing Meilisearch engine, local embedding/config boundaries. | Tests cover disabled config, failed sync observability, successful fake backend indexing, account-scoped document lookup, and no external network call in default test mode. | Done by `feat(captain): gate document sync backend` | | B11.3d | Finish Copilot task, suggestion, and tool-call persistence depth. | `resource :tasks` routes in `reference/chatwoot/config/routes.rb:62-89`, Copilot/Captain task services, dashboard Copilot clients. | Tests cover rewrite/summarize/reply suggestion/label suggestion/follow-up request payloads, persisted task/tool-call records, reloadable serializers, account/user scoping, and raw disabled state when provider config is missing. | Done by `feat(captain): align task payload persistence` | -| B11.3e | Add streaming/realtime compatibility or deterministic non-streaming fallback. | Chatwoot Captain/Copilot streaming paths, push payloads, current Go dispatcher/realtime boundaries. | Tests prove endpoints never return placeholder success/local envelopes; non-streaming fallback payloads are deterministic and frontend-compatible until true streaming is implemented. | Todo | +| B11.3e | Add streaming/realtime compatibility or deterministic non-streaming fallback. | Chatwoot Captain/Copilot streaming paths, push payloads, current Go dispatcher/realtime boundaries. | Tests prove endpoints never return placeholder success/local envelopes; non-streaming fallback payloads are deterministic and frontend-compatible until true streaming is implemented. | Done by `feat(captain): align streaming fallbacks` | B11.1a current checkpoint: @@ -953,6 +951,28 @@ B11.3e streaming/realtime landing contract: | Feature gates | Missing provider credentials, disabled Captain task features, or unimplemented streaming should be visible as raw disabled/error states. | No route returns a local envelope or opaque `500` for a known disabled state. | | Route stability | Existing Chatwoot-compatible paths stay mounted; legacy local paths may remain only if they do not change frontend route matching. | Route dump remains stable unless a tracked route is intentionally added and parity artifacts are regenerated. | +B11.3e current checkpoint: + +- Captain `reply_suggestion/stream`, `summarize/stream`, and `rewrite/stream` now use the same frontend DTOs as their non-streaming task routes, including `conversation_display_id`, `operation`, and legacy ID fallback only inside the service adapter. +- Stream services now guard missing LLM providers with `422 Captain is disabled` task errors instead of nil-provider panics. +- Stream handlers convert known task errors into deterministic SSE `error` events containing `error`, `status`, and `done`, followed by a `done` event. Invalid request bodies and account IDs use the same frontend-readable SSE shape. +- Stream conversation lookup is account-scoped through `FindByAccountAndDisplayIDOrID`; reply suggestion assistant lookup is also account-scoped and optional like the non-stream route. +- Rewrite stream validates the Chatwoot operation list and includes conversation context for `improve` when `conversation_display_id` is provided. +- Copilot message serializers now embed the nested `copilot_thread` using Chatwoot `CopilotThread#push_event_data` shape without REST-only `assistant`, and dedicated push payload helpers match `CopilotMessage#push_event_data` by omitting top-level `account_id`. +- The legacy Copilot SSE route now returns deterministic `Captain is disabled` SSE output when no LLM provider is configured after thread/account validation. + +B11.3e verification: + +```bash +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'CaptainTask|Captain|Copilot' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/handler/api/v1 -run 'CaptainTask|Captain|Copilot|SSE' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/repository -run 'Copilot|Captain' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/handler/api/v1 -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -count=1 +env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./... +git diff --check +``` + B12 smoke harness landing contract: | Area | Landing requirement | Done signal | @@ -1090,7 +1110,7 @@ Work proceeds top-down unless a failing test or frontend blocker forces a narrow | B8 | CSAT account/public/send completion. | Chatwoot CSAT survey responses, reports, downloads, listeners. | Metrics/list/review/download/send idempotency tests. | Review | | B9 | Automation/macros durable side effects. | Chatwoot automation/macro services and jobs. | Action execution, logs, webhook/email transcript retry tests. | Review | | B10 | Audit, CustomRole, InboxLimit. | Chatwoot enterprise admin behavior and policies. | Authorization, audit emission, limits enforcement, admin payload fixtures. | Review | -| B11 | Captain/Copilot deep behavior. | Chatwoot Captain/Copilot controllers, services, frontend clients. | Assistant/tool/document/scenario/copilot thread/task tests and feature gates. | Doing | +| B11 | Captain/Copilot deep behavior. | Chatwoot Captain/Copilot controllers, services, frontend clients. | Assistant/tool/document/scenario/copilot thread/task tests and feature gates. | Review | | B12 | Frontend smoke harness. | `reference/chatwoot` frontend. | Repeatable smoke command and checked gap report. | Todo | Remaining slice landing plan: @@ -1102,7 +1122,7 @@ Remaining slice landing plan: | B8 | Done: account-side CSAT list/metrics/review-note payload parity, public submit/update idempotency, resolve-triggered generic survey message send, and CSV download filters/columns. | Channel-specific WhatsApp/Twilio template send hooks and durable queueing are tracked under Phase 5. | Review after `b36cf07`; move to Done only after channel-specific send hooks are either implemented or formally split as external integration scope. | | B9 | Done: B9.1 automation rule CRUD/listener/log/external-action parity and B9.2 macro frontend CRUD/execute side effects. | Delayed actions, durable queued worker scheduling, and deeper macro attachment/file parity remain named B9.3/B9.4 follow-ups. | Review after `feat(macros): align chatwoot macro payloads`; move to Done only after durable worker/attachment gaps are implemented or formally split out. | | B10 | Done: audit list payload, audit writer boundary for representative mutating core resources, CustomRole permission-key parity, AccountUser permission resolution, admin gates, delete nullification, and account-level InboxLimit enforcement in inbox/channel creation paths. | Frontend smoke coverage for enterprise settings remains B12, not a hidden B10 blocker. | Review after B10.4; move to Done only after reused frontend smoke confirms audit/custom-role/limit settings flows or any smoke gaps are split into owned follow-ups. | -| B11 | Done through B11.3d: Captain Assistant CRUD, inbox binding, scenarios, documents, responses, custom tools, Copilot threads/messages, Captain preferences, playground, document sync gates, and task/tool-call payload persistence fixtures. | B11.3e streaming/realtime fallback remains active. | Done only after LLM-dependent behavior is either implemented behind config or safely stubbed with frontend-compatible disabled states. | +| B11 | Done through B11.3e: Captain Assistant CRUD, inbox binding, scenarios, documents, responses, custom tools, Copilot threads/messages, Captain preferences, playground, document sync gates, task/tool-call payload persistence, stream disabled-state SSE, and Copilot push payload fixtures. | Reused frontend smoke remains B12; provider-specific durable realtime/LLM delivery can move to Phase 5 if smoke exposes it. | Review after B11.3e; move to Done only after B12 proves reused frontend Captain/Copilot screens or names any remaining provider/deployment follow-ups. | | B12 | Boot reused Chatwoot frontend against GoChat auth/profile/inbox/conversation/contact flows. | Add smoke paths for widget init/message, public CSAT, reports, and enterprise screens as B7-B11 land. | Done only after the smoke command is repeatable and writes a checked gap report. | Per-slice documentation rule: @@ -1361,7 +1381,7 @@ Included checklist: - [x] Custom roles and permission checks. - [x] Agent capacity and assignment limits. - [ ] Assignment policies and auto-assignment compatibility. -- [ ] Captain/Copilot assistant, custom tools, scenarios, documents, responses, inbox bindings, preferences, and Copilot threads/messages. Captain assistant CRUD/tools/inbox binding, scenario/custom-tool CRUD, documents, assistant responses, bulk actions, custom-tool test payloads, preferences, and Copilot thread/message payloads are landed; external document/LLM gates and remaining Copilot task/tool-call depth remain active. +- [x] Captain/Copilot assistant, custom tools, scenarios, documents, responses, inbox bindings, preferences, Copilot threads/messages, tasks, and stream disabled-state compatibility. Provider-specific LLM depth remains feature-gated and B12/Phase 5 can name any deeper frontend or dispatch gaps. - [x] CSAT survey response flow, metrics, filters, and review notes. - [x] Inbox limits and account/inbox usage enforcement. - [x] Automation rules, macros, execution logs, and action side effects. @@ -1380,7 +1400,7 @@ Enterprise tracking table: | P4.3 | Custom roles/permissions | `internal/model/custom_role.go`, `internal/service/custom_role_service.go`, `internal/middleware/role_check.go`, `internal/handler/api/v1/custom_role_handler.go` | Permission arrays, raw admin payloads, administrator gates, account-user resolution, profile permissions, and delete nullification are implemented; keep in Review until frontend smoke covers enterprise settings. | Review | | P4.4 | Agent capacity | `internal/model/agent_capacity_policy.go`, `internal/service/agent_capacity_policy_service.go`, `internal/handler/api/v1/agent_capacity_handler.go`, `internal/autoassignment/*` | First align AgentCapacityPolicy API, serializers, nested users, and `InboxCapacityLimit` data contracts; then enforce capacity in manual/auto assignment. | Review | | P4.5 | Inbox limits | `internal/model/agent_capacity_policy.go` `InboxCapacityLimit`, plus legacy `internal/model/inbox_limit.go`, `internal/service/inbox_limit_service.go`, `internal/repository/inbox_limit_repo.go`, `internal/handler/api/v1/inbox_limit_handler.go` | Chatwoot enterprise nested `agent_capacity_policies/:policy_id/inbox_limits`, assignment-path enforcement, and account-level `usage_limits[:inboxes]` create-path enforcement are covered. | Review | -| P4.6 | Captain/Copilot | `internal/model/captain_models.go`, `internal/model/copilot_models.go`, `internal/service/captain_*`, `internal/service/copilot_*`, `internal/handler/api/v1/captain_*`, `internal/handler/api/v1/copilot_handler.go` | Captain resource payloads through B11.1c, Copilot thread/message payloads through B11.2, Captain preferences through B11.3a, Captain playground through B11.3b, document sync gates through B11.3c, and task/tool-call payload persistence through B11.3d are aligned; complete B11.3e streaming compatibility. | Doing | +| P4.6 | Captain/Copilot | `internal/model/captain_models.go`, `internal/model/copilot_models.go`, `internal/service/captain_*`, `internal/service/copilot_*`, `internal/handler/api/v1/captain_*`, `internal/handler/api/v1/copilot_handler.go` | Captain resource payloads through B11.1c, Copilot thread/message payloads through B11.2, Captain preferences through B11.3a, Captain playground through B11.3b, document sync gates through B11.3c, task/tool-call payload persistence through B11.3d, and stream/realtime fallback payloads through B11.3e are aligned. | Review | | P4.7 | CSAT | `internal/csat/*`, `internal/automation/csat_survey_*`, `internal/handler/api/v1/csat_*`, `internal/service/csat_metrics_service.go` | Account-side list/metrics/review-note payloads, public update depth, resolve-triggered generic survey send, and download CSV are aligned; channel-specific template send hooks remain Phase 5 integration work. | Review | | P4.8 | Automation and macros | `internal/automation/*`, `internal/handler/api/v1/automation_rule_handler.go`, `internal/handler/api/v1/macro_handler.go` | Automation-rule CRUD/listener/log/external-action delivery and macro CRUD/display-ID execution parity are implemented; durable delayed actions and deeper macro attachment/file parity remain B9.3/B9.4 follow-ups. | Review | | P4.9 | Assignment policies | `internal/autoassignment/*`, `internal/automation/agent_bot_rule_listener.go` | Match Chatwoot assignment policy behavior and availability/capacity rules. | Doing | @@ -1397,7 +1417,7 @@ Enterprise work package breakdown: | Audit | Audit model parity, mutating action coverage, request metadata, filters/pagination, admin endpoint payloads. | List/admin payload parity is covered by `feat(audit): align chatwoot audit log payloads`; named representative writer coverage is covered by `feat(audit): record enterprise mutations` and `feat(audit): cover operational mutations`. | Review | | Custom roles | Permission-key parity, account-user role resolution, policy middleware, create/update/delete behavior. | Permission arrays, raw admin payloads, admin gates, RBAC/account-scope/profile resolution, and delete nullification are covered by `feat(custom-roles): align chatwoot permissions`. | Review | | Inbox limits | Account/inbox limit models, create/update enforcement, UI-readable limit responses, admin overrides. | Chatwoot nested capacity-limit API, assignment-path enforcement, and account-level create-limit `402` behavior are covered by B5/B7/B10.4. | Review | -| Captain/Copilot | Assistants, inbox bindings, scenarios, responses, documents, tools, preferences, copilot threads/messages, tasks, streaming/tool-call behavior. | Assistant CRUD/tools/inbox binding fixtures are covered by `feat(captain): align assistant resources`; scenario/custom-tool fixtures are covered by `feat(captain): align scenario and tool payloads`; document/assistant-response/bulk/custom-tool-test fixtures are covered by `feat(captain): align document response actions`; Copilot thread/message fixtures are covered by `feat(copilot): align thread message payloads`; preferences fixtures are covered by `feat(captain): align preferences payloads`; playground fixtures are covered by `feat(captain): align playground fallback`; document sync gate fixtures are covered by `feat(captain): gate document sync backend`; task/tool-call payload persistence is covered by `feat(captain): align task payload persistence`; B11.3e now tracks remaining streaming fallback and frontend smoke screens. | Doing | +| Captain/Copilot | Assistants, inbox bindings, scenarios, responses, documents, tools, preferences, copilot threads/messages, tasks, streaming/tool-call behavior. | Assistant CRUD/tools/inbox binding fixtures are covered by `feat(captain): align assistant resources`; scenario/custom-tool fixtures are covered by `feat(captain): align scenario and tool payloads`; document/assistant-response/bulk/custom-tool-test fixtures are covered by `feat(captain): align document response actions`; Copilot thread/message fixtures are covered by `feat(copilot): align thread message payloads`; preferences fixtures are covered by `feat(captain): align preferences payloads`; playground fixtures are covered by `feat(captain): align playground fallback`; document sync gate fixtures are covered by `feat(captain): gate document sync backend`; task/tool-call payload persistence is covered by `feat(captain): align task payload persistence`; stream disabled states and push payload shapes are covered by `feat(captain): align streaming fallbacks`. | Review | Enterprise acceptance gates: @@ -1642,3 +1662,4 @@ Verification milestone gates: - 2026-06-05: B11.3d/B11.3e/B12 planning checkpoint prepared as `docs: land copilot task execution plan`; the active tracker now records exact Chatwoot Captain task request/response contracts, current Go handler/service gaps, follow-up context and allowed operation rules, reloadable Copilot message/tool-call persistence requirements, no-provider disabled-state expectations, streaming/realtime fallback gates, B12 smoke harness deliverables, and focused exit commands. Documentation-only checkpoint; `git diff --check` passed. Next slice is B11.3d implementation. - 2026-06-05: B11.3d Captain task/tool-call checkpoint prepared as `feat(captain): align task payload persistence`; rewrite/summarize/reply suggestion now accept Chatwoot task payloads and return raw `{ message, follow_up_context }` or `422 { error }`, label suggestion and follow-up POST routes consume dashboard `tasks.js` bodies, no-provider paths return `Captain is disabled`, task outputs persist to `copilot_suggestion_messages`, and `CopilotMessage` validates reloadable tool-call keys. Focused CaptainTask/Captain/Copilot handler and service tests, Copilot/Captain repository tests, handler/service package tests, full `go test ./...` with workspace `TMPDIR`, and `git diff --check` passed. Next slice is B11.3e streaming/realtime compatibility. - 2026-06-05: Parity tracker handoff checkpoint prepared as `docs: land parity tracker handoff`; the plan now has a front-loaded handoff contract, explicit open-work table for B11.3e/B12/Phase 5/Phase 2/3/Phase 6, exact Hermes source-plan mapping, and a rule that every future checkpoint updates this tracker before commit. Documentation-only checkpoint; `git diff --check` passed. +- 2026-06-05: B11.3e streaming/realtime checkpoint prepared as `feat(captain): align streaming fallbacks`; Captain stream task routes now share non-stream DTO/account-scope behavior, return deterministic SSE disabled/error states for missing providers and validation errors, and preserve success chunk/done shapes. Copilot message REST payloads now nest thread `push_event_data`, dedicated push payload helpers match Chatwoot event data, and the legacy Copilot SSE route has a no-provider disabled guard. Focused CaptainTask/Captain/Copilot handler/service tests and Copilot/Captain repository tests passed; full verification is recorded in the B11.3e section. B11 moves to Review; next active slice is B12 reused Chatwoot frontend smoke. diff --git a/internal/handler/api/v1/captain_task_handler.go b/internal/handler/api/v1/captain_task_handler.go index 8db8e65e..9fe97e34 100644 --- a/internal/handler/api/v1/captain_task_handler.go +++ b/internal/handler/api/v1/captain_task_handler.go @@ -111,13 +111,13 @@ func (h *CaptainTaskHandler) Rewrite(c *gin.Context) { func (h *CaptainTaskHandler) StreamReplySuggestion(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { - captainWriteSSEError(c, "invalid account_id") + captainWriteSSEError(c, http.StatusBadRequest, "invalid account_id") return } var req service.TaskReplySuggestionRequest if err := c.ShouldBindJSON(&req); err != nil { - captainWriteSSEError(c, "invalid request body: "+err.Error()) + captainWriteSSEError(c, http.StatusBadRequest, "invalid request body: "+err.Error()) return } @@ -144,8 +144,7 @@ func (h *CaptainTaskHandler) StreamReplySuggestion(c *gin.Context) { if err != nil && err != io.EOF { applogger.L().Errorf("StreamReplySuggestion: %v", err) - captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s"}`, captainEscapeJSONString(err.Error()))) - captainWriteSSEMessage(c, "done", `{"done": true}`) + captainWriteSSETaskError(c, err) } } @@ -154,13 +153,13 @@ func (h *CaptainTaskHandler) StreamReplySuggestion(c *gin.Context) { func (h *CaptainTaskHandler) StreamSummarize(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { - captainWriteSSEError(c, "invalid account_id") + captainWriteSSEError(c, http.StatusBadRequest, "invalid account_id") return } var req service.TaskSummarizeRequest if err := c.ShouldBindJSON(&req); err != nil { - captainWriteSSEError(c, "invalid request body: "+err.Error()) + captainWriteSSEError(c, http.StatusBadRequest, "invalid request body: "+err.Error()) return } @@ -187,8 +186,7 @@ func (h *CaptainTaskHandler) StreamSummarize(c *gin.Context) { if err != nil && err != io.EOF { applogger.L().Errorf("StreamSummarize: %v", err) - captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s"}`, captainEscapeJSONString(err.Error()))) - captainWriteSSEMessage(c, "done", `{"done": true}`) + captainWriteSSETaskError(c, err) } } @@ -197,13 +195,13 @@ func (h *CaptainTaskHandler) StreamSummarize(c *gin.Context) { func (h *CaptainTaskHandler) StreamRewrite(c *gin.Context) { accountID, err := strconv.ParseUint(c.Param("id"), 10, 64) if err != nil { - captainWriteSSEError(c, "invalid account_id") + captainWriteSSEError(c, http.StatusBadRequest, "invalid account_id") return } var req service.TaskRewriteRequest if err := c.ShouldBindJSON(&req); err != nil { - captainWriteSSEError(c, "invalid request body: "+err.Error()) + captainWriteSSEError(c, http.StatusBadRequest, "invalid request body: "+err.Error()) return } @@ -230,8 +228,7 @@ func (h *CaptainTaskHandler) StreamRewrite(c *gin.Context) { if err != nil && err != io.EOF { applogger.L().Errorf("StreamRewrite: %v", err) - captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s"}`, captainEscapeJSONString(err.Error()))) - captainWriteSSEMessage(c, "done", `{"done": true}`) + captainWriteSSETaskError(c, err) } } @@ -253,12 +250,21 @@ func captainWriteSSEMessage(c *gin.Context, event string, data string) { } // captainWriteSSEError writes an error SSE event followed by a done event. -func captainWriteSSEError(c *gin.Context, errMsg string) { +func captainWriteSSEError(c *gin.Context, status int, errMsg string) { captainSetSSEHeaders(c) - captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s"}`, captainEscapeJSONString(errMsg))) + captainWriteSSEMessage(c, "error", fmt.Sprintf(`{"error": "%s", "status": %d, "done": true}`, captainEscapeJSONString(errMsg), status)) captainWriteSSEMessage(c, "done", `{"done": true}`) } +func captainWriteSSETaskError(c *gin.Context, err error) { + status, message, ok := service.CaptainTaskErrorStatus(err) + if !ok { + status = http.StatusUnprocessableEntity + message = err.Error() + } + captainWriteSSEError(c, status, message) +} + // captainEscapeJSONString escapes special characters for safe JSON embedding. func captainEscapeJSONString(s string) string { var result strings.Builder diff --git a/internal/handler/api/v1/captain_task_handler_test.go b/internal/handler/api/v1/captain_task_handler_test.go index b6c6a4ca..f6c46bd5 100644 --- a/internal/handler/api/v1/captain_task_handler_test.go +++ b/internal/handler/api/v1/captain_task_handler_test.go @@ -7,6 +7,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "strings" "testing" "github.com/gin-gonic/gin" @@ -21,8 +22,10 @@ import ( ) type mockCaptainTaskHandlerLLM struct { - response *llm.ChatResponse - err error + response *llm.ChatResponse + err error + streamChunks []llm.StreamChunk + streamErr error } func (m *mockCaptainTaskHandlerLLM) ChatCompletion(_ context.Context, _ llm.ChatRequest) (*llm.ChatResponse, error) { @@ -31,7 +34,16 @@ func (m *mockCaptainTaskHandlerLLM) ChatCompletion(_ context.Context, _ llm.Chat func (m *mockCaptainTaskHandlerLLM) CreateEmbedding(_ context.Context, _ llm.EmbeddingRequest) (*llm.EmbeddingResponse, error) { return nil, nil } -func (m *mockCaptainTaskHandlerLLM) ChatCompletionStream(_ context.Context, _ llm.ChatRequest, _ func(llm.StreamChunk) error) error { + +func (m *mockCaptainTaskHandlerLLM) ChatCompletionStream(_ context.Context, _ llm.ChatRequest, onChunk func(llm.StreamChunk) error) error { + if m.streamErr != nil { + return m.streamErr + } + for _, chunk := range m.streamChunks { + if err := onChunk(chunk); err != nil { + return err + } + } return nil } @@ -101,3 +113,67 @@ func TestCaptainTaskHandler_Rewrite_NoProviderRawDisabled(t *testing.T) { assert.Equal(t, "Captain is disabled", resp["error"]) assert.NotContains(t, resp, "success") } + +func TestCaptainTaskHandler_StreamRewrite_NoProviderDisabledSSE(t *testing.T) { + handler, _ := setupCaptainTaskHandlerTest(t, nil) + + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Params = gin.Params{{Key: "id", Value: "1"}} + c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/accounts/1/captain/tasks/rewrite/stream", bytes.NewReader([]byte(`{"content":"hello","operation":"professional"}`))) + c.Request.Header.Set("Content-Type", "application/json") + + handler.StreamRewrite(c) + + body := w.Body.String() + assert.Equal(t, "text/event-stream", w.Header().Get("Content-Type")) + assert.Contains(t, body, "event: error") + assert.Contains(t, body, `"error": "Captain is disabled"`) + assert.Contains(t, body, `"status": 422`) + assert.Contains(t, body, "event: done") +} + +func TestCaptainTaskHandler_StreamSummarize_ChatwootDisplayID(t *testing.T) { + provider := &mockCaptainTaskHandlerLLM{streamChunks: []llm.StreamChunk{ + {Choices: []llm.StreamChoice{{Delta: llm.StreamDelta{Content: "Short"}}}}, + {Choices: []llm.StreamChoice{{Delta: llm.StreamDelta{Content: " summary"}, FinishReason: "stop"}}}, + }} + handler, db := setupCaptainTaskHandlerTest(t, provider) + displayID := uint(456) + conv := &model.Conversation{AccountID: 1, DisplayID: &displayID, Status: "open", ChannelType: "web_widget", Channel: "web_widget"} + require.NoError(t, db.Create(conv).Error) + require.NoError(t, db.Create(&model.Message{ConversationID: conv.ID, AccountID: 1, SenderType: "contact", MessageType: "incoming", Content: "Need help"}).Error) + + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Params = gin.Params{{Key: "id", Value: "1"}} + c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/accounts/1/captain/tasks/summarize/stream", bytes.NewReader([]byte(`{"conversation_display_id":456}`))) + c.Request.Header.Set("Content-Type", "application/json") + + handler.StreamSummarize(c) + + body := w.Body.String() + assert.Contains(t, body, "event: message") + assert.Contains(t, body, `"content": "Short"`) + assert.Contains(t, body, `"content": " summary"`) + assert.Contains(t, body, "event: done") + assert.NotContains(t, body, "success") +} + +func TestCaptainTaskHandler_StreamRewrite_InvalidOperationSSE(t *testing.T) { + handler, _ := setupCaptainTaskHandlerTest(t, &mockCaptainTaskHandlerLLM{}) + + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Params = gin.Params{{Key: "id", Value: "1"}} + c.Request = httptest.NewRequest(http.MethodPost, "/api/v1/accounts/1/captain/tasks/rewrite/stream", strings.NewReader(`{"content":"hello","operation":"pirate"}`)) + c.Request.Header.Set("Content-Type", "application/json") + + handler.StreamRewrite(c) + + body := w.Body.String() + assert.Contains(t, body, "event: error") + assert.Contains(t, body, `"error": "Invalid operation: pirate"`) + assert.Contains(t, body, `"status": 422`) + assert.Contains(t, body, "event: done") +} diff --git a/internal/handler/api/v1/copilot_handler.go b/internal/handler/api/v1/copilot_handler.go index d8a01626..41623cd1 100644 --- a/internal/handler/api/v1/copilot_handler.go +++ b/internal/handler/api/v1/copilot_handler.go @@ -349,17 +349,37 @@ func copilotThreadPayload(thread *model.CopilotThread) gin.H { } } +func copilotThreadPushPayload(thread *model.CopilotThread) gin.H { + return gin.H{ + "id": thread.ID, + "title": thread.Title, + "created_at": thread.CreatedAt.Unix(), + "user": copilotUserPayload(&thread.User, thread.UserID, thread.AccountID), + "account_id": thread.AccountID, + } +} + func copilotMessagePayload(message *model.CopilotMessage) gin.H { return gin.H{ "id": message.ID, "message": rawJSONValue(message.Message), "message_type": message.MessageType, "created_at": message.CreatedAt.Unix(), - "copilot_thread": copilotThreadPayload(&message.CopilotThread), + "copilot_thread": copilotThreadPushPayload(&message.CopilotThread), "account_id": message.AccountID, } } +func copilotMessagePushPayload(message *model.CopilotMessage) gin.H { + return gin.H{ + "id": message.ID, + "message": rawJSONValue(message.Message), + "message_type": message.MessageType, + "created_at": message.CreatedAt.Unix(), + "copilot_thread": copilotThreadPushPayload(&message.CopilotThread), + } +} + func copilotUserPayload(user *model.User, fallbackID, accountID uint) gin.H { if user == nil || user.ID == 0 { return gin.H{"id": fallbackID, "account_id": accountID, "type": "user"} diff --git a/internal/handler/api/v1/copilot_thread_handler_test.go b/internal/handler/api/v1/copilot_thread_handler_test.go index fa49856d..9324c8f9 100644 --- a/internal/handler/api/v1/copilot_thread_handler_test.go +++ b/internal/handler/api/v1/copilot_thread_handler_test.go @@ -237,6 +237,48 @@ func TestCopilotThreadMessagesListAndCreateUseNestedPayloads(t *testing.T) { require.Len(t, messages, 4) } +func TestCopilotMessagePayloadUsesThreadPushShape(t *testing.T) { + f := newCopilotParityFixture(t) + thread := f.createThread(t, "Need help") + threadID := uintString(uint(thread["id"].(float64))) + path := f.captainPath("/copilot_threads/" + threadID + "/copilot_messages/") + + w := f.request(f.router, http.MethodGet, path, nil) + require.Equal(t, http.StatusOK, w.Code, w.Body.String()) + messages := decodeMap(t, w)["payload"].([]any) + message := messages[0].(map[string]any) + nestedThread := message["copilot_thread"].(map[string]any) + + require.Equal(t, thread["id"], nestedThread["id"]) + require.NotNil(t, nestedThread["user"]) + require.Equal(t, thread["account_id"], nestedThread["account_id"]) + require.Nil(t, nestedThread["assistant"]) + require.NotNil(t, message["account_id"]) +} + +func TestCopilotMessagePushPayloadMatchesChatwootEventData(t *testing.T) { + f := newCopilotParityFixture(t) + thread := f.createThread(t, "Need help") + threadID := uintString(uint(thread["id"].(float64))) + path := f.captainPath("/copilot_threads/" + threadID + "/copilot_messages/") + w := f.request(f.router, http.MethodPost, path, map[string]any{"message": "Follow up", "conversation_id": 123}) + require.Equal(t, http.StatusOK, w.Code, w.Body.String()) + created := decodeMap(t, w) + + var stored model.CopilotMessage + require.NoError(t, f.db.Preload("CopilotThread.User").Preload("CopilotThread.Assistant").First(&stored, uint(created["id"].(float64))).Error) + push := copilotMessagePushPayload(&stored) + nestedThread := push["copilot_thread"].(gin.H) + + require.Equal(t, created["id"], float64(push["id"].(uint))) + require.Equal(t, "user", string(push["message_type"].(model.CopilotMessageType))) + require.Equal(t, "Follow up", push["message"].(map[string]any)["content"]) + require.Nil(t, push["account_id"]) + require.Equal(t, stored.CopilotThreadID, nestedThread["id"]) + require.Nil(t, nestedThread["assistant"]) + require.NotNil(t, nestedThread["user"]) +} + func TestCopilotThreadMessagesAreAccountAndUserScoped(t *testing.T) { f := newCopilotParityFixture(t) thread := f.createThread(t, "Private thread") diff --git a/internal/handler/api/v1/sse_stream_handler.go b/internal/handler/api/v1/sse_stream_handler.go index b9dee952..d76d75b9 100644 --- a/internal/handler/api/v1/sse_stream_handler.go +++ b/internal/handler/api/v1/sse_stream_handler.go @@ -87,6 +87,11 @@ func (h *SSEStreamHandler) StreamCopilotMessage(c *gin.Context) { writeSSEMessage(c, "done", `{"done": true}`) return } + if h.llmProvider == nil { + writeSSEMessage(c, "error", `{"error": "Captain is disabled", "status": 422, "done": true}`) + writeSSEMessage(c, "done", `{"done": true}`) + return + } // Build chat messages from thread history + new user message chatMessages := buildStreamChatMessages(thread, req.Content) diff --git a/internal/service/captain_task_service.go b/internal/service/captain_task_service.go index bdc404bd..ca350fa1 100644 --- a/internal/service/captain_task_service.go +++ b/internal/service/captain_task_service.go @@ -543,12 +543,28 @@ func parseSuggestions(content string) []string { // ReplySuggestionStream streams reply suggestions via an onChunk callback. // The callback receives StreamChunk events; the caller (handler) writes SSE events. func (s *CaptainTaskService) ReplySuggestionStream(ctx context.Context, accountID uint, req *TaskReplySuggestionRequest, onChunk func(llm.StreamChunk) error) error { - assistant, err := s.assistantRepo.GetByID(ctx, req.AssistantID) - if err != nil { - return fmt.Errorf("assistant not found: %w", err) + if s.llmProvider == nil { + return taskError(422, "Captain is disabled") } - messages, err := s.fetchConversationContext(ctx, req.ConversationID) + var assistant *model.CaptainAssistant + var err error + if req.AssistantID > 0 { + assistant, err = s.assistantRepo.GetByAccountAndID(ctx, accountID, req.AssistantID) + if err != nil { + return taskError(422, "assistant not found") + } + } else { + assistant = &model.CaptainAssistant{AccountID: accountID, Name: "Captain"} + assistant.Config = []byte(`{}`) + } + + conversation, err := s.resolveTaskConversation(ctx, accountID, req.ConversationDisplayID, req.ConversationID) + if err != nil { + return taskError(422, "Conversation not found") + } + + messages, err := s.fetchConversationContext(ctx, conversation.ID) if err != nil { applogger.L().Warnf("ReplySuggestionStream: conversation context fetch failed: %v, proceeding without context", err) messages = "" @@ -556,7 +572,7 @@ func (s *CaptainTaskService) ReplySuggestionStream(ctx context.Context, accountI // Enrich context with CopilotContextService (M12) if s.copilotContextSvc != nil && messages != "" { - ctxView, err := s.copilotContextSvc.GetCurrentViewingContext(ctx, accountID, req.ConversationID) + ctxView, err := s.copilotContextSvc.GetCurrentViewingContext(ctx, accountID, conversation.ID) if err != nil { applogger.L().Warnf("ReplySuggestionStream: context enrichment failed: %v", err) } else if ctxView != nil { @@ -574,11 +590,13 @@ func (s *CaptainTaskService) ReplySuggestionStream(ctx context.Context, accountI } ragContext := "" - docContext, err := s.searchDocumentation(ctx, req.AssistantID, messages) - if err != nil { - applogger.L().Warnf("ReplySuggestionStream: document search failed: %v, proceeding without RAG", err) - } else if docContext != "" { - ragContext = docContext + if req.AssistantID > 0 { + docContext, err := s.searchDocumentation(ctx, req.AssistantID, messages) + if err != nil { + applogger.L().Warnf("ReplySuggestionStream: document search failed: %v, proceeding without RAG", err) + } else if docContext != "" { + ragContext = docContext + } } cfg, _ := assistant.GetConfig() @@ -601,14 +619,26 @@ func (s *CaptainTaskService) ReplySuggestionStream(ctx context.Context, accountI // SummarizeStream streams a conversation summary via an onChunk callback. func (s *CaptainTaskService) SummarizeStream(ctx context.Context, accountID uint, req *TaskSummarizeRequest, onChunk func(llm.StreamChunk) error) error { - conversationContext, err := s.fetchConversationContext(ctx, req.ConversationID) + if s.llmProvider == nil { + return taskError(422, "Captain is disabled") + } + + conversation, err := s.resolveTaskConversation(ctx, accountID, req.ConversationDisplayID, req.ConversationID) + if err != nil { + return taskError(422, "Conversation not found") + } + + conversationContext, err := s.fetchConversationContext(ctx, conversation.ID) if err != nil { return fmt.Errorf("fetch conversation context: %w", err) } + if conversationContext == "" { + return taskError(422, fmt.Sprintf("no messages found for conversation %d", conversation.ID)) + } // Enrich context with CopilotContextService (M12) if s.copilotContextSvc != nil && conversationContext != "" { - ctxView, err := s.copilotContextSvc.GetCurrentViewingContext(ctx, accountID, req.ConversationID) + ctxView, err := s.copilotContextSvc.GetCurrentViewingContext(ctx, accountID, conversation.ID) if err != nil { applogger.L().Warnf("SummarizeStream: context enrichment failed: %v", err) } else if ctxView != nil { @@ -649,9 +679,19 @@ func (s *CaptainTaskService) SummarizeStream(ctx context.Context, accountID uint // RewriteStream streams a rewritten message via an onChunk callback. func (s *CaptainTaskService) RewriteStream(ctx context.Context, accountID uint, req *TaskRewriteRequest, onChunk func(llm.StreamChunk) error) error { - tone := req.Tone - if tone == "" { - tone = "professional" + if s.llmProvider == nil { + return taskError(422, "Captain is disabled") + } + + operation := req.Operation + if operation == "" { + operation = req.Tone + } + if operation == "" { + operation = "professional" + } + if !isAllowedRewriteOperation(operation) { + return taskError(422, "Invalid operation: "+operation) } language := req.Language @@ -660,8 +700,20 @@ func (s *CaptainTaskService) RewriteStream(ctx context.Context, accountID uint, } styleGuidelines := "" + if req.ConversationDisplayID > 0 { + conversation, err := s.resolveTaskConversation(ctx, accountID, req.ConversationDisplayID, 0) + if err != nil { + return taskError(422, "Conversation not found") + } + if operation == "improve" { + conversationContext, _ := s.fetchConversationContext(ctx, conversation.ID) + if conversationContext != "" { + styleGuidelines += "\nConversation context: " + conversationContext + } + } + } if req.AssistantID > 0 { - assistant, err := s.assistantRepo.GetByID(ctx, req.AssistantID) + assistant, err := s.assistantRepo.GetByAccountAndID(ctx, accountID, req.AssistantID) if err == nil { cfg, _ := assistant.GetConfig() if cfg.Instructions != "" { @@ -673,7 +725,7 @@ func (s *CaptainTaskService) RewriteStream(ctx context.Context, accountID uint, } } - systemPrompt := s.promptBuilder.BuildRewritePrompt(tone, language, styleGuidelines) + systemPrompt := s.promptBuilder.BuildRewritePrompt(operation, language, styleGuidelines) chatMessages := []llm.ChatMessage{ {Role: "system", Content: systemPrompt}, diff --git a/internal/service/captain_task_service_streaming_test.go b/internal/service/captain_task_service_streaming_test.go index f26d3a9b..e069c00d 100644 --- a/internal/service/captain_task_service_streaming_test.go +++ b/internal/service/captain_task_service_streaming_test.go @@ -111,28 +111,21 @@ func TestCaptainTaskService_SummarizeStream_成功(t *testing.T) { } func TestCaptainTaskService_SummarizeStream_无消息(t *testing.T) { - db, mockProvider, svc := setupCaptainTaskService(t) + db, _, svc := setupCaptainTaskService(t) account := createTestAccount(t, db) conv := createTestConversationWithMessages(t, db, account.ID, nil) - // When no messages, SummarizeStream still proceeds with empty context - // It won't return an error, but the LLM will receive minimal context - mockProvider.streamChunks = []llm.StreamChunk{ - {Choices: []llm.StreamChoice{{Delta: llm.StreamDelta{Content: "no content"}}}}, - } - - var collected []string err := svc.SummarizeStream(context.Background(), account.ID, &TaskSummarizeRequest{ ConversationID: conv.ID, }, func(chunk llm.StreamChunk) error { - if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" { - collected = append(collected, chunk.Choices[0].Delta.Content) - } return nil }) - require.NoError(t, err) - assert.NotEmpty(t, collected) + assert.Error(t, err) + status, message, ok := CaptainTaskErrorStatus(err) + require.True(t, ok) + assert.Equal(t, 422, status) + assert.Contains(t, message, "no messages found") } func TestCaptainTaskService_SummarizeStream_LLM流错误(t *testing.T) { @@ -211,6 +204,53 @@ func TestCaptainTaskService_RewriteStream_LLM流错误(t *testing.T) { assert.Error(t, err) } +func TestCaptainTaskService_Stream_NoProviderDisabled(t *testing.T) { + _, _, svc := setupCaptainTaskService(t) + svc.llmProvider = nil + + err := svc.RewriteStream(context.Background(), 1, &TaskRewriteRequest{Content: "test", Operation: "professional"}, func(chunk llm.StreamChunk) error { + return nil + }) + + status, message, ok := CaptainTaskErrorStatus(err) + require.True(t, ok) + assert.Equal(t, 422, status) + assert.Equal(t, "Captain is disabled", message) +} + +func TestCaptainTaskService_RewriteStream_InvalidOperation(t *testing.T) { + _, _, svc := setupCaptainTaskService(t) + + err := svc.RewriteStream(context.Background(), 1, &TaskRewriteRequest{Content: "test", Operation: "pirate"}, func(chunk llm.StreamChunk) error { + return nil + }) + + status, message, ok := CaptainTaskErrorStatus(err) + require.True(t, ok) + assert.Equal(t, 422, status) + assert.Equal(t, "Invalid operation: pirate", message) +} + +func TestCaptainTaskService_SummarizeStream_ChatwootDisplayID(t *testing.T) { + db, mockProvider, svc := setupCaptainTaskService(t) + account := createTestAccount(t, db) + conv := createTestConversationWithMessages(t, db, account.ID, []struct{ sender, content string }{{"contact", "hello"}}) + displayID := uint(321) + require.NoError(t, db.Model(conv).Update("display_id", displayID).Error) + mockProvider.streamChunks = []llm.StreamChunk{{Choices: []llm.StreamChoice{{Delta: llm.StreamDelta{Content: "summary"}, FinishReason: "stop"}}}} + + var collected []string + err := svc.SummarizeStream(context.Background(), account.ID, &TaskSummarizeRequest{ConversationDisplayID: displayID}, func(chunk llm.StreamChunk) error { + if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" { + collected = append(collected, chunk.Choices[0].Delta.Content) + } + return nil + }) + + require.NoError(t, err) + assert.Equal(t, []string{"summary"}, collected) +} + func TestCaptainTaskService_ReplySuggestionStream_回调中断(t *testing.T) { db, mockProvider, svc := setupCaptainTaskService(t) account := createTestAccount(t, db)