diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index 84441ed5..410a2a37 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md @@ -16,10 +16,10 @@ Build GoChat as a Go backend that can directly reuse the frontend from `referenc ## Current Baseline -- Current tracking checkpoint: 2026-06-05 after `89cdb2a feat(captain): queue document crawl jobs`, with this implementation checkpoint prepared as `feat(captain): queue response embedding jobs`. -- Latest implementation checkpoint: this checkpoint, prepared as `feat(captain): queue response embedding jobs`. +- Current tracking checkpoint: 2026-06-05 after `dd6d019 feat(captain): queue response embedding jobs`, with this implementation checkpoint prepared as `feat(captain): queue copilot response jobs`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(captain): queue copilot response jobs`. - Latest documentation-only checkpoint: `2923aae docs: land parity execution tracker`; this document is now the active follow-up plan and supersedes `.hermes/plans/*`. -- 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; B12.1 adds the reusable GoChat server/seed entrypoint plus a Meilisearch-first reused Chatwoot frontend smoke harness and report; B12.2a adds API smoke assertions for auth/profile, inbox, conversation/messages, contact/company, widget config/message, and public CSAT; B12.2b adds a zero-dependency Chrome DevTools browser smoke that loads the reused Chatwoot login and dashboard entrypoints through Vite and checks browser auth/dashboard API requests; B12.3a adds enterprise API smoke assertions for SLA reports/download, CSAT reports/download, automation/macros, audit/custom roles, capacity, Captain, and Copilot; B12.3b adds reused-frontend enterprise browser route navigation for SLA, CSAT, automation, macros, audit logs, custom roles, capacity, Captain, and Copilot request coverage; P5.1 adds the PostgreSQL-backed durable `background_jobs` model/migration plus WorkerPool enqueue, schedule, retry/backoff, dead-letter, idempotency, stale-lock recovery, and focused tests; P5.2 wires `channel.Dispatcher` and `dispatch.EventDispatcher` async paths into durable event jobs with worker replay tests; P5.3 queues Meilisearch write-side index/delete jobs for conversations, messages, contacts, companies, and articles while keeping search reads Meilisearch-first; P5.4 queues automation webhook and email transcript side effects as durable jobs while preserving fakeable delivery boundaries; P5.5 queues Chatwoot-style macro execute fan-out through durable `automation:macro_execution` jobs; P5.6 queues resolve-triggered CSAT survey sends and WhatsApp/Twilio CSAT template creation through durable jobs; P5.7 queues Chatwoot enterprise SLA account scans and applied-SLA evaluation jobs through the durable worker; P5.8 queues Chatwoot-style contact export artifact generation through durable `contact:export` jobs; P5.9 queues normalized provider inbound message persistence/dispatch through durable `webhook:incoming_message_persist` jobs; P5.10 queues Chatwoot `SendReplyJob`-style outbound message delivery through durable `message:send_reply` jobs and provider delivery-status/read-receipt updates through durable webhook status jobs; P5.11 queues Captain document sync, crawl/parser, schedule-sync, response-builder, and embedding-update work through durable jobs; P5.12 queues scheduled item fan-out, one-off campaigns, snoozed conversation reopening, account auto-resolution, widget/public message status updates, and account conversation bulk actions through durable jobs. Next active implementation slice is remaining P5.11 Copilot/conversation response jobs, followed by P5.13 analytics aggregation. +- 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; B12.1 adds the reusable GoChat server/seed entrypoint plus a Meilisearch-first reused Chatwoot frontend smoke harness and report; B12.2a adds API smoke assertions for auth/profile, inbox, conversation/messages, contact/company, widget config/message, and public CSAT; B12.2b adds a zero-dependency Chrome DevTools browser smoke that loads the reused Chatwoot login and dashboard entrypoints through Vite and checks browser auth/dashboard API requests; B12.3a adds enterprise API smoke assertions for SLA reports/download, CSAT reports/download, automation/macros, audit/custom roles, capacity, Captain, and Copilot; B12.3b adds reused-frontend enterprise browser route navigation for SLA, CSAT, automation, macros, audit logs, custom roles, capacity, Captain, and Copilot request coverage; P5.1 adds the PostgreSQL-backed durable `background_jobs` model/migration plus WorkerPool enqueue, schedule, retry/backoff, dead-letter, idempotency, stale-lock recovery, and focused tests; P5.2 wires `channel.Dispatcher` and `dispatch.EventDispatcher` async paths into durable event jobs with worker replay tests; P5.3 queues Meilisearch write-side index/delete jobs for conversations, messages, contacts, companies, and articles while keeping search reads Meilisearch-first; P5.4 queues automation webhook and email transcript side effects as durable jobs while preserving fakeable delivery boundaries; P5.5 queues Chatwoot-style macro execute fan-out through durable `automation:macro_execution` jobs; P5.6 queues resolve-triggered CSAT survey sends and WhatsApp/Twilio CSAT template creation through durable jobs; P5.7 queues Chatwoot enterprise SLA account scans and applied-SLA evaluation jobs through the durable worker; P5.8 queues Chatwoot-style contact export artifact generation through durable `contact:export` jobs; P5.9 queues normalized provider inbound message persistence/dispatch through durable `webhook:incoming_message_persist` jobs; P5.10 queues Chatwoot `SendReplyJob`-style outbound message delivery through durable `message:send_reply` jobs and provider delivery-status/read-receipt updates through durable webhook status jobs; P5.11 queues Captain document sync, crawl/parser, schedule-sync, response-builder, embedding-update, Copilot response, and Captain conversation response-builder work through durable jobs; P5.12 queues scheduled item fan-out, one-off campaigns, snoozed conversation reopening, account auto-resolution, widget/public message status updates, and account conversation bulk actions through durable jobs. Next active implementation slice is P5.13 analytics aggregation. - `go test ./...` passes. - Route dump succeeds with `TOTAL: 832` after adding the Chatwoot-compatible Twilio delivery-status route plus the legacy namespaced alias. - Route parity artifacts now exist under `docs/parity/` and are generated by `cmd/route_parity`. @@ -45,8 +45,8 @@ Next ordered checkpoints: | Order | Slice | Required outcome | Primary verification | | --- | --- | --- | --- | -| 1 | P5.11 remaining Captain/Copilot jobs | Queue remaining Copilot/conversation response work behind fakeable provider gates. | Copilot worker tests and disabled/failure-state fixtures. | -| 2 | P5.13 analytics aggregation | Replace frontend-visible placeholder report values with scheduled or cached real aggregations. | Report service/handler fixtures and freshness/idempotency tests. | +| 1 | P5.13 analytics aggregation | Replace frontend-visible placeholder report values with scheduled or cached real aggregations. | Report service/handler fixtures and freshness/idempotency tests. | +| 2 | B9.3 delayed automation actions | Confirm current-reference delayed automation params and queue any still-synchronous scheduled action execution. | Automation worker fixtures for schedule time, retry, idempotency, and observable failure. | | 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. | @@ -54,9 +54,9 @@ Next ordered checkpoints: This checkpoint is intended to make the development plan complete enough to track without reading Hermes notes first. -- The next active implementation slice is Phase 5 remaining job parity, starting with P5.11c Copilot/conversation response jobs. B12 has repeatable API/browser/enterprise smoke harnesses in Review; optional live failures should be converted into named slices instead of blocking Phase 5 job work. +- The next active implementation slice is P5.13 analytics aggregation. P5.11 Captain/Copilot response jobs are now in Review, and B12 has repeatable API/browser/enterprise smoke harnesses in Review; optional live failures should be converted into named slices instead of blocking Phase 5 job work. - 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. +- The Hermes automation/macro/CSAT plan is fully represented by B8/B9 and Phase 5. Durable delayed execution remains visible Phase 5/B9.3 work if the current reference exposes explicit delayed action params; channel-specific template delivery is in Review. - 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. - Every future commit must update one of: the Current Baseline, Commit Ledger, active B-slice board, Phase 5 backlog, B12 smoke report link, or Progress Log. @@ -64,7 +64,7 @@ Open work after the current checkpoint: | Area | Next concrete action | Tracking location | Done boundary | | --- | --- | --- | --- | -| Phase 5 jobs | Finish remaining Captain/Copilot jobs and analytics aggregation on top of the committed durable worker. | `Phase 5: Background Jobs And Integrations` | Worker tests prove enqueue, retry, idempotency, and fakeable external effects. | +| Phase 5 jobs | Finish analytics aggregation and the B9.3 delayed automation action check on top of the committed durable worker. | `Phase 5: Background Jobs And Integrations` | Report and automation tests prove real aggregation/scheduling, freshness/idempotency, and no frontend-visible placeholder values. | | B12 smoke | Run optional live API/browser/enterprise smoke in a full PostgreSQL/Redis/Meilisearch/Vite environment and convert failures into named slices. | `B12 reused frontend verification breakdown` | `docs/parity/frontend_smoke_report.md` records checked pass/fail results and maps failures to slices. | | 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. | | Phase 6 placeholders | Re-run placeholder audit and burn down any frontend-reachable stub. | `Phase 6: Core Product Placeholder Burn-down` | Stub list has no reused-frontend critical path without a named owner. | @@ -78,7 +78,7 @@ Open work after the current checkpoint: | 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, B10, and B11 are in Review; B12 reused frontend smoke harnesses exist and optional live runs can expose follow-up slices | -| Phase 5 | Background jobs and integrations | Doing | P5.1/P5.2/P5.3/P5.4/P5.5/P5.6/P5.7/P5.8/P5.9/P5.10/P5.11/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export, inbound webhook persistence, outbound/provider delivery status, Captain document sync/crawl/response/embedding, conversation maintenance, message status update, and account bulk-action cores are in Review; remaining Copilot/conversation response and analytics aggregation jobs remain open | +| Phase 5 | Background jobs and integrations | Doing | P5.1/P5.2/P5.3/P5.4/P5.5/P5.6/P5.7/P5.8/P5.9/P5.10/P5.11/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export, inbound webhook persistence, outbound/provider delivery status, Captain document sync/crawl/response/embedding/Copilot/conversation responses, conversation maintenance, message status update, and account bulk-action cores are in Review; analytics aggregation and the B9.3 delayed automation action check remain open | | Phase 6 | Core placeholder burn-down | Doing | account/contact/conversation/message/inbox placeholder groups remain broad | | Phase 7 | Verification harness | Review | B12.1 boot/readiness, B12.2a API assertions, B12.2b browser smoke harness, B12.3a enterprise API assertions, and B12.3b enterprise browser route navigation exist; optional live Meilisearch/full-browser runs remain environment-dependent | @@ -88,8 +88,8 @@ This table is the shortest authoritative handoff view. If an older lower section | Priority | Workstream | Current state | Next checkpoint | Commit close rule | | --- | --- | --- | --- | --- | -| 1 | P5.11 Captain/Copilot jobs | Captain document sync, crawl/parser/schedule, response-builder, and embedding jobs are queued; Copilot/conversation response jobs remain open. | Add fakeable durable jobs for remaining Copilot/conversation async work and disabled-provider states. | Focused Copilot worker tests plus full `go test ./...`. | -| 2 | P5.13 reports/analytics | Frontend smoke harness exists; `analytics_service` still has placeholder aggregation paths. | Replace frontend-visible report placeholders with real scheduled/cached aggregations. | Report fixtures verify values, cache/freshness behavior, and no hidden placeholder JSON. | +| 1 | P5.13 reports/analytics | Frontend smoke harness exists; `analytics_service` still has placeholder aggregation paths. | Replace frontend-visible report placeholders with real scheduled/cached aggregations. | Report fixtures verify values, cache/freshness behavior, and no hidden placeholder JSON. | +| 2 | B9.3 delayed automation actions | Macro fan-out, webhook/transcript delivery, and CSAT jobs are durable; current-reference delayed automation action params still need a final check. | Queue any still-synchronous delayed automation action execution or close the row with reference evidence if no explicit delayed params exist. | Automation worker fixtures verify schedule time, retry, idempotency, and observable failure. | | 3 | Phase 2/3 drift | Tracked route parity is 0 missing for the current critical set; serializer fixtures remain partial. | Expand route/serializer fixtures when smoke or reference inspection exposes drift. | Regenerate parity artifacts and add endpoint-family fixture tests. | | 4 | Phase 6 placeholder audit | Widget/public/webhook critical placeholders are burned down; account/contact/conversation/message/inbox audit remains broad. | Run a fresh placeholder audit and assign every frontend-reachable stub to a tracked owner. | `rg` audit result is recorded and no reused-frontend blocker is ownerless. | | 5 | B12 optional live smoke | API/browser/enterprise smoke commands are checked in; live runs need PostgreSQL, Redis, Meilisearch, Vite, and Chrome. | Run full live smoke when environment is available and map failures to the board. | `docs/parity/frontend_smoke_report.md` records pass/fail and linked owners. | @@ -102,7 +102,7 @@ These rows are the executable development plan from this point forward. A checkp | --- | --- | --- | --- | --- | | P5.11a Captain document crawl/schedule | `internal/service/captain_document_service.go`, `internal/service/captain_document_worker.go`, `internal/app/bootstrap.go` | `reference/chatwoot/enterprise/app/jobs/captain/documents/crawl_job.rb`, `schedule_syncs_job.rb`, `perform_sync_job.rb`, Firecrawl/simple parser jobs | Durable schedule/crawl producers and handlers with fakeable crawl/parser boundaries. Missing provider config is a failed `crawl_disabled` state, not placeholder success. | Review by `feat(captain): queue document crawl jobs`; focused worker tests prove enqueue, replay, account scope, idempotent scheduler, and disabled/failure states. | | P5.11b Captain response/embedding fan-out | Captain document/assistant-response services and repositories, Meilisearch/embedding boundaries | `response_builder_job.rb`, `enterprise/app/jobs/captain/llm/update_embedding_job.rb`, FAQ generator/embedding services | Queue FAQ response generation after successful document content changes, reset unedited responses, create/update assistant responses, and fan out embedding update work behind fakeable LLM gates. | Review by `feat(captain): queue response embedding jobs`; tests cover response reset/create, embedding-disabled retry, fake embedding success, account scope, and no external network in default tests. | -| P5.11c Copilot and conversation response jobs | Copilot thread/message services, Captain task services, event/realtime payload helpers | `enterprise/app/jobs/captain/copilot/response_job.rb`, `conversation/response_builder_job.rb`, Copilot chat service | Queue assistant replies after Copilot user messages and Captain conversation triggers. Persist assistant messages/tool-call JSON and emit existing push payloads from worker replay. | Copilot worker tests cover no-provider disabled reply, fake provider success/failure, user/account scope, and restart-safe persistence. | +| P5.11c Copilot and conversation response jobs | `internal/service/copilot_service.go`, `internal/service/copilot_response_worker.go`, `internal/service/captain_conversation_service.go`, `internal/service/message_service.go`, `internal/app/bootstrap.go` | `enterprise/app/jobs/captain/copilot/response_job.rb`, `enterprise/app/jobs/captain/conversation/response_builder_job.rb`, `enterprise/app/services/captain/copilot/chat_service.rb`, `enterprise/app/services/enterprise/message_templates/hook_execution_service.rb`, `enterprise/app/models/copilot_message.rb` | Queue assistant replies after Copilot user messages and Captain pending-conversation triggers. Persist assistant messages, enqueue Captain conversation replies/handoff messages, open handoff conversations, and keep fakeable provider disabled/failure states observable through durable retry. | Review by `feat(captain): queue copilot response jobs`; focused tests cover Copilot enqueue/persist/fallback/retry, Captain conversation enqueue/handoff/retry/non-pending skip, and service/worker/app package replay. | | P5.13 analytics aggregation | `internal/service/analytics_service.go`, report handlers/services, worker bootstrap | Chatwoot report controllers/services used by dashboard analytics, CSAT/SLA reporting views | Replace frontend-visible placeholder report values with real scheduled or cached aggregations. Define freshness/idempotency rules for expensive rollups. | Report fixtures prove values are derived from persisted conversations/messages/CSAT/SLA rows; `rg` finds no frontend-visible placeholder report JSON. | | Phase 2/3 drift audit | `cmd/route_parity`, `docs/parity/*`, serializer tests | `reference/chatwoot/config/routes.rb`, controller Jbuilder views, reused frontend API clients | Convert any smoke/reference mismatch into a named route, controller, or serializer slice. Static route extraction remains acceptable until Ruby/Bundler is available. | Regenerated route parity shows 0 missing tracked frontend routes; new serializer fixtures cover the drift. | | Phase 6 placeholder burn-down | Account/contact/conversation/message/inbox handlers and services | Matching reference controllers/Jbuilder views plus reused frontend screens | Re-run placeholder audit and assign every frontend-reachable stub to a specific owner. Burn down the highest-impact stubs before broad feature expansion. | `rg` placeholder audit is recorded here; no reused-frontend critical path is ownerless. | @@ -110,10 +110,9 @@ These rows are the executable development plan from this point forward. A checkp Checkpoint sequencing: -1. Finish P5.11a-c before starting analytics. Captain/Copilot async behavior is frontend-visible and depends on the durable worker already in place. -2. Run P5.13 after Captain/Copilot jobs so report aggregation can include final message/job side effects. -3. Run Phase 2/3 and Phase 6 audits after each smoke failure or route/serializer change, not as a one-time cleanup. -4. Keep B12 live smoke optional until the full external stack is available, but every failed live smoke must become a named row in this table. +1. Run P5.13 now that P5.11a-c Captain/Copilot async behavior is in Review, so report aggregation can include final message/job side effects. +2. Run Phase 2/3 and Phase 6 audits after each smoke failure or route/serializer change, not as a one-time cleanup. +3. Keep B12 live smoke optional until the full external stack is available, but every failed live smoke must become a named row in this table. Commit discipline: @@ -140,6 +139,7 @@ This ledger records the committed parity checkpoints that future slices should b | Commit | Scope | Verification summary | Follow-up state | | --- | --- | --- | --- | +| `feat(captain): queue copilot response jobs` | Completes P5.11c with durable Chatwoot `Captain::Copilot::ResponseJob` and `Captain::Conversation::ResponseBuilderJob` equivalents. Copilot thread/message creation now persists the user message and enqueues `captain:copilot_response` when a WorkerPool is configured, while worker replay reloads the account/user/thread/message scope and persists assistant replies through a fakeable backend or the existing no-provider fallback. Incoming pending conversation messages for Captain-enabled inboxes now enqueue `captain:conversation_response_builder`; replay collects public incoming/outgoing history, creates Captain outgoing replies, enqueues provider send-reply, and opens the conversation with a handoff message when the backend requests handoff. | `go test ./internal/service -run 'CopilotResponse\|CaptainConversation\|MessageService' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | P5.11 moves to Review; continue P5.13 analytics aggregation, then Phase 2/3 drift and Phase 6 placeholder audits as smoke exposes gaps. | | `feat(captain): queue response embedding jobs` | Advances P5.11b with durable Captain document response building and embedding update fan-out. Successful document sync/parser content updates enqueue `captain:document_response_builder`, worker replay resets only unedited document responses, preserves edited responses, creates approved `Captain::Document` assistant responses from a fakeable FAQ backend, and enqueues `captain:llm_update_embedding` jobs for created responses. Embedding replay reloads account-scoped responses, uses a fakeable embedding backend or configured LLM provider, and surfaces missing provider config as retryable worker failures. | `go test ./internal/service -run 'CaptainDocumentService\|EnqueueCaptainDocumentScheduleSyncs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | P5.11b moves to Review; continue P5.11c Copilot/conversation response jobs, then P5.13 analytics aggregation. | | `feat(captain): queue document crawl jobs` | Advances P5.11a with durable Captain document crawl, simple-page parser, and auto-sync scheduler jobs. Document create/crawl requests enqueue `captain:document_crawl`, crawl replay uses a fakeable crawl backend to enqueue normalized `captain:document_page_crawl_parse` jobs, parser replay creates or updates account-scoped document content through a fakeable parser backend, and the `captain:documents_schedule_syncs` root job scans stale synced/failed/syncing documents for accounts with `captain_document_auto_sync` enabled before enqueueing `captain:document_sync` work with daily idempotency. Bootstrap seeds the scheduler after Captain document services are wired. | `go test ./internal/service -run 'CaptainDocumentService\|EnqueueCaptainDocumentScheduleSyncs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | P5.11a moves to Review; continue P5.11b response/embedding fan-out, then P5.11c Copilot/conversation response jobs. | | `docs: land parity execution tracker` | Documentation-only checkpoint requested before continuing implementation. Updates the current baseline to `d756473`, front-loads executable open checkpoint contracts for P5.11 Captain/Copilot jobs, P5.13 analytics aggregation, Phase 2/3 drift, Phase 6 placeholder burn-down, and B12 live smoke, and records commit discipline for future parity slices. | `git diff --check`; documentation-only change. | Start P5.11a Captain document crawl/schedule jobs, then P5.11b response/embedding fan-out and P5.11c Copilot/conversation response jobs. | @@ -1578,8 +1578,8 @@ 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, 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.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, stream/realtime fallback payloads through B11.3e, and durable P5.11 Captain/Copilot response jobs 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, download CSV, and channel-specific template job hooks are aligned. | 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 | @@ -1595,7 +1595,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`; stream disabled states and push payload shapes are covered by `feat(captain): align streaming fallbacks`. | Review | +| Captain/Copilot | Assistants, inbox bindings, scenarios, responses, documents, tools, preferences, copilot threads/messages, tasks, streaming/tool-call behavior, durable response jobs. | 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`; durable document/Copilot/conversation jobs are covered by P5.11 checkpoints through `feat(captain): queue copilot response jobs`. | Review | Enterprise acceptance gates: @@ -1639,7 +1639,7 @@ Known hotspots: - `internal/worker/worker.go` has the committed P5.1 durable core and is already wired to dispatcher, search indexing, automation delivery, macro fan-out, CSAT survey/template sends, SLA scans, contact exports, Captain document sync, and conversation maintenance jobs. - `internal/service/message_delivery_worker.go` and `internal/handler/webhook/incoming_persister_jobs.go` now cover P5.10 SendReplyJob-style outbound delivery plus provider delivery-status/read-receipt jobs. - `internal/handler/webhook/*` and provider services now keep verification/parsing in the HTTP path while `IncomingPersister` moves normalized inbound persistence/dispatch behind durable jobs when the WorkerPool is configured. -- `internal/service/captain_*` and `internal/service/copilot_*` now cover Captain document sync, crawl/parser, schedule-sync, response-builder, and embedding-update jobs; P5.11 still needs Copilot/conversation response jobs. +- `internal/service/captain_*` and `internal/service/copilot_*` now cover Captain document sync, crawl/parser, schedule-sync, response-builder, embedding-update, Copilot response, and Captain conversation response-builder jobs. - `internal/service/analytics_service.go` has placeholder analytics/report paths and is the owner for P5.13 aggregation work. Checklist: @@ -1647,7 +1647,7 @@ Checklist: - [x] Add `background_jobs` persistence with job type, payload, queue, status, attempt counters, scheduled/locked timestamps, idempotency key, last error, and completion timestamps. - [x] Replace `internal/worker/worker.go` placeholder with enqueue, schedule, perform, retry/backoff, dead-letter, and graceful shutdown behavior. - [x] Map the committed Chatwoot job/listener families to Go worker responsibilities in the tracking table. -- [ ] Finish durable job dispatch for remaining Copilot/conversation response jobs, delayed automation actions, and report aggregation. +- [ ] Finish durable job dispatch for delayed automation actions if current-reference params require it, and finish report aggregation. - [ ] Add retry and failure logging for the remaining external provider calls. - [ ] Add tests for the remaining enqueueing, uniqueness/idempotency, delayed execution, retries, worker restart pickup, and fakeable external side effects. @@ -1670,7 +1670,7 @@ Tracking table: | P5.8 | Queue contact export artifact generation and completion notification/email. | `account/contacts_export_job.rb` | contact export service and mailer boundary | Export API returns immediately, artifact generation is durable, completion notification/email is fakeable, and download remains stable after restart. | Review by `feat(crm): queue contact exports` | | P5.9 | Queue inbound provider webhook processing where Chatwoot defers work. | `webhooks/*_events_job.rb`, `webhooks/*_delivery_job.rb` | webhook handlers, provider services | Provider HTTP ack behavior remains Chatwoot-compatible while persistence/dispatch runs through retryable jobs with signature validation preserved. | Review by `feat(webhook): queue incoming messages` | | P5.10 | Queue outbound message delivery and delivery-status updates. | `send_reply_job.rb`, provider delivery/status jobs | message send/channel services, delivery status handler | Outgoing message creation and provider delivery are separated; retries update message/delivery status exactly once. | Review by `feat(messages): queue send replies` and `feat(messages): queue delivery statuses` | -| P5.11 | Queue Captain document sync, crawl, response building, embeddings, and Copilot responses. | Captain document/crawl/response/embedding/Copilot jobs | `internal/service/captain_document_service.go`, Captain/Copilot services | Existing fakeable disabled/failure gates run under durable jobs; document statuses and Copilot message persistence survive worker restart. | Doing: document sync Review by `feat(captain): queue document syncs`; crawl/schedule Review by `feat(captain): queue document crawl jobs`; response/embedding Review by `feat(captain): queue response embedding jobs`; Copilot response jobs Todo | +| P5.11 | Queue Captain document sync, crawl, response building, embeddings, and Copilot responses. | Captain document/crawl/response/embedding/Copilot/conversation jobs | `internal/service/captain_document_service.go`, `internal/service/copilot_service.go`, `internal/service/captain_conversation_service.go`, Captain/Copilot services | Existing fakeable disabled/failure gates run under durable jobs; document statuses, Copilot message persistence, and Captain conversation replies survive worker restart. | Review by `feat(captain): queue document syncs`, `feat(captain): queue document crawl jobs`, `feat(captain): queue response embedding jobs`, and `feat(captain): queue copilot response jobs` | | P5.12 | Queue conversation maintenance jobs. | `trigger_scheduled_items_job.rb`, `campaigns/trigger_oneoff_campaign_job.rb`, `conversations/resolution_job.rb`, `reopen_snoozed_conversations_job.rb`, `update_message_status_job.rb`, `bulk_actions_job.rb` | conversation service/handlers | Auto-resolution, snooze reopen, status updates, and bulk actions are scheduled/retryable with idempotent tests. | Review by `feat(conversations): queue maintenance jobs`, `feat(conversations): queue message status updates`, and `feat(conversations): queue bulk actions` | | P5.13 | Replace placeholder analytics/report builders that need background aggregation. | reporting jobs/services and report controllers | `internal/service/analytics_service.go`, reporting services | Frontend-visible reports no longer use placeholder values; any expensive aggregation is scheduled or cached with freshness rules. | Todo | @@ -1680,7 +1680,7 @@ P5.1 current checkpoint: - Replaced the `WorkerPool` stub with enqueue/schedule APIs, handler registration, `ProcessOne`, start/stop worker loops, PostgreSQL `FOR UPDATE SKIP LOCKED` claiming, retry/backoff, dead-letter state, and stale-lock recovery for worker restart pickup. - Kept older `NewWorkerPool()` construction as a no-op-compatible path while adding `NewWorkerPoolWithOptions(db, ...)` for durable wiring. - Added focused tests proving enqueue/idempotency, due job completion, retry to dead-letter, queue/schedule filtering, and stale running job requeue. -- Remaining Phase 5 work: delayed automation scheduled-item execution, Copilot/conversation response jobs, and analytics/report aggregation. +- Remaining Phase 5 work: delayed automation scheduled-item execution if reference params require it, and analytics/report aggregation. P5.1 verification: @@ -1696,7 +1696,7 @@ P5.2 current checkpoint: - `dispatch.EventDispatcher` can now attach a `WorkerPool`; sync listeners still execute inline, while async listeners and `DispatchAsync` subscribers enqueue durable per-listener `event:listener_dispatch` jobs. - Worker replay resolves the listener by name and invokes `OnEvent` with the serialized event payload. Missing listeners or malformed payloads fail the job, preserving retry/dead-letter visibility. - `WorkerPool` default queue handling was broadened to process all queues unless explicitly filtered, so `events` jobs are processed by the default worker. -- Remaining integration work is feature-specific: delayed automation/macros, Captain/Copilot, and analytics aggregation must register producers/handlers on this durable path. +- Remaining integration work is feature-specific: delayed automation actions if still required and analytics aggregation must register producers/handlers on this durable path. P5.2 verification: @@ -1730,7 +1730,7 @@ P5.4 current checkpoint: - The durable job handlers reuse the existing fakeable HTTP webhook and SMTP transcript delivery interfaces, so timeout/retry behavior remains tested at the delivery boundary while worker attempts, errors, retry schedules, completion, and dead-letter state are visible through `background_jobs`. - `AutomationRuleService`, `AutomationRuleListener`, and `MacroService` can carry the WorkerPool through action execution. `Bootstrap` now creates a durable WorkerPool, registers automation delivery jobs, wires the dispatcher with event jobs, and starts the worker in `App.Run`. - Existing no-worker constructors still preserve the previous synchronous fallback for focused tests and development paths that do not start the durable worker yet. -- Remaining Phase 5 work: delayed automation actions, macro fan-out jobs, CSAT channel-specific sends/templates, Meilisearch indexing fan-out, SLA scans, contact export email, provider webhook/offline delivery, Captain/Copilot, and conversation maintenance. +- Remaining Phase 5 work: delayed automation actions if current-reference params require them and analytics/report aggregation; macro fan-out, CSAT sends/templates, Meilisearch indexing fan-out, SLA scans, contact export email, provider webhook/offline delivery, Captain/Copilot, and conversation maintenance are in Review. P5.4 verification: @@ -1767,7 +1767,7 @@ P5.7 current checkpoint: - `sla:process_account` mirrors `Sla::ProcessAccountAppliedSlasJob`: it finds active and active_with_misses AppliedSLA rows for the account and queues `sla:process_applied` jobs. - `sla:process_applied` mirrors `Sla::ProcessAppliedSlaJob`: it calls the existing `AppliedSlaService.Evaluate`, preserving idempotent miss events, notification fan-out, retry/backoff, and dead-letter visibility through `background_jobs`. - Bootstrap registers the SLA job handlers and enqueues the initial root scan when the app boots. -- Remaining Phase 5 work: delayed automation scheduled-item execution, Captain/Copilot durable jobs, and analytics aggregation. +- Remaining Phase 5 work: delayed automation scheduled-item execution if reference params require it, and analytics aggregation. P5.7 verification: @@ -1785,7 +1785,7 @@ P5.8 current checkpoint: - Completed export jobs are idempotent: a duplicate replay of an already-completed export is a no-op, so completion notifications and mail are not duplicated. - Failed worker execution records the export error and leaves the background job retry/dead-letter state observable through `background_jobs`. - No-worker construction still uses the synchronous fallback for focused tests and local paths that do not start the durable worker. -- Remaining Phase 5 work: delayed automation scheduled-item execution, Captain/Copilot durable jobs, and analytics aggregation. +- Remaining Phase 5 work: delayed automation scheduled-item execution if reference params require it, and analytics aggregation. P5.8 verification: @@ -1852,17 +1852,23 @@ P5.11 current checkpoint: - `captain:documents_schedule_syncs` now mirrors the reference auto-sync scheduler shape on `scheduled_jobs`. Bootstrap seeds the scheduler, scheduler jobs use daily idempotency, scan stale synced/failed/syncing completed documents for accounts with `captain_document_auto_sync` enabled, and enqueue `captain:document_sync` work on the `purgable` queue for retryable execution. - `captain:document_response_builder` now mirrors Chatwoot enterprise `Captain::Documents::ResponseBuilderJob` on the `low` queue. Successful document sync/parser content updates enqueue response generation, worker replay resets only unedited document responses, preserves edited responses, creates approved `Captain::Document` assistant responses from a fakeable FAQ backend, and keeps missing FAQ provider config as an observable retry error. - `captain:llm_update_embedding` now mirrors `Captain::Llm::UpdateEmbeddingJob` for assistant responses. Response-builder replay fans out embedding jobs for created responses, embedding replay reloads the account-scoped response, uses a fakeable embedding backend or the configured LLM provider, and records disabled/missing provider failures through `background_jobs` retry state. +- `captain:copilot_response` now mirrors Chatwoot enterprise `Captain::Copilot::ResponseJob` on the default queue. Copilot thread/message creation persists the user message first, enqueues durable assistant generation when a WorkerPool is configured, and worker replay reloads the account/user/thread/user-message scope before persisting assistant messages. +- Copilot response generation is fakeable for tests and keeps the existing no-provider unavailable-credit fallback as a persisted assistant message; backend failures remain retryable through `background_jobs` instead of being hidden behind synchronous request success. +- `captain:conversation_response_builder` now mirrors Chatwoot enterprise `Captain::Conversation::ResponseBuilderJob` for pending conversations in Captain-enabled inboxes. Incoming public messages enqueue durable work, attachment-bearing messages are scheduled with a short delay, and non-pending/private/non-Captain conversations are skipped before enqueue. +- Conversation response replay collects public incoming/outgoing history, maps incoming to user and outgoing to assistant context, creates Captain outgoing messages with optional `agent_name`, enqueues `message:send_reply`, and handles handoff responses by creating the configured handoff message and opening the pending conversation. - No-worker construction keeps the previous mark-syncing fallback for focused tests and local paths that do not start the durable worker. -- Bootstrap registers Captain document jobs on the shared WorkerPool and wires the assistant-response repository so production sync/crawl/schedule/response/embedding requests are replayable after process restart. -- Remaining P5.11 work: Copilot/conversation response jobs. +- Bootstrap registers Captain document, Copilot response, and Captain conversation response jobs on the shared WorkerPool and wires the assistant-response repository so production sync/crawl/schedule/response/embedding/Copilot/conversation requests are replayable after process restart. +- Remaining P5.11 work: none known for the named durable Captain/Copilot job row; continue P5.13 analytics aggregation. P5.11 verification: ```bash env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'CaptainDocumentService' -count=1 env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'CaptainDocumentService\|EnqueueCaptainDocumentScheduleSyncs' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'CopilotResponse\|CaptainConversation\|MessageService' -count=1 env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/handler/api/v1 -run 'CaptainDocument\|CaptainResource' -count=1 env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service ./internal/worker ./internal/app -count=1 env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./... git diff --check ``` @@ -1993,6 +1999,7 @@ Verification milestone gates: ## Progress Log +- 2026-06-05: P5.11c Copilot/conversation response checkpoint prepared as `feat(captain): queue copilot response jobs`; Copilot thread/message creation now enqueues `captain:copilot_response` after persisting the user message, worker replay persists assistant replies through a fakeable backend or the existing no-provider fallback, and backend failures surface as retryable jobs. Pending Captain-enabled incoming conversations now enqueue `captain:conversation_response_builder`, replay creates Captain outgoing replies, queues provider send-reply, and opens conversations on handoff. Focused Copilot/Captain conversation worker tests, service/worker/app package tests, full `go test ./...`, and `git diff --check` passed. P5.11 moves to Review; next slice is P5.13 analytics aggregation. - 2026-06-05: P5.11b Captain response/embedding checkpoint prepared as `feat(captain): queue response embedding jobs`; successful document sync/parser content updates now enqueue `captain:document_response_builder`, response-builder replay resets only unedited document responses, preserves edited responses, creates approved `Captain::Document` assistant responses through a fakeable FAQ backend, and fans out `captain:llm_update_embedding` jobs. Embedding replay reloads account-scoped responses and uses a fakeable embedding backend or configured LLM provider, with missing providers surfacing as retryable worker failures. Focused Captain document worker tests, service/worker/app package tests, full `go test ./...`, and `git diff --check` passed. Remaining P5.11 follow-up is Copilot/conversation response jobs. - 2026-06-05: P5.11a Captain document crawl/schedule checkpoint prepared as `feat(captain): queue document crawl jobs`; document create/crawl requests now enqueue `captain:document_crawl`, crawl replay fans out normalized `captain:document_page_crawl_parse` parser jobs through fakeable crawl/parser boundaries, parser replay creates or updates account-scoped documents with synced content/fingerprints, and `captain:documents_schedule_syncs` scans stale completed documents for `captain_document_auto_sync` accounts to enqueue `captain:document_sync` jobs with daily idempotency. Focused Captain document worker tests, service/worker/app package tests, full `go test ./...`, and `git diff --check` passed. Remaining P5.11 follow-up is now Copilot/conversation response jobs after P5.11b. - 2026-06-05: Parity execution tracker checkpoint prepared as `docs: land parity execution tracker`; the plan now records `d756473 feat(webhook): queue incoming messages` as the current implementation baseline, adds executable contracts for P5.11a/P5.11b/P5.11c Captain/Copilot jobs, P5.13 analytics aggregation, Phase 2/3 drift audit, Phase 6 placeholder burn-down, and B12 live smoke, and makes commit/update discipline explicit for future checkpoints. Documentation-only checkpoint; `git diff --check` passed. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index ec3ab358..ece8bbe5 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -589,6 +589,10 @@ func Bootstrap(env string) (*App, error) { captainScenarioService := service.NewCaptainScenarioService(captainScenarioRepo, captainAssistantRepo) captainCustomToolService := service.NewCaptainCustomToolService(captainCustomToolRepo) copilotService := service.NewCopilotService(copilotThreadRepo, copilotMessageRepo, copilotSuggestionRepo, llmProvider, captainAssistantRepo) + copilotService.SetWorkerPool(workerPool) + captainConversationService := service.NewCaptainConversationService(db, llmProvider) + captainConversationService.SetWorkerPool(workerPool) + _ = captainConversationService copilotContextService := service.NewCopilotContextService(messageRepo, conversationRepo, contactRepo, llmProvider) captainTaskService := service.NewCaptainTaskService(captainAssistantRepo, captainAssistantResponseRepo, captainCustomToolRepo, conversationRepo, messageRepo, llmProvider, copilotContextService, copilotSuggestionRepo) conversationInsightService := service.NewConversationInsightService(conversationRepo, messageRepo, captainAssistantRepo, llmProvider) diff --git a/internal/service/captain_conversation_service.go b/internal/service/captain_conversation_service.go new file mode 100644 index 00000000..65bd990b --- /dev/null +++ b/internal/service/captain_conversation_service.go @@ -0,0 +1,299 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync" + "time" + + "github.com/gochat/gochat/internal/llm" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "gorm.io/datatypes" + "gorm.io/gorm" +) + +const TaskTypeCaptainConversationResponseBuilder = "captain:conversation_response_builder" + +type CaptainConversationResponseBackend interface { + GenerateCaptainConversationResponse(ctx context.Context, req CaptainConversationResponseRequest) (*CaptainConversationResponse, error) +} + +type CaptainConversationResponseRequest struct { + AccountID uint + Conversation *model.Conversation + Assistant *model.CaptainAssistant + Messages []CaptainConversationMessage +} + +type CaptainConversationMessage struct { + Role string + Content string + AgentName string +} + +type CaptainConversationResponse struct { + Content string + Action string + AgentName string + HandoffToolCalled bool +} + +type CaptainConversationService struct { + db *gorm.DB + llmProvider llm.Provider + backend CaptainConversationResponseBackend + worker *worker.WorkerPool +} + +func NewCaptainConversationService(db *gorm.DB, llmProvider llm.Provider) *CaptainConversationService { + return &CaptainConversationService{db: db, llmProvider: llmProvider} +} + +func (s *CaptainConversationService) SetResponseBackend(backend CaptainConversationResponseBackend) { + s.backend = backend +} + +func (s *CaptainConversationService) SetWorkerPool(wp *worker.WorkerPool) { + s.worker = wp + RegisterCaptainConversationJobs(wp, s) +} + +func (s *CaptainConversationService) BuildConversationResponseByAccount(ctx context.Context, accountID, conversationID, assistantID uint) (*model.Message, error) { + if s == nil || s.db == nil { + return nil, fmt.Errorf("captain conversation service database is required") + } + var conversation model.Conversation + if err := s.db.WithContext(ctx).Where("account_id = ? AND id = ?", accountID, conversationID).First(&conversation).Error; err != nil { + return nil, fmt.Errorf("conversation not found: %w", err) + } + if conversation.Status != string(model.ConversationStatusPending) { + return nil, nil + } + var assistant model.CaptainAssistant + if err := s.db.WithContext(ctx).Where("account_id = ? AND id = ?", accountID, assistantID).First(&assistant).Error; err != nil { + return nil, fmt.Errorf("assistant not found: %w", err) + } + + history, err := s.collectConversationMessages(ctx, accountID, conversation.ID) + if err != nil { + return nil, err + } + response, err := s.generateConversationResponse(ctx, accountID, &conversation, &assistant, history) + if err != nil { + return nil, err + } + if response == nil { + return nil, fmt.Errorf("empty captain conversation response") + } + if response.Action == "handoff" || response.Content == "conversation_handoff" || response.HandoffToolCalled { + return s.createCaptainHandoffMessage(ctx, &conversation, &assistant) + } + if strings.TrimSpace(response.Content) == "" { + return nil, fmt.Errorf("captain conversation response content is blank") + } + return s.createCaptainOutgoingMessage(ctx, &conversation, &assistant, response.Content, response.AgentName) +} + +func (s *CaptainConversationService) collectConversationMessages(ctx context.Context, accountID, conversationID uint) ([]CaptainConversationMessage, error) { + var messages []model.Message + if err := s.db.WithContext(ctx). + Where("account_id = ? AND conversation_id = ? AND private = ? AND message_type IN ?", accountID, conversationID, false, []string{string(model.MessageTypeIncoming), string(model.MessageTypeOutgoing)}). + Order("created_at ASC, id ASC"). + Find(&messages).Error; err != nil { + return nil, fmt.Errorf("load conversation messages: %w", err) + } + history := make([]CaptainConversationMessage, 0, len(messages)) + for _, msg := range messages { + role := "assistant" + if msg.MessageType == string(model.MessageTypeIncoming) { + role = "user" + } + history = append(history, CaptainConversationMessage{Role: role, Content: msg.Content, AgentName: messageAgentName(msg.AdditionalAttributes)}) + } + return history, nil +} + +func (s *CaptainConversationService) generateConversationResponse(ctx context.Context, accountID uint, conversation *model.Conversation, assistant *model.CaptainAssistant, history []CaptainConversationMessage) (*CaptainConversationResponse, error) { + if s.backend != nil { + return s.backend.GenerateCaptainConversationResponse(ctx, CaptainConversationResponseRequest{AccountID: accountID, Conversation: conversation, Assistant: assistant, Messages: history}) + } + if s.llmProvider == nil { + return nil, fmt.Errorf("captain conversation response generation disabled") + } + messages := []llm.ChatMessage{{Role: "system", Content: "You are Captain, a customer support assistant. Reply to the latest customer message. Return only the reply text."}} + for _, msg := range history { + if strings.TrimSpace(msg.Content) == "" { + continue + } + messages = append(messages, llm.ChatMessage{Role: msg.Role, Content: msg.Content}) + } + resp, err := s.llmProvider.ChatCompletion(ctx, llm.ChatRequest{Model: "gpt-4", Messages: messages, Temperature: 0.7, MaxTokens: 1024}) + if err != nil { + return nil, fmt.Errorf("generate captain conversation response: %w", err) + } + if len(resp.Choices) == 0 || strings.TrimSpace(resp.Choices[0].Message.Content) == "" { + return nil, fmt.Errorf("empty captain conversation response") + } + return &CaptainConversationResponse{Content: resp.Choices[0].Message.Content}, nil +} + +func (s *CaptainConversationService) createCaptainOutgoingMessage(ctx context.Context, conversation *model.Conversation, assistant *model.CaptainAssistant, content, agentName string) (*model.Message, error) { + senderID := assistant.ID + attrs := datatypes.JSON([]byte(`{}`)) + if strings.TrimSpace(agentName) != "" { + raw, _ := json.Marshal(map[string]any{"agent_name": strings.TrimSpace(agentName)}) + attrs = datatypes.JSON(raw) + } + message := &model.Message{ + AccountID: conversation.AccountID, + ConversationID: conversation.ID, + InboxID: conversation.InboxID, + SenderID: &senderID, + SenderType: "CaptainAssistant", + Content: content, + ContentType: string(model.MessageContentTypeText), + MessageType: string(model.MessageTypeOutgoing), + Status: string(model.MessageStatusSent), + AdditionalAttributes: attrs, + } + if err := s.db.WithContext(ctx).Create(message).Error; err != nil { + return nil, fmt.Errorf("create captain outgoing message: %w", err) + } + if s.worker != nil { + if _, err := EnqueueSendReply(ctx, s.worker, message.ID); err != nil { + return message, err + } + } + return message, nil +} + +func (s *CaptainConversationService) createCaptainHandoffMessage(ctx context.Context, conversation *model.Conversation, assistant *model.CaptainAssistant) (*model.Message, error) { + content := assistantConfigString(assistant.Config, "handoff_message") + if content == "" { + content = "Transferring to another agent for further assistance." + } + message, err := s.createCaptainOutgoingMessage(ctx, conversation, assistant, content, "") + if err != nil { + return nil, err + } + if err := s.db.WithContext(ctx).Model(&model.Conversation{}).Where("account_id = ? AND id = ? AND status = ?", conversation.AccountID, conversation.ID, string(model.ConversationStatusPending)).Update("status", string(model.ConversationStatusOpen)).Error; err != nil { + return message, fmt.Errorf("handoff captain conversation: %w", err) + } + return message, nil +} + +func messageAgentName(raw datatypes.JSON) string { + if len(raw) == 0 { + return "" + } + var attrs map[string]any + if err := json.Unmarshal(raw, &attrs); err != nil { + return "" + } + if value, ok := attrs["agent_name"].(string); ok { + return value + } + return "" +} + +func assistantConfigString(raw []byte, key string) string { + if len(raw) == 0 { + return "" + } + var cfg map[string]any + if err := json.Unmarshal(raw, &cfg); err != nil { + return "" + } + if value, ok := cfg[key].(string); ok { + return strings.TrimSpace(value) + } + return "" +} + +type captainConversationResponseBuilderJob struct { + AccountID uint `json:"account_id"` + ConversationID uint `json:"conversation_id"` + AssistantID uint `json:"assistant_id"` + MessageID uint `json:"message_id,omitempty"` +} + +var captainConversationRegistrations sync.Map + +func RegisterCaptainConversationJobs(wp *worker.WorkerPool, svc *CaptainConversationService) { + if wp == nil || svc == nil { + return + } + if _, loaded := captainConversationRegistrations.LoadOrStore(wp, struct{}{}); loaded { + return + } + wp.Register(TaskTypeCaptainConversationResponseBuilder, svc.performConversationResponseBuilderJob) +} + +func EnqueueCaptainConversationResponseForMessage(ctx context.Context, wp *worker.WorkerPool, db *gorm.DB, messageID uint) (*model.BackgroundJob, error) { + if wp == nil || db == nil || messageID == 0 || !db.Migrator().HasTable(&model.CaptainInbox{}) { + return nil, nil + } + var message model.Message + if err := db.WithContext(ctx).First(&message, messageID).Error; err != nil { + return nil, err + } + if message.MessageType != string(model.MessageTypeIncoming) || message.Private { + return nil, nil + } + var conversation model.Conversation + if err := db.WithContext(ctx).Where("account_id = ? AND id = ?", message.AccountID, message.ConversationID).First(&conversation).Error; err != nil { + return nil, err + } + if conversation.Status != string(model.ConversationStatusPending) { + return nil, nil + } + var ci model.CaptainInbox + if err := db.WithContext(ctx).Where("account_id = ? AND inbox_id = ?", conversation.AccountID, conversation.InboxID).First(&ci).Error; err != nil { + if err == gorm.ErrRecordNotFound { + return nil, nil + } + return nil, err + } + options := []worker.EnqueueOption{ + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("captain:conversation_response_builder:message:%d", message.ID)), + } + if scheduledAt := captainConversationResponseScheduledAt(ctx, db, message.ID); !scheduledAt.IsZero() { + options = append(options, worker.WithScheduledAt(scheduledAt)) + } + return wp.Enqueue(ctx, TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: conversation.AccountID, ConversationID: conversation.ID, AssistantID: ci.AssistantID, MessageID: message.ID}, options...) +} + +func captainConversationResponseScheduledAt(ctx context.Context, db *gorm.DB, messageID uint) time.Time { + if !db.Migrator().HasTable(&model.Attachment{}) { + return time.Time{} + } + var count int64 + if err := db.WithContext(ctx).Model(&model.Attachment{}).Where("message_id = ?", messageID).Count(&count).Error; err != nil || count == 0 { + return time.Time{} + } + wait := time.Second + time.Duration(minInt64(count, 4))*time.Second + return time.Now().Add(wait) +} + +func minInt64(a, b int64) int64 { + if a < b { + return a + } + return b +} + +func (s *CaptainConversationService) performConversationResponseBuilderJob(ctx context.Context, job *model.BackgroundJob) error { + var payload captainConversationResponseBuilderJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal captain conversation response builder job: %w", err) + } + if payload.AccountID == 0 || payload.ConversationID == 0 || payload.AssistantID == 0 { + return fmt.Errorf("invalid captain conversation response builder job payload: %#v", payload) + } + _, err := s.BuildConversationResponseByAccount(ctx, payload.AccountID, payload.ConversationID, payload.AssistantID) + return err +} diff --git a/internal/service/captain_conversation_worker_test.go b/internal/service/captain_conversation_worker_test.go new file mode 100644 index 00000000..6f0a9206 --- /dev/null +++ b/internal/service/captain_conversation_worker_test.go @@ -0,0 +1,140 @@ +package service + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/gochat/gochat/internal/channel" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/repository" + "github.com/gochat/gochat/internal/worker" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func setupCaptainConversationWorkerTest(t *testing.T) (*gorm.DB, *CaptainConversationService, *MessageService, *model.Account, *model.Inbox, *model.Conversation, *model.CaptainAssistant) { + t.Helper() + dbName := fmt.Sprintf("file:%s?mode=memory&cache=private", t.Name()) + db, err := gorm.Open(sqlite.Open(dbName), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.Account{}, &model.Inbox{}, &model.Contact{}, &model.Conversation{}, &model.Message{}, &model.Attachment{}, &model.CaptainAssistant{}, &model.CaptainInbox{}, &model.BackgroundJob{})) + t.Cleanup(func() { + sqlDB, _ := db.DB() + sqlDB.Close() + }) + account := &model.Account{Name: "Captain Org", Active: true} + require.NoError(t, db.Create(account).Error) + inbox := &model.Inbox{AccountID: account.ID, Name: "Web", ChannelType: "web_widget", ChannelID: 1} + require.NoError(t, db.Create(inbox).Error) + contact := &model.Contact{AccountID: account.ID, Name: "Customer", Email: "customer@example.com"} + require.NoError(t, db.Create(contact).Error) + conversation := &model.Conversation{AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, Status: string(model.ConversationStatusPending), ChannelType: inbox.ChannelType, Channel: inbox.ChannelType} + require.NoError(t, db.Create(conversation).Error) + assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Fin", Config: []byte(`{"handoff_message":"Let me connect you."}`), Status: model.AssistantStatusActive} + require.NoError(t, db.Create(assistant).Error) + require.NoError(t, db.Create(&model.CaptainInbox{AccountID: account.ID, AssistantID: assistant.ID, InboxID: inbox.ID}).Error) + conversationSvc := NewCaptainConversationService(db, nil) + messageSvc := NewMessageService(repository.NewMessageRepo(db), channel.NewDispatcher(), nil) + return db, conversationSvc, messageSvc, account, inbox, conversation, assistant +} + +func TestCaptainConversationResponseJobQueuesFromIncomingMessage(t *testing.T) { + db, conversationSvc, messageSvc, account, _, conversation, assistant := setupCaptainConversationWorkerTest(t) + conversationSvc.SetResponseBackend(&fakeCaptainConversationBackend{response: &CaptainConversationResponse{Content: "Welcome to Captain", AgentName: "Fin"}}) + wp := worker.NewWorkerPool(db) + conversationSvc.SetWorkerPool(wp) + messageSvc.SetWorkerPool(wp) + + incoming, err := messageSvc.Create(context.Background(), account.ID, 99, CreateMessageRequest{ConversationID: conversation.ID, Content: "Hello", MessageType: string(model.MessageTypeIncoming), ContentType: string(model.MessageContentTypeText)}) + require.NoError(t, err) + require.NotZero(t, incoming.ID) + + var count int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ? AND status = ?", TaskTypeCaptainConversationResponseBuilder, model.BackgroundJobStatusQueued).Count(&count).Error) + assert.Equal(t, int64(1), count) + + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var outgoing model.Message + require.NoError(t, db.Where("conversation_id = ? AND message_type = ?", conversation.ID, model.MessageTypeOutgoing).First(&outgoing).Error) + assert.Equal(t, assistant.ID, *outgoing.SenderID) + assert.Equal(t, "CaptainAssistant", outgoing.SenderType) + assert.Equal(t, "Welcome to Captain", outgoing.Content) + + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeMessageSendReply).Count(&count).Error) + assert.Equal(t, int64(1), count) +} + +func TestCaptainConversationResponseJobHandoffOpensConversation(t *testing.T) { + db, conversationSvc, _, account, _, conversation, _ := setupCaptainConversationWorkerTest(t) + conversationSvc.SetResponseBackend(&fakeCaptainConversationBackend{response: &CaptainConversationResponse{Action: "handoff"}}) + wp := worker.NewWorkerPool(db) + conversationSvc.SetWorkerPool(wp) + + _, err := wp.Enqueue(context.Background(), TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: account.ID, ConversationID: conversation.ID, AssistantID: 1, MessageID: 1}, worker.WithMaxAttempts(3)) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var updated model.Conversation + require.NoError(t, db.First(&updated, conversation.ID).Error) + assert.Equal(t, string(model.ConversationStatusOpen), updated.Status) + var outgoing model.Message + require.NoError(t, db.Where("conversation_id = ? AND message_type = ?", conversation.ID, model.MessageTypeOutgoing).First(&outgoing).Error) + assert.Equal(t, "Let me connect you.", outgoing.Content) +} + +func TestCaptainConversationResponseJobRetriesWhenProviderDisabled(t *testing.T) { + db, conversationSvc, _, account, _, conversation, _ := setupCaptainConversationWorkerTest(t) + now := time.Date(2026, 6, 6, 4, 30, 0, 0, time.UTC) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute })) + conversationSvc.SetWorkerPool(wp) + + _, err := wp.Enqueue(context.Background(), TaskTypeCaptainConversationResponseBuilder, captainConversationResponseBuilderJob{AccountID: account.ID, ConversationID: conversation.ID, AssistantID: 1, MessageID: 1}, worker.WithMaxAttempts(3)) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.Error(t, err) + assert.True(t, processed) + + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainConversationResponseBuilder).First(&job).Error) + assert.Equal(t, model.BackgroundJobStatusRetrying, job.Status) + assert.Contains(t, job.LastError, "captain conversation response generation disabled") +} + +func TestCaptainConversationResponseSkipsNonPendingConversation(t *testing.T) { + db, conversationSvc, messageSvc, account, _, conversation, _ := setupCaptainConversationWorkerTest(t) + require.NoError(t, db.Model(&model.Conversation{}).Where("id = ?", conversation.ID).Update("status", string(model.ConversationStatusOpen)).Error) + wp := worker.NewWorkerPool(db) + conversationSvc.SetWorkerPool(wp) + messageSvc.SetWorkerPool(wp) + + _, err := messageSvc.Create(context.Background(), account.ID, 99, CreateMessageRequest{ConversationID: conversation.ID, Content: "Hello", MessageType: string(model.MessageTypeIncoming), ContentType: string(model.MessageContentTypeText)}) + require.NoError(t, err) + var count int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeCaptainConversationResponseBuilder).Count(&count).Error) + assert.Equal(t, int64(0), count) +} + +type fakeCaptainConversationBackend struct { + response *CaptainConversationResponse + err error +} + +func (b *fakeCaptainConversationBackend) GenerateCaptainConversationResponse(ctx context.Context, req CaptainConversationResponseRequest) (*CaptainConversationResponse, error) { + if b.err != nil { + return nil, b.err + } + if b.response == nil { + return nil, errors.New("missing response") + } + return b.response, nil +} diff --git a/internal/service/copilot_response_worker.go b/internal/service/copilot_response_worker.go new file mode 100644 index 00000000..631f3eb6 --- /dev/null +++ b/internal/service/copilot_response_worker.go @@ -0,0 +1,46 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "sync" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" +) + +const TaskTypeCaptainCopilotResponse = "captain:copilot_response" + +type captainCopilotResponseJob struct { + AccountID uint `json:"account_id"` + UserID uint `json:"user_id"` + ConversationID uint `json:"conversation_id,omitempty"` + CopilotThreadID uint `json:"copilot_thread_id"` + MessageID uint `json:"message_id"` + Message string `json:"message,omitempty"` +} + +var copilotResponseRegistrations sync.Map + +func RegisterCopilotResponseJobs(wp *worker.WorkerPool, svc *CopilotService) { + if wp == nil || svc == nil { + return + } + if _, loaded := copilotResponseRegistrations.LoadOrStore(wp, struct{}{}); loaded { + return + } + wp.Register(TaskTypeCaptainCopilotResponse, svc.performCopilotResponseJob) +} + +func (s *CopilotService) performCopilotResponseJob(ctx context.Context, job *model.BackgroundJob) error { + var payload captainCopilotResponseJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal captain copilot response job: %w", err) + } + if payload.AccountID == 0 || payload.UserID == 0 || payload.CopilotThreadID == 0 || payload.MessageID == 0 { + return fmt.Errorf("invalid captain copilot response job payload: %#v", payload) + } + _, err := s.GenerateCopilotResponseByAccount(ctx, payload.AccountID, payload.UserID, payload.CopilotThreadID, payload.ConversationID, payload.MessageID, payload.Message) + return err +} diff --git a/internal/service/copilot_response_worker_test.go b/internal/service/copilot_response_worker_test.go new file mode 100644 index 00000000..29ceec81 --- /dev/null +++ b/internal/service/copilot_response_worker_test.go @@ -0,0 +1,113 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "testing" + "time" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/repository" + "github.com/gochat/gochat/internal/worker" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func setupCopilotResponseWorkerTest(t *testing.T) (*gorm.DB, *CopilotService, *model.Account, *model.User, *model.CaptainAssistant) { + t.Helper() + dbName := fmt.Sprintf("file:%s?mode=memory&cache=private", t.Name()) + db, err := gorm.Open(sqlite.Open(dbName), &gorm.Config{}) + require.NoError(t, err) + require.NoError(t, db.AutoMigrate(&model.Account{}, &model.User{}, &model.CaptainAssistant{}, &model.CopilotThread{}, &model.CopilotMessage{}, &model.CopilotSuggestionMessage{}, &model.BackgroundJob{})) + t.Cleanup(func() { + sqlDB, _ := db.DB() + sqlDB.Close() + }) + account := &model.Account{Name: "Copilot Org", Active: true} + require.NoError(t, db.Create(account).Error) + user := &model.User{AccountID: account.ID, Name: "Agent", Email: "agent@example.com", Password: "secret", Active: true} + require.NoError(t, db.Create(user).Error) + assistant := &model.CaptainAssistant{AccountID: account.ID, Name: "Helper", Config: json.RawMessage(`{}`), Status: model.AssistantStatusActive} + require.NoError(t, db.Create(assistant).Error) + svc := NewCopilotService(repository.NewCopilotThreadRepo(db), repository.NewCopilotMessageRepo(db), repository.NewCopilotSuggestionRepo(db), nil, repository.NewCaptainAssistantRepo(db)) + return db, svc, account, user, assistant +} + +func TestCopilotResponseJobQueuesAndPersistsAssistantReply(t *testing.T) { + db, svc, account, user, assistant := setupCopilotResponseWorkerTest(t) + svc.SetResponseBackend(&fakeCopilotResponseBackend{messages: []CopilotGeneratedMessage{{MessageType: model.CopilotMessageTypeAssistant, Message: map[string]any{"content": "Here is the answer"}}}}) + wp := worker.NewWorkerPool(db) + svc.SetWorkerPool(wp) + + thread, err := svc.CreateThread(context.Background(), account.ID, user.ID, &CreateThreadRequest{Message: "Need help", AssistantID: assistant.ID, ConversationID: 123}) + require.NoError(t, err) + require.NotNil(t, thread) + + var count int64 + require.NoError(t, db.Model(&model.CopilotMessage{}).Where("copilot_thread_id = ?", thread.ID).Count(&count).Error) + assert.Equal(t, int64(1), count) + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ? AND status = ?", TaskTypeCaptainCopilotResponse, model.BackgroundJobStatusQueued).Count(&count).Error) + assert.Equal(t, int64(1), count) + + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var messages []model.CopilotMessage + require.NoError(t, db.Where("copilot_thread_id = ?", thread.ID).Order("id ASC").Find(&messages).Error) + require.Len(t, messages, 2) + assert.Equal(t, model.CopilotMessageTypeUser, messages[0].MessageType) + assert.Equal(t, model.CopilotMessageTypeAssistant, messages[1].MessageType) + assert.Equal(t, "Here is the answer", messages[1].GetMessageContent()) +} + +func TestCopilotResponseJobUsesDisabledFallbackWithoutProvider(t *testing.T) { + db, svc, account, user, assistant := setupCopilotResponseWorkerTest(t) + wp := worker.NewWorkerPool(db) + svc.SetWorkerPool(wp) + + thread, err := svc.CreateThread(context.Background(), account.ID, user.ID, &CreateThreadRequest{Message: "Need help", AssistantID: assistant.ID}) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var assistantMsg model.CopilotMessage + require.NoError(t, db.Where("copilot_thread_id = ? AND message_type = ?", thread.ID, model.CopilotMessageTypeAssistant).First(&assistantMsg).Error) + assert.Equal(t, CopilotUnavailableMessage, assistantMsg.GetMessageContent()) +} + +func TestCopilotResponseJobRetriesBackendFailure(t *testing.T) { + db, svc, account, user, assistant := setupCopilotResponseWorkerTest(t) + svc.SetResponseBackend(&fakeCopilotResponseBackend{err: errors.New("backend down")}) + now := time.Date(2026, 6, 6, 4, 0, 0, 0, time.UTC) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute })) + svc.SetWorkerPool(wp) + + _, err := svc.CreateThread(context.Background(), account.ID, user.ID, &CreateThreadRequest{Message: "Need help", AssistantID: assistant.ID}) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.Error(t, err) + assert.True(t, processed) + + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainCopilotResponse).First(&job).Error) + assert.Equal(t, model.BackgroundJobStatusRetrying, job.Status) + assert.Contains(t, job.LastError, "backend down") +} + +type fakeCopilotResponseBackend struct { + messages []CopilotGeneratedMessage + err error +} + +func (b *fakeCopilotResponseBackend) GenerateCopilotResponse(ctx context.Context, req CopilotResponseRequest) ([]CopilotGeneratedMessage, error) { + if b.err != nil { + return nil, b.err + } + return b.messages, nil +} diff --git a/internal/service/copilot_service.go b/internal/service/copilot_service.go index 5d2a5d8a..37ee7f9d 100644 --- a/internal/service/copilot_service.go +++ b/internal/service/copilot_service.go @@ -10,17 +10,38 @@ import ( "github.com/gochat/gochat/internal/llm" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" + "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" ) // CopilotService implements business logic for Copilot (conversation assistance) operations. // Reference: Chatwoot enterprise/app/controllers/api/v1/copilot_threads_controller.rb type CopilotService struct { - threadRepo *repository.CopilotThreadRepo - messageRepo *repository.CopilotMessageRepo - suggestionRepo *repository.CopilotSuggestionRepo - assistantRepo *repository.CaptainAssistantRepo - llmProvider llm.Provider + threadRepo *repository.CopilotThreadRepo + messageRepo *repository.CopilotMessageRepo + suggestionRepo *repository.CopilotSuggestionRepo + assistantRepo *repository.CaptainAssistantRepo + llmProvider llm.Provider + responseBackend CopilotResponseBackend + worker *worker.WorkerPool +} + +type CopilotResponseBackend interface { + GenerateCopilotResponse(ctx context.Context, req CopilotResponseRequest) ([]CopilotGeneratedMessage, error) +} + +type CopilotResponseRequest struct { + AccountID uint + UserID uint + ConversationID uint + Thread *model.CopilotThread + Assistant *model.CaptainAssistant + Message string +} + +type CopilotGeneratedMessage struct { + MessageType model.CopilotMessageType + Message map[string]any } // NewCopilotService creates a new CopilotService. @@ -43,6 +64,15 @@ func NewCopilotService( return s } +func (s *CopilotService) SetWorkerPool(wp *worker.WorkerPool) { + s.worker = wp + RegisterCopilotResponseJobs(wp, s) +} + +func (s *CopilotService) SetResponseBackend(backend CopilotResponseBackend) { + s.responseBackend = backend +} + const CopilotUnavailableMessage = "You are out of Copilot credits. You can buy more credits from the billing section." // --- Thread CRUD --- @@ -89,6 +119,12 @@ func (s *CopilotService) CreateThread(ctx context.Context, accountID, userID uin if err != nil { return nil, err } + if s.worker != nil { + if err := s.enqueueCopilotResponse(ctx, accountID, userID, thread.ID, req.ConversationID, userMsg); err != nil { + return nil, err + } + return s.threadRepo.GetByAccountUserAndID(ctx, accountID, userID, thread.ID) + } if _, err := s.createAssistantReply(ctx, accountID, thread.ID, userMsg, req.ConversationID); err != nil { return nil, err } @@ -152,6 +188,13 @@ func (s *CopilotService) SendMessage(ctx context.Context, accountID, userID, thr return nil, err } + if s.worker != nil { + if err := s.enqueueCopilotResponse(ctx, accountID, userID, threadID, req.ConversationID, userMsg); err != nil { + return &SendMessageResult{UserMessage: userMsg}, err + } + return &SendMessageResult{UserMessage: userMsg}, nil + } + assistantMsg, err := s.createAssistantReply(ctx, accountID, threadID, userMsg, req.ConversationID) if err != nil { return &SendMessageResult{UserMessage: userMsg}, err @@ -161,6 +204,88 @@ func (s *CopilotService) SendMessage(ctx context.Context, accountID, userID, thr return &SendMessageResult{UserMessage: userMsg, AssistantMessage: assistantMsg}, nil } +func (s *CopilotService) enqueueCopilotResponse(ctx context.Context, accountID, userID, threadID, conversationID uint, userMsg *model.CopilotMessage) error { + if s.worker == nil || userMsg == nil { + return nil + } + _, err := s.worker.Enqueue(ctx, TaskTypeCaptainCopilotResponse, captainCopilotResponseJob{ + AccountID: accountID, + UserID: userID, + ConversationID: conversationID, + CopilotThreadID: threadID, + MessageID: userMsg.ID, + Message: userMsg.GetMessageContent(), + }, + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("captain:copilot_response:%d", userMsg.ID)), + ) + return err +} + +func (s *CopilotService) GenerateCopilotResponseByAccount(ctx context.Context, accountID, userID, threadID, conversationID, messageID uint, message string) ([]model.CopilotMessage, error) { + thread, err := s.threadRepo.GetByAccountUserAndID(ctx, accountID, userID, threadID) + if err != nil { + return nil, fmt.Errorf("thread not found: %w", err) + } + if messageID != 0 { + stored, err := s.messageRepo.GetByID(ctx, messageID) + if err != nil { + return nil, fmt.Errorf("message not found: %w", err) + } + if stored.AccountID != accountID || stored.CopilotThreadID != threadID || stored.MessageType != model.CopilotMessageTypeUser { + return nil, fmt.Errorf("message does not belong to copilot thread") + } + message = stored.GetMessageContent() + } + assistant := &thread.Assistant + if thread.AssistantID == nil || thread.Assistant.ID == 0 { + return nil, fmt.Errorf("assistant not found") + } + + generated, err := s.generateCopilotMessages(ctx, accountID, userID, conversationID, thread, assistant, message) + if err != nil { + return nil, err + } + created := make([]model.CopilotMessage, 0, len(generated)) + for _, item := range generated { + messageType := item.MessageType + if messageType == "" { + messageType = model.CopilotMessageTypeAssistant + } + raw, err := json.Marshal(item.Message) + if err != nil { + return created, fmt.Errorf("marshal copilot response message: %w", err) + } + msg := &model.CopilotMessage{AccountID: accountID, CopilotThreadID: threadID, MessageType: messageType, Message: raw} + if err := s.messageRepo.Create(ctx, msg); err != nil { + return created, fmt.Errorf("save copilot response message: %w", err) + } + reloaded, err := s.messageRepo.GetByID(ctx, msg.ID) + if err != nil { + return created, err + } + created = append(created, *reloaded) + } + return created, nil +} + +func (s *CopilotService) generateCopilotMessages(ctx context.Context, accountID, userID, conversationID uint, thread *model.CopilotThread, assistant *model.CaptainAssistant, message string) ([]CopilotGeneratedMessage, error) { + if s.responseBackend != nil { + return s.responseBackend.GenerateCopilotResponse(ctx, CopilotResponseRequest{AccountID: accountID, UserID: userID, ConversationID: conversationID, Thread: thread, Assistant: assistant, Message: message}) + } + content := CopilotUnavailableMessage + if s.llmProvider != nil { + generated, err := s.generateAssistantContent(ctx, thread, "") + if err != nil { + return nil, err + } + if strings.TrimSpace(generated) != "" { + content = generated + } + } + return []CopilotGeneratedMessage{{MessageType: model.CopilotMessageTypeAssistant, Message: map[string]any{"content": content}}}, nil +} + func (s *CopilotService) CreateThreadMessage(ctx context.Context, accountID, userID, threadID uint, req *SendMessageRequest) (*model.CopilotMessage, error) { content := strings.TrimSpace(req.Message) if content == "" { @@ -231,10 +356,12 @@ func (s *CopilotService) generateAssistantContent(ctx context.Context, thread *m Content: m.Content, }) } - llmHistory = append(llmHistory, llm.ChatMessage{ - Role: "user", - Content: content, - }) + if strings.TrimSpace(content) != "" { + llmHistory = append(llmHistory, llm.ChatMessage{ + Role: "user", + Content: content, + }) + } // Build system prompt systemPrompt := "You are an AI assistant helping a customer support agent." diff --git a/internal/service/message_service.go b/internal/service/message_service.go index e3891253..8c66aab0 100644 --- a/internal/service/message_service.go +++ b/internal/service/message_service.go @@ -212,6 +212,11 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint // Dispatch additional event based on message type if req.MessageType == "incoming" { s.dispatchMessageEvent(ctx, channel.EventMessageIncoming, message) + if s.worker != nil { + if _, err := EnqueueCaptainConversationResponseForMessage(ctx, s.worker, s.repo.DB(), message.ID); err != nil { + return message, err + } + } } else if req.MessageType == "outgoing" { if s.worker != nil { if _, err := EnqueueSendReply(ctx, s.worker, message.ID); err != nil {