diff --git a/doc/idle-voice-call-reports.md b/doc/idle-voice-call-reports.md new file mode 100644 index 00000000000..49d2287f33a --- /dev/null +++ b/doc/idle-voice-call-reports.md @@ -0,0 +1,108 @@ +# Voice call reports and idle work + +Voice report reconciliation uses the shared delivery coordinator and scheduler. +It no longer runs from the chat maintenance interval. Existing +`chat_voice_sessions` and `chat_voice_reports` rows own its durable state; this +change adds no queue table, provider endpoint, or independent timer. A concurrent +partial index supports bounded reconstruction of the existing report window. + +## Trigger and deadline + +A terminal session with a provider call ID remains eligible for report refresh +for 24 hours after `ended_at`, inclusive. A missing report is due immediately. +An existing report uses its saved `next_check_at`. The retry starts at 60 seconds +and doubles to one hour, as before. Available reports remain refreshable during +this finite window: provider availability is not a promise that transcript or +billing data can no longer change. + +Session transitions and signed callbacks notify the report worker before their +writes, inside the owning database transaction. The reply sender's `call_ended` +result does the same. Healthy active-session observations do not activate report +work. Nested transactions wait for the outer commit; rollback does not publish +new work. The existing work-signals protocol owns uncertain outcomes. + +Startup reconstructs eligible sessions in pages ordered by `(ended_at, id)`, +yielding after at most 100 rows per scope refresh. Re-registering the same owner +resets partial scan cursors. The explicit one-shot inventory still loads every +page when requested. +A partial index selects terminal sessions with provider IDs. The query uses +its ended-at range to restrict reads to the current 24-hour window. The cursor retains PostgreSQL timestamp precision. Committed report hints reload the exact changed session. Configuration, +authority and connection metadata changes do not alter report eligibility or +saved retries, so they do not wake report reconstruction. Each cached entry carries the +saved report deadline or the first millisecond after the refresh window ends, +whichever comes first. Submillisecond PostgreSQL timestamps round in the safe +direction. Expiring the window removes the cached obligation without a provider +request. Configured idle endpoints, active calls, old terminal history, and +sessions without provider IDs do not cause recurring report SQL or requests. + +## Claims, responses, and projection + +The worker locks and rechecks the session and report, then saves its attempt and +next retry before contacting the provider. Network I/O happens outside that +transaction. Settlement compares the saved attempt and retry deadline. A newer +claim or callback that moves the deadline invalidates the older response. +Incomplete or older responses cannot overwrite a retained available report. +Eight failures can mark a pending report unavailable; failure cannot downgrade +an already available report. + +Guest transcripts remain confined to the call's quarantined intake task. The +worker reads the retained report, keeps the task's source-trust metadata, and +uses the existing document revision contract. A document lock or conflicting +edit leaves the saved report and retry intact. After the lock clears, the next +successful refresh retries projection. The existing 24-hour eligibility policy +still applies; this PR does not add indefinite transcript repair or extend it. + +Provider credentials, call-report parsing, and delivery diagnostics remain in +the voice integration. The shared scheduler receives only normalized work and +deadlines. No provider-specific policy is added to app startup. + +## Lifecycle and limitations + +The shared scoped worker coalesces notifications, processes fair company batches, +and retains dirty scopes across drain or failed reads. Shutdown joins in-flight +work. Restart reconstructs the persisted retry instead of immediately repeating +the provider GET. The explicit one-shot voice reconciliation method remains for +diagnostics; production maintenance does not call it periodically. + +The supported deployment has one active server process writing voice session +and report state through its owning database client. Active-active servers and +out-of-band SQL writers are not supported by this worker. Process-local signals +do not reach another process; such a deployment needs a durable cross-process +wake contract before removing its reconciliation backstop. This PR neither adds +that transport nor asserts that startup reconstruction covers concurrent writers. This is worker quiet, not a claim of physical host sleep. Cloud alarm +coverage and conservative sleep eligibility are unchanged; a local deadline +cannot wake stopped compute. + +## Verification + +`voice-call-reports.test.ts` exercises real PostgreSQL claims, configured-idle +quiet, finite expiry, timestamp precision, retries and restart, committed writes, +rollback and nesting, drain/resume, callback races, concurrent claims, backlog +fairness, result ordering, and shutdown. The full voice integration suite checks +real hangup/callback/reply-error producers and transcript projection after a lock. +The app wiring suite protects removal of the voice maintenance lane. + +The audit and evidence ledger is +[voice call-report scheduling](plans/2026-10-25-voice-call-report-sleep-audit.md). + + +The [historical parent acceptance](plans/2026-10-25-voice-call-report-sleep-audit.md#current-parent-reconstruction-and-migration-acceptance-october-28) +records the earlier 302 distinct passing cases, the bounded startup and same-owner restart +regressions, deterministic company fairness, generated migration 0396 and the +preserved failures. The report owner keeps its finite provider refresh obligation; +this is not an unconditional idle scan or proof of deployed suspension. + + +The subsequent [configuration-scope repair](plans/2026-10-25-voice-call-report-sleep-audit.md#configuration-scope-review-repair-october-28) +records the earlier 303-case acceptance at `9bf20467df`. Five real connection saves over 205 saved +reports select zero session rows, after reproducing 1,025 selected rows before +the repair. The notification helper accepts only an exact session scope. + + +The [landed-master acceptance](plans/2026-10-25-voice-call-report-sleep-audit.md#landed-master-migration-and-acceptance-october-28) +records the current 185-case acceptance and generated migration +`0397_known_madrox.sql`, after landed cleanup migration 0396. It preserves all +existing SQL and journal entries, retains snapshots 0393 through 0397, and checks +fresh application migration plus concurrent replay and invalid-index recovery. +The older 302/303-case sections above are historical evidence, not the current +migration filename or a claim that those full groups were rerun on this head. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index a70e2892888..e51747d7775 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1617,6 +1617,38 @@ Next separate owner: voice call-report reconciliation. Preserve the existing 24-hour post-call refresh obligation, transcript ordering, task projection and provider diagnostics while removing scans with no eligible calls. Inspect completion semantics before treating a report marked available as fully settled. + +Slice 24 is published as [#17583](https://github.com/paperclipai/paperclip/pull/17583) +at `9c38ddfe53`, based on #17580 `4f0364d269`. It is attached to this chat. +CI/review are running; no merge. The next branch is +`codex/sleep-25-voice-call-reports`, with its owner audit recorded separately. + +## Slice 25: voice call reports + +Report refresh now owns a scoped deadline worker. Terminal transitions, signed +callbacks, and reply call-ended responses activate existing report/session rows. +Attempts save the next retry before provider I/O, and callback/new-attempt fences +prevent stale settlement. The existing 24-hour refresh obligation is preserved, +including available reports. Empty/configured-idle/live/old history is quiet. +This removes the last voice duty from recurring chat maintenance. + +Acceptance: 174 selected real-database/wiring cases, workspace typecheck and build +pass. A new integration case exposed missing baseRevisionId on an existing guest +transcript; the fix uses the document revision contract and preserves lock +behavior. Full failure history and external-wake limits are in +`2026-10-25-voice-call-report-sleep-audit.md`. Publication and remote checks follow. +Parent #17583 has a Discord rollback/drain test failure to repair at its owner; +parent #17580 has two new cleanup review findings. Neither is marked complete. +The remaining chat duties, original ledger, #17490 E2E failure, and final +master/stack reconciliation remain required. No stack merge is authorized. + +Slice 25 published as [#17586](https://github.com/paperclipai/paperclip/pull/17586). +Parent Discord test settlement and denial-latency repairs are propagated. At the +report tip, 41 Discord tests and 138 full voice/app tests pass; unchanged report, +cleanup and retention workers had 37 passing cases. CI/reviews remain required. +Next owner audit splits Teams unused-consent expiry from abandoned send/upload +attempt recovery, preserving one reason per PR. Company deletion cleanup policy +and the earlier E2E failure remain unresolved completion duties. ## Acceptance follow-up after the plugin parent merged (October 27) Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at @@ -1754,3 +1786,26 @@ saved deadline. Confirmed deletes also avoid a redundant reload. The [retention review ledger](2026-10-25-voice-inbound-retention-sleep-audit.md#configuration-rescan-review-repair-october-28) records 282 passing cases, types/build, failures, writer coverage and limits. Current-head review/CI and the earlier stack remain required. + + +## Report current-parent acceptance, October 28 + +Report scheduling #17586 integrates retention `68b62b6d13` and uses the existing +shared paged result contract. It yields after 100 headers per scope refresh, +resets same-owner partial scans and retains the finite 24-hour report obligation. +Officially generated migration 0396 preserves the preceding 394 SQL files and +retains the five newest snapshots. The [report acceptance ledger](2026-10-25-voice-call-report-sleep-audit.md#current-parent-reconstruction-and-migration-acceptance-october-28) +records 302 passing cases, types/build, migration/plan checks, guarded restart and +fairness probes, failures and limits. Current-head review, all CI and parent +landing still gate this slice. The full remaining stack stays in scope. + + +## Report notification review follow-up, October 28 + +Report reconstruction also received configuration hints that could not alter +its eligibility or saved retry. Five real connection saves over 205 saved +reports selected 1,025 session rows. The report owner now accepts exact session +notifications and startup; those saves select zero rows and preserve the deadline. +The [report notification ledger](2026-10-25-voice-call-report-sleep-audit.md#configuration-scope-review-repair-october-28) +records 303 passing cases, types/build, the narrower contract and limits. Fresh +exact-head review and CI are required after the repair. diff --git a/doc/plans/2026-10-25-voice-call-report-sleep-audit.md b/doc/plans/2026-10-25-voice-call-report-sleep-audit.md new file mode 100644 index 00000000000..f72b1067841 --- /dev/null +++ b/doc/plans/2026-10-25-voice-call-report-sleep-audit.md @@ -0,0 +1,446 @@ +# Voice call-report scheduling sleep audit + +Original baseline: voice history retention PR #17583 at +9c38ddfe5349ef1872306a91c7e4c5c1bc083259. +Current integrated parent: 68b62b6d13f28ac5f17f35b98adf5488752f9b97. +Branch: codex/sleep-25-voice-call-reports. +Status: implementation and local acceptance complete; publication and remote gates pending. + +## The obligation + +`voiceSessionService.reconcileReports` is the final voice duty called from the +chat maintenance cadence. It scans terminal sessions with a provider call ID and +endedAt within the last 24 hours. A missing report is immediately due; an existing +report uses nextCheckAt. The existing exponential retry is 30 * 2^attempts seconds, +capped at one hour. Its first post-fetch retry is 60 seconds. + +The current policy deliberately revisits even available reports during that +24-hour window. Do not assume available means no future obligation: later provider +reports can update transcript and billing, and delivery diagnostics may still +need their final snapshot. The provider parser's complete flag only means the +response has a report object. It is not an explicit final-and-immutable promise. +Preserve this finite refresh window while replacing the repeated global scan. +No recording URL or provider metadata is stored; the provider parser allowlists +spoken turns and safe billing/duration values. + +## Ownership and implementation direction + +Existing chat_voice_sessions and chat_voice_reports remain durable truth. Keep +selection, refresh policy, fetch, result ordering and projection within the voice +owner. Use the scoped deadline helper for coalescing, fair batches, backoff, drain +and shutdown. Register a report topic on the existing voice-change composition; +do not add caller recovery wrappers or a queue table. + +- Startup loads paged terminal-session candidates and actual nextCheckAt values. +- Missing reports are due at terminal transition. Future nextCheckAt remains + unchanged across inspection/restart. +- A report entry also has a finite window end. Expire its cached obligation when + the 24-hour window ends, even if its next attempt would be later. Define SQL + timestamp precision and inclusive boundary consistently; do not spin at cutoff. +- Lock/recheck current session eligibility and report deadline before reserving + an attempt. Save retry intent before network I/O. Provider GET remains outside + database transactions. Inspect whether a concurrent manual reconciliation or + callback can invalidate a result; use existing persisted fields for ownership + where sufficient. Do not add speculative leases or schema without evidence. +- Preserve providerUpdatedAt ordering. An incomplete or older response must not + replace a newer available report. Stale responses must still back off. +- Preserve the guest-task transcript projection, its advisory lock, low-trust + quarantine check, owner access, and activity log. A saved report is not proof + that projection completed. Preserve retry after a projection failure. +- Preserve bounded diagnostic attempts and never replay speech. Diagnostic + failure must not prevent transcript/history reconciliation. +- Keep explicit one-shot diagnostic reconciliation compatible. Remove the + recurring voice lane only after all report producers are covered. + +## Producer audit + +- `voice-inbound-service.ts` signed terminal and report callbacks call + notifyVoiceChange before the session update and report upsert. Existing reports + get nextCheckAt = now, which may advance a scheduled refresh. +- `voice-session-service.ts` close and inspect paths use withVoiceSessionChange. +- `voice-session-recovery.ts` uses full voice-change notification for terminal + inspection results. Healthy inspection and error-only updates intentionally + notify only recovery; do not make those trigger empty report scans. +- `voice-reply-push.ts` handles HTTP 409 by writing state=ended and endedAt. It + currently notifies reply, recovery and inbound explicitly; report notification + must be included before this terminal write. +- Configuration/authority/company changes use notifyVoiceChange composition. +- The existing report loop inserts and updates report rows without transaction + notifications. Move that ownership into the new report service. Account for + callbacks during an in-flight provider response and for any remaining direct + writes discovered in the full source audit. + +## Verification required + +- Configured idle, live calls, old terminal history and calls without provider IDs + are SQL/network quiet after startup. +- Real session end, signed callbacks, and reply call-ended results wake reports. +- Saved refresh deadlines survive restart; no early reads, bounded failures, and + final window expiry disarm. Available reports retain the existing window. +- Nested commit/rollback, moved-earlier callback deadlines, concurrent writes, + first work during drain, failed work entering drain, and shutdown keep work. +- Multi-page backlog and per-company fairness do not starve later calls. +- Provider failure, incomplete responses, older provider timestamps, guest + transcript projection failures and diagnostic behavior retain correct data. +- Existing full voice integration/history/access tests pass on a fresh database. + +No Cloud alarm coverage or sleep-eligibility expansion is authorized by this +migration. A future in-process refresh cannot wake stopped compute. Preserve +current external-wake limitations explicitly. + +## Parent state + +#17583 is published and attached. Its 218 selected tests, workspace types/build, +and final server typecheck pass locally; fresh CI/reviews are running. +#17580 migration-runner repair 4f0364d269 has fresh Greptile 5/5. At the last +inspection it had no failing checks and three pending checks. Its prior database +setup failure and regression/repair evidence remain in its audit. + +The complete original recurring-server ledger, older #17490 E2E failure, +remaining PR reviews and final master/stack reconciliation remain required. + +## Implementation and acceptance, 2026-10-25 + +The parent retention repair at `360e1253a0` is incorporated. The report owner now +loads scoped, paged obligations, claims each due attempt before network I/O, and +uses attempt/deadline ownership for settlement. Report writes notify the same +shared coordinator. Existing voice-change producers and the explicit reply +`call_ended` path activate it. App startup registers it once and the final voice +lane is removed from the chat interval. No schema or Cloud eligibility change. + +The saved report worker suite passed 13 tests. Initial full integration acceptance +had 136 passes and one failure: an unlocked existing transcript still could not +update because the copied projection omitted the document's base revision ID. +The repair supplies the observed `latestRevisionId`, preserving document locks +and conflict checks. A focused rerun against the already-used full-suite database +verified the document update but failed its global provider-call count because +133 retained fixture calls were eligible; this is not accepted as a clean run. +Final full acceptance uses a new database migrated through the application +migrator. Results are recorded below when terminal. + +Earlier worker fixture failures (assignedAgentId vs agentId) and the production +missing `or` import were repaired before the 13-case worker pass. Source wiring +assertions still expecting reconcileMaintenance were updated to require its +removal. Full-suite tests remain Vitest coverage; server tsconfig excludes tests, +so production typecheck is not described as test typechecking. + +Parent #17583 at `360e1253a0` has a failed server shard: the Discord rollback/drain +quiet test observes one scoped select after its nextWakeAt becomes null. The job +log is retained at `/tmp/sleep24-review-ci-failure.log`. This needs investigation +at its owning earlier slice and propagation; it is not silently retried or +claimed green. Parent #17580 also has two review findings about cleanup latency +and deletion with unconfirmed provider cleanup. These remain open. The older +#17490 E2E failure and final master reconciliation remain part of completion. + +Final local acceptance after the revision fix: 137 full voice/app wiring cases +passed on `paperclip_sleep25_projection_fixed_20261025_test`; 37 report/cleanup/ +retention worker cases passed serially on the isolated worker database. Workspace +`pnpm -r typecheck` and `pnpm build` passed. Logs: `/tmp/sleep25-voice-fixed-acceptance.log`, +`/tmp/sleep25-voice-workers-acceptance.log`, `/tmp/sleep25-workspace-types.log`, +and `/tmp/sleep25-workspace-build.log`. These 174 selected cases do not claim the +entire `pnpm test:run` suite passed locally. This host's embedded PostgreSQL test +initialization limitation remains; CI must run the complete relevant gates. + +## Publication and parent repairs + +Published as #17586. The Discord quiet-test settlement repair is propagated from +#17557 and passes all 41 command ownership tests at the report branch. The inbound +denial latency fix is also propagated; its held-provider regression is included. +After both merges, all 138 full voice/app cases pass on a fresh migrated database +(`/tmp/sleep25-denial-acceptance.log`). The earlier 37 worker cases remain valid; +those worker implementations are unchanged by the parent repairs. Remote checks +and reviews are required for the final pushed head. Parent #17580's company +deletion finding remains unresolved; denial latency finding is resolved. + + +## Paperclip review: report window index and writer scope + +Review findings 4370594150 and 4370594357 identified deployment-contract clarity +and expensive report reconstruction. The supported writer model is now explicit: +one active server process owns voice writes through its DB client. Active-active +servers and independent SQL mutation are unsupported without a separate durable +wake contract. This is a documented scope restriction, not enforcement added by +this PR, and not a claim that startup recovers concurrent foreign writers. + +The query finding reproduces on fresh PostgreSQL with 20,001 old sessions and +205 eligible sessions: an actual reconstruction query filters 20,001 historical +rows. The repair adds partial `(ended_at, id)` index 0389 and uses that order and +keyset cursor. The cursor retains PostgreSQL text timestamp precision, so sessions +100 microseconds apart do not repeat or disappear across a page. All 14 report +lifecycle cases pass after the migration; the query-plan regression checks all +three pages and each stays below 1,000 rows removed by filtering. Report scope, +24-hour policy, and provider behavior are unchanged. + +The generated migration is concurrent and idempotent and registered with the +application's top-level migration runner. It applies and a second application is +a no-op. Migration numbering/safety checks pass. The generator retained exactly +five snapshots, 0385 through 0389, pruning 0384 per repository policy. + +Retained failures: the first probe reused a database with two unrelated eligible +fixture sessions and failed its exact fixture count. It is not performance proof. +A fresh database reproduced the 20,001-row filtering problem. The first migration +invocation used unsupported top-level await in tsx eval and did not apply; its +subsequent test still failed without the index. The corrected chained invocation +applied/replayed successfully and all 14 cases passed. No assertion was relaxed. +Logs: /tmp/sleep25-report-index-before.log (fixture contamination), +/tmp/sleep25-report-index-clean-before.log (cost reproduction), +/tmp/sleep25-report-index-apply.log (invalid invocation), +/tmp/sleep25-report-index-after.log (unapplied index), +/tmp/sleep25-report-index-apply-fixed.log (application/replay), +/tmp/sleep25-report-index-fixed-tests.log (14 pass). +Workspace typecheck and build pass. Fresh full voice integration and app wiring +pass all 138 cases after the index repair. Logs: +/tmp/sleep25-report-review-{types,build,acceptance}.log. +Publish and propagate through Teams #17590/#17592 after acceptance. Fresh remote +review/checks and all prior full-stack requirements remain open. + + +## Current-parent reconstruction and migration acceptance, October 28 + +The report branch normally integrates retention `68b62b6d13` and its cleanup, +recovery and reply foundations. The retention repair prevents configuration saves +from rescanning history. Its real 205-admission, five-save regression is retained +and passes in the complete voice suite here. Both report and retention tests were +preserved when their additions conflicted at the same point in that suite. +The original 20 owned paths remain the complete report slice. Shared scheduler, +workflow and lockfile changes are inherited context, not additions by reports. + +### Bounded discovery, restart and fair dispatch + +The old report `load` executed every page before yielding. A new 105-session +fixture with saved future report deadlines selected 105 rows in its first pass, +failing the 100-row startup budget. The registered worker now returns one page +per scope refresh through the existing shared `{ rows, complete }` contract. +The domain owner retains full explicit `load` and one-shot reconciliation APIs. +The keyset cursor still stores PostgreSQL timestamp text, not a truncated Date. +Only successful, non-aborted reads advance a cursor, and registration resets it. + +The real same-owner restart case stops after 100 of 105 saved deadlines, +re-registers the service, reconstructs the original deadline, and refreshes each +of the 105 reports once at that time. It observes no SQL/provider activity before +the deadline or between saved retries afterward. Removing the cursor reset fails +at five rows instead of 100 on restart. This measures selected rows, not physical +tuple visits or instance cost. + +Fairness applies to discovered work. The prior test demanded that a company +beyond the first page run in the initial batch; that assumes an eager inventory. +The revised test puts that company deterministically on page two and requires +it in the next 25-work batch as soon as it is discovered. It also verifies all +105 calls run and the failing provider retains one saved attempt without +starving the backlog. Bypassing company rotation makes the next-batch assertion +fail. Both guarded mutations restore the exact original source afterward. + +Preserved test failures distinguish this contract change from production defects. +The first bounded full run failed the old first-batch assertion. The first +controlled-clock revision allowed random UUID order to put the other company +on page one; a fixed final key now makes page placement deterministic. An +intermediate revision expected zero active intent holds before all dirty scopes +settled. The test now uses 20 controlled one-second scheduler turns, retaining +those holds until reconciliation completes, and finishes before the provider's +60-second retry. It still checks fair dispatch, all 105 calls, one failed attempt, +and quiet after settlement. It does not advance time in every assertion retry. + +The external fixture first failed on missing tables. It now applies the actual +application migrations with the standard setup timeout before creating data. +The failed setup and 105-versus-100 budget remain recorded. No test is skipped +and no eligibility or provider assertion was removed to repair setup. + +### Generated migration lineage + +The old unmerged report migration 0393 conflicts with the landed checkpoint and +cleanup lineage. The complete retention parent migration tree and client were +restored, then the unchanged report-index schema edit was regenerated with +`pnpm db:generate`. The result is `0396_careful_blindfold.sql`. It creates the +existing proposed partial `(ended_at, id)` index concurrently and replayably, +through the application's top-level migration runner. No table or field is added. + +All 394 parent SQL files and the journal prefix are preserved byte-for-byte. +Snapshots 0392 through 0396 are the five retained snapshots. The generator's +required deletion of 0391 is included; no retained snapshot or ID was hand-edited. +Generation used parent `ffe9d47b`; retention review head `68b62b6d13` has the exact +same database tree. The snapshot/schema drift regression passes. The report +fixture boots through the application migrator, and two later complete migration +invocations succeed with the report index valid. The existing query-plan case +still tests 20,001 old sessions, 205 eligible sessions and all three precision-safe +pages; each query removes fewer than 1,000 rows by filtering. + +### Final local evidence + +302 distinct cases pass on this combined source: + +- 131 complete voice integration cases, including real hangup, callback and reply + terminal producers, transcript projection after a document lock, and the parent + retention amplification regression. This suite uses an initially empty DB. +- 170 cases across 11 complete report, retention, cleanup, recovery, company + removal, shared deadline/coordinator/batching, startup and hostname suites. + This includes 15 report, 13 retention and 14 cleanup lifecycle cases. Recovery, + company removal, retention and cleanup use separate initially empty DBs and + migrate themselves. Reports use their dedicated DB after the setup repair. +- One complete snapshot/schema drift case. The schema and migration tree are + identical to the checked generation after the final retention merge. + +Database and server typechecks/builds, migration numbering/safety and the module +boundary check pass. Earlier 174/138-case and workspace-wide claims apply only to +their stated heads. No new full local repository suite, live provider, Cloud wake, +physical suspension or savings claim is made. The 24-hour mutable-report window, +retry, confirmation, callback ownership and quarantined transcript policy remain. + +Evidence logs are `/tmp/sleep25-final-voice-consumers.log`, +`/tmp/sleep25-final-lifecycle-consumers.log`, +`/tmp/sleep25-final-snapshot-drift.log`, +`/tmp/sleep25-final-integrated-{types,build}.log`, +`/tmp/sleep25-current-migration-{check,replay}.log`, and +`/tmp/sleep25-current-migration-lineage-proof.json`. Failed setup, eager paging, +and the three fairness-clock attempts are in `/tmp/sleep25-empty-fixture-red.log`, +`/tmp/sleep25-page-budget-red.log`, `/tmp/sleep25-bounded-report-first-full.log`, +`/tmp/sleep25-bounded-report-green.log` and +`/tmp/sleep25-bounded-report-fairness-green.log`. The passing initial full report +run is `/tmp/sleep25-bounded-report-final-green.log`; its 15 cases are included +in the final 170 and not counted twice. Negative probes are +`/tmp/sleep25-{cursor-reset,company-fairness}-mutant-red.log`. + +Fresh exact-head review, CI and earlier stack landing remain required. Cleanup +parent `b8b12199b9` completed green CI after its single recorded E2E retry, but +lacks current Ponytail acceptance. Retention's earlier 4/5 review was repaired +in `68b62b6d13`; that new parent still requires its own fresh reviews and CI. +Neither parent state is a waiver for reports or for remaining stack duties. + + +## Configuration scope review repair, October 28 + +The earlier security thread correctly identified another source of repeated +report inventory. A voice connection save reached `notifyVoiceChange`, which +notified reports for connection, endpoint, company or global scopes. Bounded +paging controlled each burst but still reloaded the whole eligible set after +separate saves. A real producer regression seeds 205 terminal sessions with saved +future report deadlines, then performs five separate connection saves. It selects +1,025 session rows before repair and zero afterward. The original next deadline +is unchanged, and no provider GET occurs during the saves. + +The composed notification now signals reports only for actual session changes. +The report-specific helper accepts only an exact session scope, preventing a +future configuration caller from reintroducing broad hints through that boundary. +Startup still uses the shared worker's initial paged reconstruction. Report claims, +receipts and signed callbacks keep exact session notifications. Existing terminal +hangup and reply-ended paths remain covered through their real producer APIs. +Live recovery, replies and inbound cleanup retain their own broader authority +and configuration signals; their eligibility can depend on those changes. + +The report predicate depends on session terminal state, provider identity and +endedAt, plus saved report state. A configuration save changes none of them. +Previously such a save only reloaded nextCheckAt and did not advance a future +provider attempt; the repair retains that same saved retry. There is no new +credential policy, refresh cadence, schema or scheduler contract. Manual inventory +APIs remain complete. Unsupported external SQL still needs its own wake contract. + +303 distinct cases pass after the repair: 132 complete voice integration cases, +170 complete lifecycle/shared/startup cases, and the unchanged schema/snapshot +case. The voice suite uses a fresh separate DB; the lifecycle suites use their +existing dedicated isolated DBs. Server typecheck/build and module boundaries +pass. The earlier 302 count applies to `80207ba6bb` and is not added to this count. +The current Ponytail 5/5 on that earlier head cannot be reused after this edit; +fresh exact-head review and all CI remain required. + +Logs: `/tmp/sleep25-config-amplification-red.log`, +`/tmp/sleep25-scope-final-voice-consumers.log`, +`/tmp/sleep25-scope-final-lifecycle-consumers.log`, +`/tmp/sleep25-scope-final-types.log`, `/tmp/sleep25-scope-final-build.log`, and +`/tmp/sleep25-scope-final-module-boundaries.log`. No full local repository suite, +live provider, Cloud wake or physical suspension claim is added. The original +20-path report scope and the existing finite 24-hour obligation remain intact. + +## Landed-master migration and acceptance — October 28 + +The normal merge uses actual master `980775e0e956eecdc28c73c9b4aca0407ad45e1a` +after cleanup and retention landed. The reviewed report head was +`9bf20467dfd50b81848e1c8dd7ddc905d206418f`, based on retention `68b62b6d13`. +An explicit-base composition preserved the complete non-database report patch +before the acceptance-test repair below. Only the shared client, journal, and +396 snapshot conflicted. Those metadata files were restored from actual master +before invoking the official `pnpm db:generate` workflow. + +The generated report migration is now `0397_known_madrox.sql`. It retains the +reviewed partial `(ended_at, id)` concurrent, idempotent index SQL byte-for-byte. +The existing top-level migration registry adds only the new filename and keeps +landed cleanup migration 396 registered. All 395 existing SQL files and the +complete journal prefix match actual master byte-for-byte. Snapshots 393 through +397 remain, with the generated 397 snapshot pointing to actual parent 396. +No landed migration, snapshot contents, or journal entry was renumbered. + +A new empty PostgreSQL database boots through the application migrator with a +valid report index. A deliberately failed unique concurrent index build under +the report index name leaves an invalid index; two concurrent invocations of +the application migrator repair it, restore the intended report index, and +record one migration history row. Another replay succeeds. This exercises the +new 397 registration and invalid-index recovery rather than assuming that a +successful Drizzle generation proves executable migration acceptance. + +The first complete four-suite run failed: 183 passed and one report integration +case failed, 184 total. `retains one safe report per call and rejects history +after access revocation` expected cost 240000 but retained 120000. The test used +a global one-shot transport response while reconciliation correctly visits +other due calls. An unrelated call could consume the target's newer response. +Adding an explicit earlier-ended competing call reproduced the same assertion +in isolation on unchanged production. The test now selects responses by the +target provider session ID, asserts the target accepts 240000, asserts the +other call independently keeps 120000, and then proves stale cost 1 cannot +overwrite the target. Existing access-revocation and one-report-per-call +assertions remain. No production report or provider behavior changed. + +The isolated repaired case passes, followed by the complete 132-case voice +integration suite on a new empty database. The unchanged complete report (15), +cleanup (24), and retention (13) suites passed in the first combined run, each +on its separate empty database. These 184 distinct consumer cases plus one +complete schema/snapshot drift case yield 185 distinct passing cases for this +integration. The isolated case is not added again. Database build, server +build, final server typecheck, migration safety/numbering and module-boundary +checks pass. The earlier 303-case evidence belongs to the stated previous head; +it is not added to this acceptance count or claimed rerun here. + +Logs: `/tmp/sleep25-landed-parent-db-generate.log`, +`/tmp/sleep25-landed-parent-db-build.log`, +`/tmp/sleep25-landed-parent-real-migration.log`, +`/tmp/sleep25-landed-parent-real-consumers.log` (the retained 183-pass/one-fail run), +`/tmp/sleep25-report-mock-isolation-red.log`, +`/tmp/sleep25-report-mock-isolation-green.log`, +`/tmp/sleep25-landed-parent-final-voice.log`, +`/tmp/sleep25-landed-parent-snapshot.log`, +`/tmp/sleep25-landed-parent-final-types.log`, +`/tmp/sleep25-landed-parent-server-build.log`, and +`/tmp/sleep25-landed-parent-module-boundaries.log`. + +Fresh current-head CI and Greptile 5/5 remain required. Prior-head review scores +are stale after this normal merge. A stopped or paused Ponytail turn without a +verdict does not block under the user's authorization; an actual failing +verdict must be fixed. No new full local repository suite, live-provider test, +Cloud wake, host/database suspension, or cost-saving result is claimed. + + +## Review documentation and preserved remote timeout — October 28 + +Ponytail reviewed `c28de398` at 4/5 advisory, found no must-fix production defect, +and said to ship after correcting the design document's stale migration +reference. The design document now links the current 185-case acceptance and +migration `0397_known_madrox.sql`; the older 302/303-case paragraphs are marked +historical. Its two remaining suggestions are explicitly optional: one-shot +diagnostic `reconcile(limit)` still inventories the whole eligible window before +slicing, and some existing test helpers advance fake time during assertion +retries. They are not silently claimed fixed or treated as current idle polling. + +CI run `40703490340`, server shard 8/12 job `122869422044`, failed one unchanged +approval-queue test: `resets consecutive failure backoff after a successful batch +with remaining debt` exceeded its existing 15-second timeout. Teardown then +reported a company foreign-key violation from activity-log work. The shard had +2,097 passed tests, one failed and one skipped. No report assertion failed in +that job, and the exact CI timing cause is not proven. The complete unchanged +28-case approval-queue suite passes on a new isolated PostgreSQL database. Its +test, worker, coordinator and transaction-signals source were checked byte-for- +byte against `c28de398` before that probe, which ran from the local child checkout. +No timeout, workflow, approval logic or production source was changed. The old +CI failure remains recorded and the new documentation head requires fresh CI; +no blind rerun of the failed old job or waiver is claimed. + +Evidence: `/tmp/sleep25-c28-server8-job.log`, +`/tmp/sleep25-approval-timeout-local-probe.log` and its exit status 0. The 28 cases +are a separate investigation result, not added to the 185-case report acceptance. +Prior-head review scores are stale after this documentation commit. diff --git a/doc/plans/2026-10-25-voice-drain-repair.md b/doc/plans/2026-10-25-voice-drain-repair.md index da0ae09914d..dbe0754408e 100644 --- a/doc/plans/2026-10-25-voice-drain-repair.md +++ b/doc/plans/2026-10-25-voice-drain-repair.md @@ -60,3 +60,13 @@ owner test now advances the actual retained nextWakeAt, waits for row removal, and checks disarm. Before resume it also proves a stable idle snapshot through a minute of quiet. All 11 retention cases pass in /tmp/sleep24-drain-tests.log. No production scheduling deadline or dispatch behavior changed for this repair. + +## Report resume assertion + +The report test had the same active-counter-only flush assumption. It now checks +idle snapshot stability while paused, advances the retained nextWakeAt on resume, +and verifies one fetch and the available durable report before testing rollback. +All 14 report cases pass in /tmp/sleep25-drain-tests.log. An intermediate new +assertion incorrectly expected state=completed; the schema uses status=available. +That test-authoring failure is retained in /tmp/sleep25-drain-invalid-assertion.log. +The scheduler repair and business deadlines were not changed to satisfy tests. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index bb2b6797ec6..9f583812d56 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -548,6 +548,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0393_open_grandmaster.sql", "0395_tranquil_firestar.sql", "0396_tan_paladin.sql", + "0397_known_madrox.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0397_known_madrox.sql b/packages/db/src/migrations/0397_known_madrox.sql new file mode 100644 index 00000000000..df744394402 --- /dev/null +++ b/packages/db/src/migrations/0397_known_madrox.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_voice_sessions_report_window_idx" ON "chat_voice_sessions" USING btree ("ended_at","id") WHERE "chat_voice_sessions"."state" in ('ended', 'failed', 'expired') and "chat_voice_sessions"."provider_session_id" is not null; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0392_snapshot.json b/packages/db/src/migrations/meta/0397_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0392_snapshot.json rename to packages/db/src/migrations/meta/0397_snapshot.json index 0f1a1921bbb..648ee0ff654 100644 --- a/packages/db/src/migrations/meta/0392_snapshot.json +++ b/packages/db/src/migrations/meta/0397_snapshot.json @@ -1,6 +1,6 @@ { - "id": "6974dd25-b8e9-4143-b4b7-ee5a5c3ff54a", - "prevId": "e080fc1b-af8e-4012-8ce5-3b5065b65143", + "id": "c7947622-f079-44c9-a57a-5a11d1bec5e8", + "prevId": "be87a5d8-4164-4d95-89ae-b895701fa21b", "version": "7", "dialect": "postgresql", "tables": { @@ -1350,6 +1350,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "activity_log_company_user_issue_actor_idx": { + "name": "activity_log_company_user_issue_actor_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "actor_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "entity_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"activity_log\".\"actor_type\" = 'user' AND \"activity_log\".\"entity_type\" = 'issue'", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -14160,6 +14188,24 @@ "primaryKey": false, "notNull": false }, + "provider_account_id": { + "name": "provider_account_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "cleanup_confirmed_at": { + "name": "cleanup_confirmed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "cleanup_next_attempt_at": { + "name": "cleanup_next_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "expires_at": { "name": "expires_at", "type": "timestamp with time zone", @@ -14182,6 +14228,22 @@ } }, "indexes": { + "chat_voice_inbound_cleanup_idx": { + "name": "chat_voice_inbound_cleanup_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_voice_inbound_expiry_idx": { "name": "chat_voice_inbound_expiry_idx", "columns": [ @@ -14934,6 +14996,28 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_voice_sessions_report_window_idx": { + "name": "chat_voice_sessions_report_window_idx", + "columns": [ + { + "expression": "ended_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -41945,6 +42029,28 @@ } }, "indexes": { + "join_requests_pending_retention_idx": { + "name": "join_requests_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"join_requests\".\"status\" = 'pending_approval'", + "concurrently": false, + "method": "btree", + "with": {} + }, "join_requests_invite_unique_idx": { "name": "join_requests_invite_unique_idx", "columns": [ @@ -42965,6 +43071,99 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.native_runner_checkpoints": { + "name": "native_runner_checkpoints", + "schema": "", + "columns": { + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "issue_id": { + "name": "issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "part": { + "name": "part", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "material": { + "name": "material", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "native_runner_checkpoints_headers_idx": { + "name": "native_runner_checkpoints_headers_idx", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_runner_checkpoints\".\"part\" = 'header'", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk": { + "name": "native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk", + "tableFrom": "native_runner_checkpoints", + "tableTo": "heartbeat_runs", + "columnsFrom": [ + "company_id", + "issue_id", + "run_id" + ], + "columnsTo": [ + "company_id", + "native_issue_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "native_runner_checkpoints_run_id_part_pk": { + "name": "native_runner_checkpoints_run_id_part_pk", + "columns": [ + "run_id", + "part" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.pipeline_automation_executions": { "name": "pipeline_automation_executions", "schema": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 41df4aca3f7..b402ee53314 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2766,6 +2766,13 @@ "when": 1793191988458, "tag": "0396_tan_paladin", "breakpoints": true + }, + { + "idx": 397, + "version": "7", + "when": 1793195976732, + "tag": "0397_known_madrox", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/chat_voice_sessions.ts b/packages/db/src/schema/chat_voice_sessions.ts index 6a652f97696..ac3dfcfe54e 100644 --- a/packages/db/src/schema/chat_voice_sessions.ts +++ b/packages/db/src/schema/chat_voice_sessions.ts @@ -36,6 +36,8 @@ export const chatVoiceSessions = pgTable("chat_voice_sessions", { uniqueIndex("chat_voice_sessions_request_uq").on(t.companyId, t.callerId, t.idempotencyKey), uniqueIndex("chat_voice_sessions_live_caller_task_uq").on(t.companyId, t.callerId, t.issueId).where(sql`${t.state} not in ('ended', 'failed', 'expired')`), index("chat_voice_sessions_expiry_idx").on(t.state, t.expiresAt), + index("chat_voice_sessions_report_window_idx").on(t.endedAt, t.id) + .where(sql`${t.state} in ('ended', 'failed', 'expired') and ${t.providerSessionId} is not null`), foreignKey({ columns: [t.companyId, t.endpointId], foreignColumns: [chatEndpoints.companyId, chatEndpoints.id], name: "chat_voice_sessions_endpoint_fk" }), foreignKey({ columns: [t.companyId, t.conversationId], foreignColumns: [chatConversations.companyId, chatConversations.id], name: "chat_voice_sessions_conversation_fk" }), foreignKey({ columns: [t.companyId, t.issueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_sessions_issue_fk" }), diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 8e7fbd20d38..bc4f3461e03 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -25,7 +25,6 @@ describe("createChatReconciliationCoordinator", () => { processFailedGitHubWebhookDeliveries, processPendingDeliveries, flushPublications, - reconcileVoiceSessions: async () => undefined, projectRunMilestones: async () => 0, processPendingSlackFileUploadReceipts: async () => undefined, processPendingSlackSessionSyncs: async () => undefined, @@ -74,11 +73,11 @@ describe("createChatReconciliationCoordinator", () => { }, ); - it("keeps receiver inventory out of periodic chat maintenance while preserving voice", () => { + it("keeps receiver and voice inventory out of periodic chat maintenance", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals")); expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()"); - expect(maintenance).toContain("chatChannels.voice.reconcileMaintenance()"); + expect(maintenance).not.toContain("chatChannels.voice."); expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); }); @@ -88,7 +87,8 @@ describe("createChatReconciliationCoordinator", () => { expect(source).toContain("chatChannels.voice.registerReplyWork(deliveryWork,"); expect(source).toContain("chatChannels.voice.inbound.registerCleanupWork(deliveryWork,"); expect(source).toContain("chatChannels.voice.inbound.registerRetentionWork(deliveryWork,"); - expect(source).toContain("chatChannels.voice.reconcileMaintenance()"); + expect(source).toContain("chatChannels.voice.registerReportWork(deliveryWork,"); + expect(source).not.toContain("reconcileVoiceSessions"); expect(source).not.toContain("chatChannels.voice.reconcile()"); }); @@ -153,7 +153,6 @@ describe("createChatReconciliationCoordinator", () => { const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ projectRunMilestones, - reconcileVoiceSessions: async () => undefined, processPendingDeliveries: async () => undefined, flushPublications, processPendingSlackFileUploadReceipts: async () => undefined, @@ -216,7 +215,6 @@ describe("createChatReconciliationCoordinator", () => { const coordinator = createChatReconciliationCoordinator({ projectRunMilestones, flushPublications, - reconcileVoiceSessions: async () => undefined, processPendingDeliveries: async () => undefined, processPendingSlackFileUploadReceipts: async () => undefined, processPendingSlackSessionSyncs: async () => undefined, @@ -256,7 +254,6 @@ describe("createChatReconciliationCoordinator", () => { const slackStatusReleased = new Promise((resolve) => { releaseSlackStatus = resolve; }); - const reconcileVoiceSessions = vi.fn(async () => undefined); const processPendingDeliveries = vi.fn(async () => deliveryReleased); const flushPublications = vi.fn(async () => undefined); const processPendingSlackFileUploadReceipts = vi.fn(async () => undefined); @@ -266,7 +263,6 @@ describe("createChatReconciliationCoordinator", () => { const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ projectRunMilestones: async () => 0, - reconcileVoiceSessions, processPendingDeliveries, flushPublications, processPendingSlackFileUploadReceipts, @@ -276,7 +272,6 @@ describe("createChatReconciliationCoordinator", () => { coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileVoiceSessions).toHaveBeenCalledTimes(1); expect(processPendingDeliveries).toHaveBeenCalledTimes(1); expect(flushPublications).toHaveBeenCalledTimes(1); expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(1); @@ -284,7 +279,6 @@ describe("createChatReconciliationCoordinator", () => { }); coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2); expect(flushPublications).toHaveBeenCalledTimes(2); expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2); }); @@ -311,14 +305,12 @@ describe("createChatReconciliationCoordinator", () => { await receiptReleased; if (rejects) throw lookupError; }); - const reconcileVoiceSessions = vi.fn(async () => undefined); const processPendingDeliveries = vi.fn(async () => undefined); const flushPublications = vi.fn(async () => undefined); const processPendingSlackSessionSyncs = vi.fn(async () => undefined); const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ projectRunMilestones: async () => 0, - reconcileVoiceSessions, processPendingDeliveries, flushPublications, processPendingSlackFileUploadReceipts, @@ -336,7 +328,6 @@ describe("createChatReconciliationCoordinator", () => { }); coordinator.reconcile(); await vi.waitFor(() => { - expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2); expect(processPendingDeliveries).toHaveBeenCalledTimes(2); expect(flushPublications).toHaveBeenCalledTimes(2); expect(processPendingSlackSessionSyncs).toHaveBeenCalledTimes(2); diff --git a/server/src/__tests__/voice-call-reports.test.ts b/server/src/__tests__/voice-call-reports.test.ts new file mode 100644 index 00000000000..fa614a5ed8a --- /dev/null +++ b/server/src/__tests__/voice-call-reports.test.ts @@ -0,0 +1,309 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, agents, chatConversations, chatEndpoints, chatVoiceSessions, chatVoiceReports, companies, createDb, issues, toolApplications, toolConnections } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { voiceCallReports } from "../services/voice/voice-call-reports.js"; +import type { SpekoCallReport } from "../services/voice/speko-call-report.js"; +import { notifyVoiceReports } from "../services/voice/voice-report-notifications.js"; +import { withVoiceSessionChange } from "../services/voice/voice-recovery-notifications.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_REPORT_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("voice call-report lifecycle", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string, endpointId: string, agentId: string, issueId: string, conversationId: string; + let coordinator: ReturnType | undefined; + let enabled = true; + const errors: unknown[] = []; + const fixtureCompanies: string[] = []; + const day = 86_400_000; + const result = (): SpekoCallReport => ({ complete: true, transcript: [], costMicroUsd: "100", durationSeconds: 10, providerUpdatedAt: new Date() }); + const fetchReport = vi.fn(async (_session: typeof chatVoiceSessions.$inferSelect) => result()); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("voice-call-reports-"); + if (external) await applyPendingMigrations(external); + db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + async function fixture() { + const companyId = randomUUID(), endpointId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), conversationId = randomUUID(); + fixtureCompanies.push(companyId); + const applicationId = randomUUID(), connectionId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Voice recovery", issuePrefix: `V${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Voice", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Voice", assigneeAgentId: agentId, status: "in_progress" }); + await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Voice", type: "chat" }); + await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Voice", uid: connectionId, transport: "voice", connectionPurpose: "channel", enabled: true, status: "active" }); + await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, assignedAgentId: agentId, publicId: endpointId, provider: "speko", status: "active", setup: { step: "complete", runtimeGeneration: 1 } }); + await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId, externalConversationId: conversationId, externalThreadId: conversationId, externalLabel: "Voice", isDirectMessage: true }); + return { companyId, endpointId, assignedAgentId: agentId, issueId, conversationId }; + } + beforeEach(async () => { + enabled = true; errors.length = 0; fetchReport.mockReset().mockImplementation(async () => result()); + ({ companyId, endpointId, assignedAgentId: agentId, issueId, conversationId } = await fixture()); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); + for (const companyId of fixtureCompanies.splice(0)) { + await db.delete(chatVoiceSessions).where(eq(chatVoiceSessions.companyId, companyId)); + await db.delete(chatConversations).where(eq(chatConversations.companyId, companyId)); + await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)); + await db.delete(issues).where(eq(issues.companyId, companyId)); + await db.delete(agents).where(eq(agents.companyId, companyId)); + await db.delete(companies).where(eq(companies.id, companyId)); + } + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function session(patch: Partial = {}, writer = db) { + const id = patch.id ?? randomUUID(); + return withVoiceSessionChange(writer, id, tx => tx.insert(chatVoiceSessions).values({ id, companyId, endpointId, conversationId, issueId, assignedAgentId: agentId, + callerId: randomUUID(), callerAuthority: "member", mode: "browser", state: "ended", endedAt: new Date(), generation: 1, credentialFingerprint: "test", providerSessionId: randomUUID(), + toolTokenHash: "a".repeat(64), idempotencyKey: id, requestFingerprint: id, createdAt: new Date(Date.now() - 60_000), updatedAt: new Date(Date.now() - 60_000), expiresAt: new Date(Date.now() + 600_000), ...patch }).returning()).then(rows => rows[0]); + } + async function start(service = voiceCallReports(db, fetchReport)) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + await service.register(coordinator, () => enabled).ready; + return service; + } + async function flush() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 }); + } + async function quiet(ms = 10_000) { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + const requests = fetchReport.mock.calls.length; + try { await vi.advanceTimersByTimeAsync(ms); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(fetchReport).toHaveBeenCalledTimes(requests); } + finally { reads.mockRestore(); writes.mockRestore(); } + } + async function report(id: string) { return (await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.sessionId, id)))[0]; } + async function due(id: string) { + await db.transaction(async tx => { + await notifyVoiceReports(tx, { kind: "session", id }); + await tx.update(chatVoiceReports).set({ nextCheckAt: new Date(0) }).where(eq(chatVoiceReports.sessionId, id)); + }); + } + it("bounds report reconstruction and resets a partial scan when the same owner restarts", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + const deadline = Date.now() + 120_000; + const ids: string[] = []; + for (let index = 0; index < 105; index++) { + const call = await session({ endedAt: new Date(Date.now() - 60_000) }); ids.push(call.id); + await db.insert(chatVoiceReports).values({ sessionId: call.id, companyId, nextCheckAt: new Date(deadline) }); + } + const counts: number[] = [], select = db.select.bind(db); + const reads = vi.spyOn(db, "select").mockImplementation((...args) => { + const selection = select(...args), from = selection.from.bind(selection); + vi.spyOn(selection, "from").mockImplementation(table => { + const query = from(table); + if (table === chatVoiceSessions) { + const execute = query.execute.bind(query); + vi.spyOn(query, "execute").mockImplementation(async () => { + const rows = await execute(); counts.push(rows.length); return rows; + }); + } + return query; + }); + return selection; + }); + try { + const service = await start(); + expect(counts.reduce((sum, count) => sum + count, 0)).toBe(100); + await coordinator!.stop(); coordinator = undefined; + const beforeRestart = counts.length; + await start(service); + expect(counts.slice(beforeRestart).reduce((sum, count) => sum + count, 0)).toBe(100); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(deadline), { interval: 0 }); + expect(Math.max(...counts)).toBeLessThanOrEqual(100); reads.mockRestore(); + await quiet(); expect(fetchReport).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(deadline - Date.now()); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(fetchReport).toHaveBeenCalledTimes(105); + }, { timeout: 5000 }); + expect(fetchReport.mock.calls.map(([call]) => call.id).sort()).toEqual(ids.sort()); + expect((await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.companyId, companyId))) + .every(row => row.attempts === 1)).toBe(true); + await quiet(); + } finally { reads.mockRestore(); } + }); + it("leaves configured idle, live calls, old history and unidentified calls quiet", async () => { + await session({ state: "active", endedAt: null }); + await session({ endedAt: new Date(Date.now() - day - 1000) }); + await session({ providerSessionId: null }); + await start(); await flush(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(120_000); + expect(fetchReport).not.toHaveBeenCalled(); + }); + it("wakes on committed termination and preserves refresh of available reports", async () => { + await start(); await flush(); await quiet(); + const call = await session(); await flush(); + expect(fetchReport).toHaveBeenCalledOnce(); expect((await report(call.id)).status).toBe("available"); + const next = (await report(call.id)).nextCheckAt.getTime(); + expect(coordinator!.nextWakeAt()).toBe(next); await quiet(); + await vi.advanceTimersByTimeAsync(next - Date.now() + 1000); await flush(); + expect(fetchReport).toHaveBeenCalledTimes(2); expect((await report(call.id)).attempts).toBe(2); + }); + it("reconstructs persisted retries after restart without early provider reads", async () => { + const call = await session(); fetchReport.mockRejectedValueOnce(new Error("provider unavailable")); + await start(); await flush(); const saved = await report(call.id); + expect(saved.attempts).toBe(1); expect(saved.status).toBe("pending"); + await coordinator!.stop(); await start(); await flush(); + expect(coordinator!.nextWakeAt()).toBe(saved.nextCheckAt.getTime()); await quiet(); + await vi.advanceTimersByTimeAsync(saved.nextCheckAt.getTime() - Date.now() + 1000); await flush(); + expect(fetchReport).toHaveBeenCalledTimes(2); expect((await report(call.id)).status).toBe("available"); + }); + it("disarms at the end of the finite window even when the next check is later", async () => { + const endedAt = new Date(Date.now() - day + 25_000), call = await session({ endedAt }); + await start(); await flush(); + expect(coordinator!.nextWakeAt()).toBe(endedAt.getTime() + day + 1); + expect((await report(call.id)).nextCheckAt.getTime()).toBeGreaterThan(coordinator!.nextWakeAt()!); + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(120_000); expect(fetchReport).toHaveBeenCalledOnce(); + await coordinator!.stop(); await start(); await flush(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("rounds the inclusive PostgreSQL window end up without another provider fetch", async () => { + const call = await session(), end = Math.trunc(Date.now()) + 60_000; + await db.execute(sql`UPDATE chat_voice_sessions SET ended_at = to_timestamp(${end - day}::double precision / 1000) + interval '0.0005 seconds' WHERE id = ${call.id}`); + await db.insert(chatVoiceReports).values({ companyId, sessionId: call.id, status: "available", nextCheckAt: new Date(end + 60_000) }); + const service = voiceCallReports(db, fetchReport); + expect((await service.load())[0].dueAt).toBe(end + 1); + vi.setSystemTime(end + 1); + expect(await service.check(call.id)).toBeNull(); expect(fetchReport).not.toHaveBeenCalled(); + }); + it("marks repeated missing reports unavailable but never downgrades retained available data", async () => { + const available = await session(), missing = await session(), service = voiceCallReports(db, fetchReport); + await service.check(available.id); + await db.update(chatVoiceReports).set({ attempts: 7, nextCheckAt: new Date(0) }).where(eq(chatVoiceReports.sessionId, available.id)); + await db.insert(chatVoiceReports).values({ companyId, sessionId: missing.id, attempts: 7, nextCheckAt: new Date(0) }); + fetchReport.mockRejectedValue(new Error("provider unavailable")); + await service.check(available.id); await service.check(missing.id); + expect(await report(available.id)).toMatchObject({ attempts: 8, status: "available", costMicroUsd: "100" }); + expect(await report(missing.id)).toMatchObject({ attempts: 8, status: "unavailable" }); + expect((await report(missing.id)).nextCheckAt.getTime()).toBeGreaterThan(Date.now()); + }); + it("retains committed work during drain and ignores rolled-back nested writes", async () => { + await start(); await flush(); enabled = false; + const call = await session(); await flush(); + const drained = idleWorkSnapshot(); + await quiet(60_000); expect(idleWorkSnapshot()).toEqual(drained); expect(await report(call.id)).toBeUndefined(); + const resumeAt = coordinator!.nextWakeAt(); expect(resumeAt).not.toBeNull(); + enabled = true; await vi.advanceTimersByTimeAsync(Math.max(0, resumeAt! - Date.now())); + await vi.waitFor(async () => { await flush(); expect(fetchReport).toHaveBeenCalledOnce(); }); + expect(await report(call.id)).toMatchObject({ status: "available" }); + await expect(db.transaction(async tx => { await session({}, tx as unknown as typeof db); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await flush(); expect(fetchReport).toHaveBeenCalledOnce(); await quiet(); + }); + it("does not fetch an uncommitted nested termination", async () => { + await start(); await flush(); + await db.transaction(async tx => { + await session({}, tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(5000); expect(fetchReport).not.toHaveBeenCalled(); + }); + await flush(); expect(fetchReport).toHaveBeenCalledOnce(); + }); + it("keeps a callback's earlier deadline when an older provider response finishes", async () => { + const call = await session(), service = voiceCallReports(db, fetchReport); + let release!: (value: SpekoCallReport) => void; + fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + const checking = service.check(call.id); + await vi.waitFor(() => expect(fetchReport).toHaveBeenCalledOnce()); + await due(call.id); + release({ ...result(), costMicroUsd: "1" }); await checking; + expect((await report(call.id)).nextCheckAt.getTime()).toBe(0); + expect((await report(call.id)).costMicroUsd).toBeNull(); + await service.check(call.id); expect((await report(call.id)).costMicroUsd).toBe("100"); + }); + it("does not regress an available report from incomplete or older responses", async () => { + const call = await session(), service = voiceCallReports(db, fetchReport); + const latest = new Date(Date.now() + 10000); + fetchReport.mockResolvedValueOnce({ ...result(), providerUpdatedAt: latest, costMicroUsd: "200" }); + await service.check(call.id); + for (const response of [{ ...result(), complete: false, providerUpdatedAt: new Date(latest.getTime() + 1), costMicroUsd: "1" }, + { ...result(), providerUpdatedAt: new Date(0), costMicroUsd: "2" }]) { + await due(call.id); fetchReport.mockResolvedValueOnce(response); await service.check(call.id); + expect(await report(call.id)).toMatchObject({ status: "available", costMicroUsd: "200" }); + } + }); + it("allows only one concurrent due claim and saves its retry before provider contact", async () => { + const call = await session(), first = voiceCallReports(db, fetchReport), second = voiceCallReports(db, fetchReport); + let release!: (value: SpekoCallReport) => void; + fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + const checking = first.check(call.id); + await vi.waitFor(() => expect(fetchReport).toHaveBeenCalledOnce()); + const saved = await report(call.id); expect(saved.attempts).toBe(1); expect(saved.nextCheckAt.getTime()).toBeGreaterThan(Date.now()); + await second.check(call.id); expect(fetchReport).toHaveBeenCalledOnce(); + release(result()); await checking; + }); + it("pages and fairly schedules a backlog with one failing provider", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + const broken = await session(); for (let i = 0; i < 103; i++) await session(); + const other = await fixture(); const last = await session({ ...other, id: "ffffffff-ffff-ffff-ffff-ffffffffffff" }); + fetchReport.mockImplementation(async call => { if (call.id === broken.id) throw new Error("unavailable"); return result(); }); + await start(); expect(fetchReport).toHaveBeenCalledTimes(25); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(fetchReport).toHaveBeenCalledTimes(50), { interval: 0 }); + expect(fetchReport.mock.calls.slice(25, 50).some(([call]) => call.id === last.id)).toBe(true); + for (let pass = 0; pass < 20; pass++) { + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt() !== null || idleWorkSnapshot().active === 0).toBe(true), { interval: 0 }); + } + expect(fetchReport).toHaveBeenCalledTimes(105); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { interval: 0 }); + expect((await report(broken.id)).attempts).toBe(1); await quiet(); + }); + it("reconstructs recent reports without scanning old history and preserves submillisecond page boundaries", async () => { + const seed = await session({ endedAt: new Date(Date.now() - 2 * day) }); + await db.execute(sql`insert into chat_voice_sessions + (company_id, endpoint_id, conversation_id, issue_id, assigned_agent_id, caller_id, + caller_authority, mode, state, generation, credential_fingerprint, provider_session_id, + tool_token_hash, idempotency_key, request_fingerprint, expires_at, ended_at, created_at) + select company_id, endpoint_id, conversation_id, issue_id, assigned_agent_id, gen_random_uuid()::text, + caller_authority, mode, state, generation, credential_fingerprint, gen_random_uuid()::text, + tool_token_hash, gen_random_uuid()::text, request_fingerprint, expires_at, ended_at, created_at + from chat_voice_sessions cross join generate_series(1, 20000) where id = ${seed.id}`); + const stamp = new Date(Date.now() - 60_000).toISOString(); + const expected: string[] = []; + for (let i = 0; i < 205; i++) { + const row = await session(); expected.push(row.id); + await db.execute(sql`update chat_voice_sessions set ended_at = ${stamp}::timestamptz + (${i % 3} * interval '0.0001 second') where id = ${row.id}`); + } + await db.execute(sql`analyze chat_voice_sessions`); + const queries: Array<{ text: string; parameters: unknown[] }> = []; + const original = db.$client.unsafe.bind(db.$client); + const capture = vi.spyOn(db.$client, "unsafe").mockImplementation(((text: string, parameters: unknown[] = [], ...rest: unknown[]) => { + if (text.includes('from "chat_voice_sessions"') && text.includes('left join "chat_voice_reports"')) queries.push({ text, parameters }); + return original(text, parameters as never, ...rest as []); + }) as typeof db.$client.unsafe); + let loaded: Awaited["load"]>>; + try { loaded = await voiceCallReports(db, fetchReport).load(); } + finally { capture.mockRestore(); } + expect(loaded!.map(row => row.id).sort()).toEqual(expected.sort()); + expect(queries).toHaveLength(3); + function removed(node: Record): number { + return Math.max(Number(node["Rows Removed by Filter"] ?? 0), + ...(Array.isArray(node.Plans) ? node.Plans.map(child => removed(child)) : [])); + } + for (const query of queries) { + const [row] = await db.$client.unsafe(`explain (analyze, buffers, format json) ${query.text}`, query.parameters as never); + expect(removed(row["QUERY PLAN"][0].Plan)).toBeLessThan(1000); + } + expect(fetchReport).not.toHaveBeenCalled(); + }, 30_000); + it("joins an in-flight report at shutdown and resumes its saved retry", async () => { + const call = await session(); let release!: (value: SpekoCallReport) => void; + fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = voiceCallReports(db, fetchReport).register(coordinator, () => true); + await vi.waitFor(() => expect(fetchReport).toHaveBeenCalledOnce()); + let stopped = false; const stopping = coordinator.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release(result()); await stopping; await worker.ready; + expect((await report(call.id)).status).toBe("available"); + await start(); await flush(); await quiet(); expect(fetchReport).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index fc36fad8f49..4ad00c8157d 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -82,6 +82,126 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication }; } + it("keeps saved report deadlines quiet across repeated voice connection saves", async () => { + const f = await nativeServiceFixture(); + const [seed] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, f.sessionId)); + const endedAt = new Date(), deadline = Date.now() + 600_000; + await db.update(chatVoiceSessions).set({ state: "ended", endedAt }).where(eq(chatVoiceSessions.id, seed.id)); + const ids = [seed.id, ...Array.from({ length: 204 }, () => randomUUID())]; + await db.insert(chatVoiceSessions).values(ids.slice(1).map(id => ({ + ...seed, id, callerId: randomUUID(), idempotencyKey: id, providerSessionId: randomUUID(), + state: "ended" as const, endedAt, + }))); + await db.insert(chatVoiceReports).values(ids.map(sessionId => ({ companyId: f.companyId, sessionId, nextCheckAt: new Date(deadline) }))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const settle = () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBe(deadline); + }, { timeout: 5000 }); + let selected = 0; const select = db.select.bind(db); + try { + await f.service.registerReportWork(coordinator, () => true).ready; await settle(); + const reads = vi.spyOn(db, "select").mockImplementation((...args) => { + const selection = select(...args), from = selection.from.bind(selection); + vi.spyOn(selection, "from").mockImplementation(table => { + const query = from(table); + if (table === chatVoiceSessions) { + const execute = query.execute.bind(query); + vi.spyOn(query, "execute").mockImplementation(async () => { + const rows = await execute(); selected += rows.length; return rows; + }); + } + return query; + }); + return selection; + }); + try { + for (let index = 0; index < 5; index++) { + await toolAccessService(db).updateConnection(f.connectionId, { name: `Saved report ${index}` }); + await settle(); + } + expect(selected).toBe(0); expect(f.transport.callReport).not.toHaveBeenCalled(); + expect(onError).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); } + } finally { + await coordinator.stop(); vi.useRealTimers(); vi.restoreAllMocks(); + await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.companyId, f.companyId)); + } + }); + + it.each(["hangup", "signed callback", "reply ended"])("wakes report refresh from real %s and advances it on a later callback", async trigger => { + const f = await inboundFixture(); + await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.id, f.sessionId)); + await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, + { approve: true, approvalCode: f.call.approvalCode, issueId: f.issueId }); + const [bound] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const settle = () => vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 }); + const quiet = async () => { + const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(10_000); expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); transactions.mockRestore(); } + }; + try { + await f.service.registerReportWork(coordinator, () => true).ready; + await settle(); expect(coordinator.nextWakeAt()).toBeNull(); await quiet(); + if (trigger === "hangup") await f.service.end(f.companyId, bound.sessionId!, f.input.caller); + else if (trigger === "signed callback") { + const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" }); + await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers); + } else { + const publication = await f.publication(); await f.store.enqueuePublication(f.companyId, publication.id); + f.transport.sendCallMessage.mockRejectedValueOnce(new SpekoProviderError("provider_unavailable", false, 409)); + await f.service.pushReplies(25, f.companyId, f.endpointId); + expect(f.transport.sendCallMessage).toHaveBeenCalledOnce(); + } + await settle(); expect(f.transport.callReport).toHaveBeenCalledOnce(); + const [saved] = await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.sessionId, bound.sessionId!)); + expect(saved.status).toBe("available"); expect(coordinator.nextWakeAt()).toBe(saved.nextCheckAt.getTime()); + await quiet(); expect(f.transport.callReport).toHaveBeenCalledOnce(); + const report = f.sign({ type: "call.report", session_id: f.call.providerSessionId, organization_id: f.event.organization_id }); + await f.service.inbound.lifecycle(f.endpoint.publicId, report.body, report.headers); + await settle(); expect(f.transport.callReport).toHaveBeenCalledTimes(2); expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); vi.useRealTimers(); + const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" }); + await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers); + await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.companyId, f.companyId)); + } + }); + + it("retains a saved report and retries guest transcript projection after a document lock", async () => { + const f = await inboundFixture(true); + await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.id, f.sessionId)); + await f.tool("submit_request", { text: "Draft a public inquiry" }); + const docs = documentService(db), key = `voice-transcript-${f.call.sessionId}`; + await docs.upsertIssueDocument({ issueId: f.call.intakeIssueId!, key, format: "markdown", body: "Older transcript" }); + await docs.lockIssueDocument({ issueId: f.call.intakeIssueId!, key, lockedByUserId: f.callerId }); + const ended = f.sign({ type: "call.report", session_id: f.call.providerSessionId, organization_id: f.event.organization_id }); + await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const settle = () => vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 }); + try { + await f.service.registerReportWork(coordinator, () => true).ready; await settle(); + const [saved] = await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.sessionId, f.call.sessionId!)); + expect(saved.status).toBe("available"); expect(saved.transcript).toHaveLength(1); + expect((await docs.getIssueDocumentByKey(f.call.intakeIssueId!, key))!.body).toBe("Older transcript"); + await docs.unlockIssueDocument(f.call.intakeIssueId!, key); + await vi.advanceTimersByTimeAsync(saved.nextCheckAt.getTime() - Date.now() + 1000); await settle(); + const projected = await docs.getIssueDocumentByKey(f.call.intakeIssueId!, key); + expect(projected!.body).toContain("A test request"); + expect(projected!.sourceTrust).toMatchObject({ preset: "low_trust_review", disposition: "quarantined" }); + expect(f.transport.callReport).toHaveBeenCalledTimes(2); expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); vi.useRealTimers(); + await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.companyId, f.companyId)); + } + }); + it("keeps retained history quiet when a voice connection configuration is saved repeatedly", async () => { const f = await inboundFixture(); const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" }); @@ -150,7 +270,6 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.update(chatVoiceInboundCalls).set({ expiresAt: new Date(0), updatedAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id)); const onError = vi.fn(); const maintenance = createChatReconciliationCoordinator({ - reconcileVoiceSessions: () => f.service.reconcileMaintenance(), processPendingDeliveries: async () => undefined, projectRunMilestones: async () => 0, flushPublications: async () => undefined, onError, }); @@ -1766,12 +1885,19 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.update(chatVoiceReports).set({transcript: [...detail.report.transcript, {...detail.report.transcript[0]!, id: "legacy-control", index: 1, text: VOICE_RESULT_NOTIFICATION}]}).where(eq(chatVoiceReports.sessionId, first.session.id)); expect((await f.service.callDetail(f.companyId, first.session.id, f.input.caller)).report.transcript).toEqual(detail.report.transcript); + const other = await f.service.start({ ...f.input, idempotencyKey: randomUUID() }); + await f.service.end(f.companyId, other.session.id, f.input.caller); + await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 60_000) }).where(eq(chatVoiceSessions.id, other.session.id)); await db.update(chatVoiceReports).set({ nextCheckAt: new Date(0) }).where(eq(chatVoiceReports.sessionId, first.session.id)); - const latest = new Date(Date.now() + 10000); - f.transport.callReport.mockResolvedValueOnce({ ...await f.transport.callReport(), costMicroUsd: "240000", providerUpdatedAt: latest }); + const [stored] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, first.session.id)); + const baseline = await f.transport.callReport(), latest = new Date(Date.now() + 10000); + let response = { ...baseline, costMicroUsd: "240000", providerUpdatedAt: latest }; + f.transport.callReport.mockImplementation(async (providerSessionId?: string) => providerSessionId === stored.providerSessionId ? response : baseline); await f.service.reconcile(1000); + expect((await f.service.callDetail(f.companyId, first.session.id, f.input.caller)).report.costMicroUsd).toBe("240000"); + expect((await f.service.callDetail(f.companyId, other.session.id, f.input.caller)).report.costMicroUsd).toBe("120000"); await db.update(chatVoiceReports).set({ nextCheckAt: new Date(0) }).where(eq(chatVoiceReports.sessionId, first.session.id)); - f.transport.callReport.mockResolvedValueOnce({ ...await f.transport.callReport(), costMicroUsd: "1", providerUpdatedAt: new Date(0) }); + response = { ...baseline, costMicroUsd: "1", providerUpdatedAt: new Date(0) }; await f.service.reconcile(1000); expect((await f.service.callDetail(f.companyId, first.session.id, f.input.caller)).report.costMicroUsd).toBe("240000"); expect(await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.sessionId, first.session.id))).toHaveLength(1); diff --git a/server/src/app.ts b/server/src/app.ts index 50c1596b1d2..169cc9634f4 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -321,7 +321,6 @@ export function shouldEnablePrivateHostnameGuard(opts: { } type ChatReconciliationLane = - | "voice sessions" | "deliveries" | "GitHub webhook recovery" | "run milestones" @@ -335,7 +334,6 @@ type ChatReconciliationLane = * next publication sweep for every endpoint. */ export function createChatReconciliationCoordinator(input: { - reconcileVoiceSessions: () => Promise; processPendingDeliveries: () => Promise; processFailedGitHubWebhookDeliveries?: () => Promise; projectRunMilestones: () => Promise; @@ -377,7 +375,6 @@ export function createChatReconciliationCoordinator(input: { return { reconcile() { if (stopped) return; - start("voice sessions", input.reconcileVoiceSessions); start("deliveries", input.processPendingDeliveries); if (input.processFailedGitHubWebhookDeliveries) { start( @@ -1298,6 +1295,7 @@ export async function createApp( } registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); + chatChannels.voice.registerReportWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.voice.registerRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.voice.registerReplyWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); @@ -1309,7 +1307,6 @@ export async function createApp( await chatChannels.processPublicationMaintenance(); }; const chatReconciliation = createChatReconciliationCoordinator({ - reconcileVoiceSessions: () => chatChannels.voice.reconcileMaintenance(), // Chat delivery, action, and Telegram discovery workers now own these // duties. Keep the coordinator's other maintenance lanes independent. processPendingDeliveries: async () => undefined, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index af91b88a76e..99b6f396067 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -45,6 +45,7 @@ export const DELIVERY_QUEUES = { chatTelegramMaintenance: "chat-telegram-maintenance", chatReceivers: "chat-receivers", voiceSessionRecovery: "voice-session-recovery", + voiceReports: "voice-call-reports", voiceInboundRetention: "voice-inbound-retention", voiceInboundCleanup: "voice-inbound-cleanup", voiceReplies: "voice-reply-delivery", diff --git a/server/src/services/voice/voice-call-reports.ts b/server/src/services/voice/voice-call-reports.ts new file mode 100644 index 00000000000..8f597f9f16b --- /dev/null +++ b/server/src/services/voice/voice-call-reports.ts @@ -0,0 +1,146 @@ +import { and, eq, gte, inArray, isNotNull, lte, ne, or, sql } from "drizzle-orm"; +import { activityLog, chatEndpoints, chatVoiceReports, chatVoiceSessions, issues, type Db } from "@paperclipai/db"; +import { logger } from "../../middleware/logger.js"; +import { documentService } from "../documents.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { voiceRecoveryScope, type VoiceRecoveryScope } from "./voice-recovery-notifications.js"; +import { notifyVoiceReports } from "./voice-report-notifications.js"; +import type { VoiceSessionRow } from "./voice-session-store.js"; +import type { SpekoCallReport } from "./speko-call-report.js"; +import { SpekoProviderError } from "./speko-provider.js"; + +type Entry = { id: string; companyId: string; endpointId: string; connectionId: string; dueAt: number }; +type Report = SpekoCallReport; +const WINDOW_MS = 24 * 60 * 60 * 1000; +const eligible = (now: number) => and(inArray(chatVoiceSessions.state, ["ended", "failed", "expired"]), + isNotNull(chatVoiceSessions.providerSessionId), gte(chatVoiceSessions.endedAt, new Date(now - WINDOW_MS))); + +/** Reports may change for 24 hours after a call. Available is a presentation + * state, not a promise of immutable provider data. Keep that finite obligation. */ +export function voiceCallReports(db: Db, fetchReport: (session: VoiceSessionRow) => Promise) { + type Cursor = { id: string; endedAt: string }; + const cursors = new Map(); + async function page(scope?: VoiceRecoveryScope, cursor?: Cursor, signal?: AbortSignal) { + signal?.throwIfAborted(); + const rows = await db.select({ id: chatVoiceSessions.id, companyId: chatVoiceSessions.companyId, + endpointId: chatVoiceSessions.endpointId, connectionId: chatEndpoints.connectionId, + endedAt: sql`${chatVoiceSessions.endedAt}::text`, + dueAt: sql`least(coalesce(ceil(extract(epoch from ${chatVoiceReports.nextCheckAt}) * 1000), 0), floor(extract(epoch from ${chatVoiceSessions.endedAt}) * 1000) + ${WINDOW_MS} + 1)`, + }).from(chatVoiceSessions) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceSessions.endpointId), eq(chatEndpoints.companyId, chatVoiceSessions.companyId))) + .leftJoin(chatVoiceReports, and(eq(chatVoiceReports.sessionId, chatVoiceSessions.id), eq(chatVoiceReports.companyId, chatVoiceSessions.companyId))) + .where(and(eligible(Date.now()), + scope?.kind === "session" ? eq(chatVoiceSessions.id, scope.id) : undefined, + scope?.kind === "endpoint" ? eq(chatVoiceSessions.endpointId, scope.id) : undefined, + scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined, + scope?.kind === "company" ? eq(chatVoiceSessions.companyId, scope.id) : undefined, + cursor ? sql`(${chatVoiceSessions.endedAt}, ${chatVoiceSessions.id}) > (${cursor.endedAt}::timestamptz, ${cursor.id}::uuid)` : undefined)) + .orderBy(chatVoiceSessions.endedAt, chatVoiceSessions.id).limit(100); + signal?.throwIfAborted(); + return rows; + } + function entries(rows: Awaited>): Entry[] { + return rows.map(({ endedAt: _cursor, ...row }) => ({ ...row, dueAt: Number(row.dueAt) })); + } + async function load(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise { + const result: Entry[] = []; let cursor: Cursor | undefined; + for (;;) { + const rows = await page(scope, cursor, signal); + result.push(...entries(rows)); + if (rows.length < 100) return result; + cursor = rows[rows.length - 1]; + } + } + async function refresh(scope?: VoiceRecoveryScope, signal?: AbortSignal) { + const key = scope ? `${scope.kind}:${scope.id}` : "*"; + const rows = await page(scope, cursors.get(key), signal), result = entries(rows), complete = rows.length < 100; + signal?.throwIfAborted(); + if (complete) cursors.delete(key); else cursors.set(key, rows[rows.length - 1]); + return { rows: result, complete }; + } + async function projectTranscript(session: VoiceSessionRow) { + // Keep both actual spoken sides on this public call's own task. This + // does not publish a document or grant access to any other task. + if (session.callerAuthority !== "guest_intake") return; + await db.transaction(async tx => { + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`voice-transcript:${session.id}`}, 0))`); + const [retained] = await tx.select().from(chatVoiceReports).where(and(eq(chatVoiceReports.companyId, session.companyId), eq(chatVoiceReports.sessionId, session.id))); + if (!retained?.transcript.length) return; + const [task] = await tx.select({sourceTrust: issues.sourceTrust}).from(issues).where(and(eq(issues.companyId, session.companyId), eq(issues.id, session.issueId))); + if (task?.sourceTrust?.preset !== "low_trust_review" || task.sourceTrust.disposition !== "quarantined") return; + const key = `voice-transcript-${session.id}`; + const body = retained.transcript.map(turn => `**${turn.speaker === "caller" ? "Caller" : "Agent"}** · ${turn.startedAt}${turn.interrupted ? " · interrupted" : ""}\n\n${turn.text}`).join("\n\n---\n\n"); + const docs = documentService(tx as unknown as Db), previous = await docs.getIssueDocumentByKey(session.issueId, key); + if (previous?.body === body) return; + await docs.upsertIssueDocument({issueId: session.issueId, key, baseRevisionId: previous?.latestRevisionId, title: "Phone conversation transcript", format: "markdown", body, sourceTrust: task.sourceTrust, changeSummary: "Speko spoken conversation, including caller and agent turns"}); + await tx.insert(activityLog).values({companyId: session.companyId, actorType: "system", actorId: "speko-report", action: "voice.transcript.updated", entityType: "issue", entityId: session.issueId, details: {sessionId: session.id}}); + }); + } + async function check(id: string, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + const claim = await db.transaction(async tx => { + const [session] = await tx.select().from(chatVoiceSessions).where(and(eq(chatVoiceSessions.id, id), eligible(Date.now()))).for("update"); + if (!session) return null; + await notifyVoiceReports(tx, { kind: "session", id }); + await tx.insert(chatVoiceReports).values({ sessionId: id, companyId: session.companyId, nextCheckAt: new Date() }).onConflictDoNothing(); + const [report] = await tx.select().from(chatVoiceReports).where(and(eq(chatVoiceReports.sessionId, id), + eq(chatVoiceReports.companyId, session.companyId), lte(chatVoiceReports.nextCheckAt, new Date()))).for("update"); + if (!report) return null; + signal?.throwIfAborted(); + const attempts = report.attempts + 1; + const retryAt = new Date(Date.now() + Math.min(3600, 30 * 2 ** Math.min(attempts, 7)) * 1000); + await tx.update(chatVoiceReports).set({ attempts, nextCheckAt: retryAt }).where(eq(chatVoiceReports.sessionId, id)); + return { session, attempts, retryAt }; + }); + if (claim) { + const { session, attempts, retryAt } = claim; + // A callback can advance nextCheckAt while this GET is in flight. + // That callback or a newer attempt owns settlement; never overwrite it. + const owned = () => and(eq(chatVoiceReports.sessionId, id), eq(chatVoiceReports.companyId, session.companyId), + eq(chatVoiceReports.attempts, attempts), eq(chatVoiceReports.nextCheckAt, retryAt)); + signal?.throwIfAborted(); + try { + const result = await fetchReport(session); + logger.info({ event: "voice.call.report_received", companyId: session.companyId, endpointId: session.endpointId, + issueId: session.issueId, sessionId: id, providerSessionId: session.providerSessionId, attempts, + complete: result.complete, transcriptTurns: result.transcript.length, providerUpdatedAt: result.providerUpdatedAt?.toISOString() ?? null }, "Voice call report received for reconciliation"); + await db.transaction(async tx => { + await notifyVoiceReports(tx, { kind: "session", id }); + await tx.update(chatVoiceReports).set({ transcript: result.transcript, costMicroUsd: result.costMicroUsd, + durationSeconds: result.durationSeconds, providerUpdatedAt: result.providerUpdatedAt, + status: result.complete ? "available" : "pending", updatedAt: new Date(), + }).where(and(owned(), result.complete ? undefined : ne(chatVoiceReports.status, "available"), + result.providerUpdatedAt ? or(sql`${chatVoiceReports.providerUpdatedAt} is null`, lte(chatVoiceReports.providerUpdatedAt, result.providerUpdatedAt)) : sql`${chatVoiceReports.providerUpdatedAt} is null`)); + }); + // Project the retained authoritative report, including when a response + // was stale. A projection failure retains the already saved retry. + await projectTranscript(session); + } catch (error) { + logger.warn({ event: "voice.call.report_failed", companyId: session.companyId, endpointId: session.endpointId, + issueId: session.issueId, sessionId: id, providerSessionId: session.providerSessionId, attempts, + errorCode: error instanceof SpekoProviderError ? error.code : "report_reconciliation_failed", + httpStatus: error instanceof SpekoProviderError ? error.httpStatus ?? null : null, retryAt: retryAt.toISOString() }, "Voice call report reconciliation deferred"); + if (attempts >= 8) await db.transaction(async tx => { + await notifyVoiceReports(tx, { kind: "session", id }); + await tx.update(chatVoiceReports).set({ status: "unavailable" }).where(and(owned(), ne(chatVoiceReports.status, "available"))); + }); + } + } + return (await load({ kind: "session", id }, signal))[0] ?? null; + } + async function reconcile(limit: number) { + const rows = (await load()).filter(row => row.dueAt <= Date.now()).sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)).slice(0, limit); + for (const row of rows) await check(row.id); + return rows.length; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + cursors.clear(); + return registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.voiceReports, decodeScope: voiceRecoveryScope, + matches: (entry, scope) => !scope || (scope.kind === "session" ? entry.id : scope.kind === "endpoint" ? entry.endpointId : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id, + load: refresh, check: (entry, signal) => check(entry.id, signal), + }, canRun); + } + return { load, check, reconcile, register }; +} diff --git a/server/src/services/voice/voice-recovery-notifications.ts b/server/src/services/voice/voice-recovery-notifications.ts index f95e3627cdb..05ac776a646 100644 --- a/server/src/services/voice/voice-recovery-notifications.ts +++ b/server/src/services/voice/voice-recovery-notifications.ts @@ -1,3 +1,4 @@ +import { notifyVoiceReports } from "./voice-report-notifications.js"; import { notifyVoiceInbound } from "./voice-inbound-notifications.js"; import { notifyVoiceReplies } from "./voice-reply-notifications.js"; import type { Db } from "@paperclipai/db"; @@ -20,6 +21,7 @@ export async function notifyVoiceChange(tx: object, scope?: VoiceRecoveryScope) await notifyVoiceRecovery(tx, scope); await notifyVoiceReplies(tx, scope); await notifyVoiceInbound(tx, scope); + if (scope?.kind === "session") await notifyVoiceReports(tx, { kind: "session", id: scope.id }); } export function withVoiceSessionChange(db: Db, id: string, write: (tx: Parameters[0]>[0]) => PromiseLike, options: { replyDelivery?: boolean } = {}): Promise { return db.transaction(async tx => { diff --git a/server/src/services/voice/voice-reply-push.ts b/server/src/services/voice/voice-reply-push.ts index 31b1d257ef3..9b63fcedab2 100644 --- a/server/src/services/voice/voice-reply-push.ts +++ b/server/src/services/voice/voice-reply-push.ts @@ -1,3 +1,4 @@ +import { notifyVoiceReports } from "./voice-report-notifications.js"; import { notifyVoiceInbound } from "./voice-inbound-notifications.js"; import { notifyVoiceReplies } from "./voice-reply-notifications.js"; import { recoverVoiceReplyDispatches } from "./voice-reply-dispatch-recovery.js"; @@ -84,7 +85,10 @@ export async function pushVoiceReplies(db: Db, options: { await notifyVoiceReplies(tx, { kind: "session", id: candidate.id }); await tx.update(chatActions).set({status, result: {errorCode: providerError?.code ?? "voice_authority_unavailable", ...(status === "retry" ? {retryAt: Date.now() + 10_000} : {})}, updatedAt: new Date()}).where(eq(chatActions.id, claim.action.id)); if (status === "call_ended" || status === "unknown") await notifyVoiceRecovery(tx, { kind: "session", id: candidate.id }); - if (status === "call_ended") await notifyVoiceInbound(tx, { kind: "session", id: candidate.id }); + if (status === "call_ended") { + await notifyVoiceInbound(tx, { kind: "session", id: candidate.id }); + await notifyVoiceReports(tx, { kind: "session", id: candidate.id }); + } if (status === "call_ended") await tx.update(chatVoiceSessions).set({state: "ended", endedAt: new Date(), updatedAt: new Date()}).where(and(eq(chatVoiceSessions.id, candidate.id), eq(chatVoiceSessions.companyId, candidate.companyId))); if (status === "unknown") await tx.update(chatVoiceSessions).set({errorCode: "message_delivery_unknown", updatedAt: new Date()}).where(eq(chatVoiceSessions.id, candidate.id)); }); diff --git a/server/src/services/voice/voice-report-notifications.ts b/server/src/services/voice/voice-report-notifications.ts new file mode 100644 index 00000000000..077b6939c90 --- /dev/null +++ b/server/src/services/voice/voice-report-notifications.ts @@ -0,0 +1,5 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; + +export function notifyVoiceReports(tx: object, scope: { kind: "session"; id: string }) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceReports, `${scope.kind}:${scope.id}`); +} diff --git a/server/src/services/voice/voice-session-service.ts b/server/src/services/voice/voice-session-service.ts index 49a2cc796d5..21b3f9350eb 100644 --- a/server/src/services/voice/voice-session-service.ts +++ b/server/src/services/voice/voice-session-service.ts @@ -5,17 +5,17 @@ import { registerVoiceRecoveryWork } from "./voice-recovery-work.js"; import { notifyVoiceChange, withVoiceSessionChange } from "./voice-recovery-notifications.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { logger } from "../../middleware/logger.js"; +import { voiceCallReports } from "./voice-call-reports.js"; import { captureVoiceDeliveryDiagnostics } from "./voice-delivery-diagnostics.js"; import { pushVoiceReplies } from "./voice-reply-push.js"; import { voiceInboundService } from "./voice-inbound-service.js"; import { createHash, randomUUID } from "node:crypto"; -import { and, asc, desc, eq, inArray, lte, notInArray, or, sql } from "drizzle-orm"; -import { chatVoiceReports, chatVoiceCallbacks, chatConversations, chatEndpointResources, chatEndpoints, chatVoiceSessions, issues, toolConnections, activityLog, type Db } from "@paperclipai/db"; +import { and, desc, eq, inArray, notInArray, or, sql } from "drizzle-orm"; +import { chatVoiceReports, chatVoiceCallbacks, chatConversations, chatEndpointResources, chatEndpoints, chatVoiceSessions, issues, activityLog, type Db } from "@paperclipai/db"; import { VOICE_RESULT_NOTIFICATION, VOICE_REPEAT_NOTIFICATION, type VoiceSession, type VoiceSessionMedia, type VoiceCallHistoryEntry, type VoiceCallReport } from "@paperclipai/shared"; import { conflict, forbidden, notFound, HttpError } from "../../errors.js"; import { instanceSettingsService } from "../instance-settings.js"; import { buildChatCommunicationGuidance } from "../chat-communication-guidance.js"; -import { documentService } from "../documents.js"; import { issueService } from "../issues.js"; import { createSpekoProvider, SpekoProviderError } from "./speko-provider.js"; import { serializeVoiceSession, voiceSessionStore, type VoiceCaller, type VoiceSessionRow } from "./voice-session-store.js"; @@ -229,8 +229,16 @@ export function voiceSessionService(db: Db, options: { function registerReplyWork(coordinator: ReturnType, canRun: () => boolean) { return registerVoiceReplyWork(coordinator, replies, canRun); } - async function reconcileMaintenance(limit = 25) { - await reconcileReports(limit); + const reports = voiceCallReports(db, async session => { + const endpoint = await endpointFor(session.companyId, session.endpointId); + const credentials = await options.credentials(endpoint, session.credentialFingerprint); + const client = provider(credentials.apiKey); + const result = await client.callReport(session.providerSessionId!, [credentials.signingSecret]); + await captureVoiceDeliveryDiagnostics(db, session, client, result.complete); + return result; + }); + function registerReportWork(coordinator: ReturnType, canRun: () => boolean) { + return reports.register(coordinator, canRun); } /** Explicit one-shot recovery for diagnostics/tests; never register on a recurring lane. */ async function reconcile(limit = 25) { @@ -238,7 +246,7 @@ export function voiceSessionService(db: Db, options: { for (const row of due) await recovery.check(row.id).catch(() => undefined); await pushReplies(limit); await inbound.reconcile(limit); - await reconcileReports(limit); + await reports.reconcile(limit); return due.length; } async function pushReplies(limit = 25, companyId?: string, endpointId?: string) { @@ -270,65 +278,6 @@ export function voiceSessionService(db: Db, options: { const [report] = await db.select().from(chatVoiceReports).where(and(eq(chatVoiceReports.companyId, companyId), eq(chatVoiceReports.sessionId, session.id))); return { session: serializeVoiceSession(session), report: reportProjection(report) }; } - async function reconcileReports(limit: number) { - // Repair missed lifecycle callbacks, including later recording-ready reports. - // GET the same call; never create a call or download provider recordings. - const recent = new Date(Date.now() - 24 * 60 * 60 * 1000); - const rows = await db.select({ session: chatVoiceSessions, report: chatVoiceReports }).from(chatVoiceSessions) - .leftJoin(chatVoiceReports, and(eq(chatVoiceReports.sessionId, chatVoiceSessions.id), eq(chatVoiceReports.companyId, chatVoiceSessions.companyId))) - .where(and(inArray(chatVoiceSessions.state, terminalStates), sql`${chatVoiceSessions.providerSessionId} is not null`, - sql`${chatVoiceSessions.endedAt} >= ${recent.toISOString()}`, or(sql`${chatVoiceReports.sessionId} is null`, lte(chatVoiceReports.nextCheckAt, new Date())))) - .orderBy(asc(sql`coalesce(${chatVoiceReports.nextCheckAt}, ${chatVoiceSessions.endedAt})`)).limit(limit); - for (const {session, report} of rows) { - const attempts = (report?.attempts ?? 0) + 1; - const retryAt = new Date(Date.now() + Math.min(3600, 30 * 2 ** Math.min(attempts, 7)) * 1000); - await db.insert(chatVoiceReports).values({ sessionId: session.id, companyId: session.companyId }).onConflictDoNothing(); - try { - const endpoint = await endpointFor(session.companyId, session.endpointId); - const credentials = await options.credentials(endpoint, session.credentialFingerprint); - const client = provider(credentials.apiKey); - const result = await client.callReport(session.providerSessionId!, [credentials.signingSecret]); - logger.info({event: "voice.call.report_received", companyId: session.companyId, endpointId: session.endpointId, - issueId: session.issueId, sessionId: session.id, providerSessionId: session.providerSessionId, attempts, - complete: result.complete, transcriptTurns: result.transcript.length, providerUpdatedAt: result.providerUpdatedAt?.toISOString() ?? null}, "Speko call report received for reconciliation"); - await captureVoiceDeliveryDiagnostics(db, session, client, result.complete); - if (!result.complete && report?.status === "available") { - await db.update(chatVoiceReports).set({ attempts, nextCheckAt: retryAt }).where(eq(chatVoiceReports.sessionId, session.id)); - continue; - } - await db.update(chatVoiceReports).set({ - transcript: result.transcript, costMicroUsd: result.costMicroUsd, durationSeconds: result.durationSeconds, - providerUpdatedAt: result.providerUpdatedAt, status: result.complete ? "available" : "pending", attempts, nextCheckAt: retryAt, updatedAt: new Date(), - }).where(and(eq(chatVoiceReports.sessionId, session.id), eq(chatVoiceReports.companyId, session.companyId), - // A stale report must never replace a newer completed report. - result.providerUpdatedAt ? or(sql`${chatVoiceReports.providerUpdatedAt} is null`, lte(chatVoiceReports.providerUpdatedAt, result.providerUpdatedAt)) : sql`${chatVoiceReports.providerUpdatedAt} is null`)); - // Keep both actual spoken sides on this public call's own task. This - // does not publish a document or grant access to any other task. - if (session.callerAuthority === "guest_intake") await db.transaction(async tx => { - await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`voice-transcript:${session.id}`}, 0))`); - const [retained] = await tx.select().from(chatVoiceReports).where(and(eq(chatVoiceReports.companyId, session.companyId), eq(chatVoiceReports.sessionId, session.id))); - if (!retained?.transcript.length) return; - const [task] = await tx.select({sourceTrust: issues.sourceTrust}).from(issues).where(and(eq(issues.companyId, session.companyId), eq(issues.id, session.issueId))); - if (task?.sourceTrust?.preset !== "low_trust_review" || task.sourceTrust.disposition !== "quarantined") return; - const key = `voice-transcript-${session.id}`; - const body = retained.transcript.map(turn => `**${turn.speaker === "caller" ? "Caller" : "Agent"}** · ${turn.startedAt}${turn.interrupted ? " · interrupted" : ""}\n\n${turn.text}`).join("\n\n---\n\n"); - const docs = documentService(tx as unknown as Db), previous = await docs.getIssueDocumentByKey(session.issueId, key); - if (previous?.body === body) return; - await docs.upsertIssueDocument({issueId: session.issueId, key, title: "Phone conversation transcript", format: "markdown", body, sourceTrust: task.sourceTrust, changeSummary: "Speko spoken conversation, including caller and agent turns"}); - await tx.insert(activityLog).values({companyId: session.companyId, actorType: "system", actorId: "speko-report", action: "voice.transcript.updated", entityType: "issue", entityId: session.issueId, details: {sessionId: session.id}}); - }); - // Even a stale provider reply must back off instead of spinning. - await db.update(chatVoiceReports).set({ attempts, nextCheckAt: retryAt }).where(eq(chatVoiceReports.sessionId, session.id)); - } catch (error) { - logger.warn({event: "voice.call.report_failed", companyId: session.companyId, endpointId: session.endpointId, - issueId: session.issueId, sessionId: session.id, providerSessionId: session.providerSessionId, attempts, - errorCode: error instanceof SpekoProviderError ? error.code : "report_reconciliation_failed", - httpStatus: error instanceof SpekoProviderError ? error.httpStatus ?? null : null, retryAt: retryAt.toISOString()}, "Speko call report reconciliation deferred"); - await db.update(chatVoiceReports).set({ attempts, nextCheckAt: retryAt, ...(report?.status !== "available" && attempts >= 8 ? { status: "unavailable" as const } : {}) }).where(eq(chatVoiceReports.sessionId, session.id)); - } - } - return rows.length; - } async function callbackPreference(companyId: string, endpointId: string, caller: VoiceCaller) { await endpointFor(companyId, endpointId); await db.transaction(tx => store.authorizeCaller(tx, companyId, caller)); @@ -351,7 +300,7 @@ export function voiceSessionService(db: Db, options: { const result = await start({ ...binding, mode: "outbound_phone", phoneNumber: preference.phoneNumber, idempotencyKey: invocationId, maxDurationSeconds: 600 }); return { session: result.session }; }); - return { start, inspect, end, tool, reconcile, reconcileMaintenance, registerRecoveryWork, registerReplyWork, pushReplies, inbound, notification: store.notification, callbackPreference, saveCallbackPreference, history, callDetail, unregisterAgentTools }; + return { start, inspect, end, tool, reconcile, registerReportWork, registerRecoveryWork, registerReplyWork, pushReplies, inbound, notification: store.notification, callbackPreference, saveCallbackPreference, history, callDetail, unregisterAgentTools }; } export type VoiceSessionService = ReturnType;