diff --git a/doc/idle-voice-session-recovery.md b/doc/idle-voice-session-recovery.md new file mode 100644 index 0000000000..e0a138ab7b --- /dev/null +++ b/doc/idle-voice-session-recovery.md @@ -0,0 +1,184 @@ +# Voice session recovery without idle discovery + +The live-session worker replaces the session-discovery portion of the recurring +voice maintenance call. It uses the existing delivery coordinator, transaction +signals and shared scheduler. It adds no tables or migrations. Reply delivery, +inbound admission cleanup, inbound history retention and call reports are separate +obligations; their existing maintenance remains until their own migrations land. + +## Work and deadlines + +Existing `chat_voice_sessions` rows are authoritative. Startup reconstructs live +sessions in UUID pages of 100, joined to their same-company endpoint and connection. +Each background refresh reads at most one inventory page. An unfinished pass has +a one-second continuation; it does not create a permanent empty scan. Only timing +and eligibility fields are selected for discovery. The explicit foreground +reconciliation method keeps its complete inventory behavior. +Terminal history does not create a live-session recovery obligation. With no live +sessions, the worker also avoids reading experimental settings after reconstruction. + +| State | Next obligation | +| --- | --- | +| No provider identity yet | Existing `expiresAt` plus two minutes of teardown grace; no provider creation replay | +| Known provider, ending or invalidated configuration | Immediate cleanup; a recorded cleanup failure waits 30 seconds | +| Connecting or active call | Earlier of session expiry and `updatedAt + 30 seconds` for current authorization and missed-hangup inspection | +| Other live state with provider identity | Saved session expiry | +| Ended, failed or expired | No live-session work; reports have a separate owner | + +Provider maximum duration remains the final bound for uncertain creation. A +late provider response can bind identity while expiry is being considered. The +expiration write requires that provider identity is still absent, so a newly +known call retains exact-call cleanup rather than being marked terminal prematurely. + +Each due call is reloaded before execution. The voice store checks current +company, endpoint, credentials, caller, task, guest sandbox and binding authority. +A 403/404 means authority no longer permits the call. A transient lookup error +retains a retry; it does not authorize ending an otherwise valid call. The existing +voice service owns provider inspection and termination, including single-flight +termination and ambiguous-response confirmation. Recovery never creates a call. + +## Producers and transaction ownership + +| Owner | Changes that wake live-session recovery | +| --- | --- | +| `voice-session-store.ts` | Reservation, guest promotion, connecting-to-active transition and reply cursor writes that change the inspection timestamp | +| `voice-session-service.ts` | Creation claim, binding, known/unknown creation failure, user end, cleanup outcome and foreground inspection | +| `voice-inbound-service.ts` | Guest session creation, guest denial and signed terminal lifecycle callbacks | +| `voice-reply-push.ts` | Confirmed remote end and unknown delivery outcome that changes session state/timestamp | +| Chat configuration helper | Endpoint configuration, pause, reconnect, generation/credential changes, connection hints and company removal | +| Generic tool/access services | Voice connection API mutation, removal and credential revocation during access removal | +| Instance settings | Explicit chat-connector feature enable/disable | + +Existing transactions register the awaited signal before the session/configuration +write. Standalone session updates use the small `withVoiceSessionChange` helper. +Provider requests remain outside database transactions. Nested writes retain outer +commit ownership, rollback does not dispatch work, and uncertain commit settlement +stays with the shared database protocol. + +Membership changes, task reassignment, phone preferences and other authorization +inputs that do not mutate the session or connection remain checked on the existing +30-second active-call inspection deadline, as well as on use. This slice does not +promise instant cleanup for every authority mutation or add cross-cutting voice +hooks throughout those owners. Session activity that changes `updatedAt` retains +the existing inspection-time semantics. The session expiry is always an upper +bound for authorized active work. + +The generic chat connection hint does not always carry a transport. It can cause +a bounded empty voice lookup after a chat configuration write. It does not create +idle polling or a recurring deadline. The shared helper contains only routing; +voice eligibility, authority and provider policy stay with the voice owner. + +## Bounded configuration refreshes + +The voice owner remembers the endpoint/connection fields that affect its existing +deadline policy: endpoint status, runtime generation, connection enabled state, +connection identity, and the chat feature flag. A broad configuration hint first +reads a bounded page of these parent records. If those facts are unchanged, the +cached call deadlines stay in place and no call inventory is read. Metadata-only +writes do not move business deadlines. An unknown or changed parent requires +bounded reconstruction; a real disablement still settles every affected call. + +Only the voice owner interprets these facts. The generic health producer now +sends a health-observation notification through the existing chat routing owner. +It preserves receiver and command signals without generating a voice-configuration +wake. An observation changes neither voice eligibility nor the saved call deadline. +Supported session producers continue to signal their exact session ID. + +The worker processes at most 25 dirty scopes and 25 provider checks per pass. +Scope pagination continues without restarting on every newer hint. Failures retain +their backoff and block stale entries in the failed scope. Successful pages can +make progress while the remaining inventory is discovered. The existing +coordinator can use the actual error deadline, including an already-due healthy +backlog, rather than postponing it to an unrelated fallback retry. + +## Failure and lifecycle behavior + +Refreshes coalesce by scope. Paged reconstruction keeps narrower producer hints, +including a newly committed session behind the UUID cursor. A scope generation +retains configuration changes that arrive between pages. Seen IDs protect +concurrent refresh results; absent cached entries are removed only when the +scope's inventory is complete. A new worker resets partial inventory state before +reconstructing its saved obligations. Changes arriving during refresh remain dirty. Failed refreshes block use of +stale cached entries in that scope. Per-scope and per-call failure backoff starts +at 30 seconds and caps at five minutes. New hints do not erase failure backoff. +Batches rotate across companies, process at most 25 calls, and continue an already +due backlog after one second. A failed provider does not stall independent calls. + +Drain and standby retain dirty work without passive SQL. Resume processes that +work, and restart reconstructs deadlines from the rows. Shutdown joins admitted +provider work and removes the coordinator's subscriptions and timers. Retained +terminal history becomes quiet after any already committed hints settle. + +Process-local signals support the existing single active database writer and its +shared clients. Independent SQL writers or another replica need an explicit wake +contract. This change neither expands Cloud alarm coverage nor relaxes sleep +eligibility. Active calls and their real deadlines remain obligations; local worker +quiet is not proof of host suspension or fleet savings. + +## Verification + +The voice integration suite exercises actual service producers, guest intake, +signed callbacks, current authority, unknown creation and adjacent reply/report +behavior. The recovery-owner suite uses real PostgreSQL and controlled provider +operations for outer commit/rollback, drain, restart, scope failure, mutation during +refresh, paging, company fairness, shutdown and the late-provider-binding race. +The latter race failed before the write-time guard and passes afterward. + +The dated audit records exact commands, results, initial fixture failures and +remaining acceptance work: [voice migration audit](plans/2026-10-25-voice-maintenance-sleep-audit.md). + + +## Landed-foundation acceptance, October 27 + +The voice repair now integrates receiver parent `6e88ca0201`, including Discord +`547fdf2139` and master `f8563d2507`. Exact-patch comparison preserves the repaired +22-file scope and master's shared task-drain and native-checkpoint queue names. +No database file is part of this slice. + +All 725 distinct targeted cases pass again: 112 complete voice/shared/startup +cases, 97 complete voice-store cases, 469 complete tool-access cases, and 47 +complete health/receiver cases. Server production typecheck and build pass. +Voice worker/store and health/receiver fixtures use separate initially empty +self-migrating databases. Tool access uses a separate migrated clone. One initial +tool-access invocation supplied the generic fixture variable instead of the +suite-specific variable: 23 cases passed and 446 skipped. It was not a complete +suite result or an assertion failure. The corrected invocation executes and +passes all 469. Tests are not counted twice. Expected controlled failures and +listener warnings remain in logs. Only documentation changes follow these checks. +Fresh current-head reviews, CI and parent landing remain required. No full local +matrix, provider acceptance, external wake or deployed sleep result is claimed. + + +## Wildcard failure and deterministic generation acceptance, October 28 + +Current-head CI `40633667141` fails server shard 1 at the feature-change test's +34,050 ms result against its 25,000 ms bound; verification fails as an aggregate. +This is an actual required assertion failure, distinct from the older interrupted +build and missing policy script. The old test mixed automatically advancing fake +time with real database latency. It now disables automatic clock advancement, +drives 20 explicit scheduler turns, waits for each pass without advancing the +clock, and requires every invalidated call to close before the unrelated periodic +authority deadline. Removing generation replay still fails with missing closures. +The test no longer benchmarks host/database speed or raises its business bound. + +A real committed-session regression also reproduces wildcard starvation: while +one broad reconstruction backs off, a newly ending exact call is not closed. +The worker now retains the broad retry but considers ready exact hints and cached +due calls independently. Each due call still reloads state and authorization in +its owning check before effects. Failed exact scopes retain their own fence; +the existing stale-cache failure regression still passes. Removing wildcard +isolation fails the new exact-call assertion. Broad reconstruction becomes quiet +after its successful retry, with no continuing SQL afterward. + +The full-row execution lookup and lean inventory lookup now share the live-state, +scope and cursor predicate. Their projections remain intentionally different. +An initial edit script rejected the header cursor name before applying the timing +and predicate changes; the intermediate 19-case result covers only the earlier +worker repair. It is not the final repair evidence. The corrected source passes +all 19 lifecycle cases and 191 complete voice-store/shared/startup cases: 210 +distinct targeted cases. Production server typecheck and build pass. Both guarded +regressions fail at their intended effect assertion and restore source afterward. +The fresh voice-store database self-migrates. Existing controlled failure logs +and signal-listener warnings remain recorded. The older 725-case acceptance +belongs to its stated revision and is not added to this count. Fresh current-head +reviews and all CI remain required before merging after receiver discovery. 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 d4446d4c92..7c63307a24 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1504,6 +1504,15 @@ The older #17490 E2E cancellation failure remains unresolved. The entire origina inventory stays in scope, with voice the next separate reason after receivers. +Receiver discovery is published and attached as [#17572](https://github.com/paperclipai/paperclip/pull/17572) +at 643f4766227e15006b684d2cc7bea7ddd5cc2147, based on #17557. Its live CI run is +40173962833; security and reviews are pending. All 218 selected local cases, +workspace types and build pass. The next audit splits voice into five separate +reasons: live-session recovery, reply push, inbound admission expiry/provider +cleanup, inbound history retention, and call reports. See +[the voice owner audit](2026-10-25-voice-maintenance-sleep-audit.md). +Branch codex/sleep-21-voice-session-recovery starts with the first reason; the +other four remain explicit required slices. No voice migration is implemented yet. ## Receiver review repair The real company-removal producer regression reproduced two inventory reads for @@ -1519,6 +1528,15 @@ full-refresh failure/backoff case. Server typecheck and build pass. Prior broad consumer/workspace verification remains recorded above; these 45 overlap that coverage and are not an additional unique total. Publication requires fresh remote CI and review. The full migration inventory remains outstanding. + + +Voice recovery checkpoint: receiver repair `1604a9fcba` is published to #17572 +and incorporated into the unpublished voice child. The complete 93-case voice +suite, eight new recovery lifecycle tests, and 16 startup wiring cases pass. +See the voice audit for precise coverage and remaining work. Receiver CI +40175242445 is live with no failures at inspection; no final acceptance or merge +is claimed. All remaining reasons, older #17490 failure, and final master +reconciliation stay in scope. ## Acceptance follow-up after the plugin parent merged (October 27) Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at @@ -1545,6 +1563,19 @@ for exact evidence and the older CI runner-loss failure. Fresh CI/reviews and parent landing are still required. The original remaining-duty inventory is unchanged. +## Voice inventory amplification acceptance, October 27 + +The voice recovery slice keeps the original single-duty scope and adds the +normalized health-observation notification contract. Five real metadata writes +and one health check over 205 future calls previously selected 1,230 call rows; +they now select zero and preserve the saved deadline. Startup and changed-parent +reconstruction use bounded pages and retain concurrent wakes. + +The updated [voice audit](2026-10-25-voice-maintenance-sleep-audit.md#voice-session-inventory-amplification-repair-october-27) +records all 725 distinct targeted passing cases, typecheck, build, failed attempts +and guarded regressions. This is worker acceptance, not deployed sleep proof. +Fresh exact-head review and CI remain required before the voice PR can merge. + ## Discord registration landing compatibility, October 27 Company deletion has landed. The Discord command slice retains only its original @@ -1559,3 +1590,31 @@ Receiver discovery keeps its 27-file scope on Discord parent `547fdf2139` and preserves current master queue declarations. All 131 targeted cases, server typecheck and build pass. See the [receiver audit](2026-10-25-chat-receiver-discovery-sleep-audit.md#landed-foundation-integration-acceptance-october-27). Fresh exact-head review, CI and parent landing remain required. + + +## Voice integration with landed foundations, October 27 + +The bounded voice repair now includes receiver `6e88ca0201` and retains its +22-file scope. All 725 targeted cases, production typecheck and server build pass +again. The [voice audit](2026-10-25-voice-maintenance-sleep-audit.md#landed-foundation-acceptance-october-27) +records exact scope, database setup, the initial partial tool-access run and +remaining remote gates. + + +## Voice trusted-policy checkout repair, October 27 + +Receiver `ab39569250` supplies the landed policy checker now required by trusted +CI. Voice server/package code and tests remain identical to the 725-case tested +head. The policy checker and its Node tests pass. The [voice audit](2026-10-25-voice-maintenance-sleep-audit.md#trusted-policy-checkout-compatibility-october-27) +keeps the four old failed gates, including the independent runner shutdown, +separate from current local evidence. Fresh review and CI remain required. + + +## Voice review follow-up, October 28 + +Voice recovery isolates a failed broad inventory retry from ready exact hints +and cached due calls, while retaining on-use authority checks and exact-scope +fences. The generation test uses controlled scheduler turns instead of elapsed +SQL time. All 210 current targeted cases, production typecheck and server build +pass; guarded mutations catch both regressions. The [voice audit](2026-10-25-voice-maintenance-sleep-audit.md#wildcard-failure-and-deterministic-generation-acceptance-october-28) +keeps the actual CI assertion failure and earlier failed attempts separate. diff --git a/doc/plans/2026-10-25-voice-drain-repair.md b/doc/plans/2026-10-25-voice-drain-repair.md new file mode 100644 index 0000000000..8b327b1c57 --- /dev/null +++ b/doc/plans/2026-10-25-voice-drain-repair.md @@ -0,0 +1,34 @@ +# Voice recovery drain repair + +Owner: slice 21, PR #17576. Reviewed baseline: 1bf70185d2. + +Greptile finding 4370544173 is valid. The worker retained dirty scopes while +paused, and reported those in-memory hints as pending work. After the producer +committed, the coordinator therefore retained its intent activity hold. No SQL +was required to reproduce the blocked drain; the earlier quiet assertion checked +SQL only and missed the nonzero idle counter. + +The worker now reports pending dirty scopes only while dispatch is permitted. +It still retains every hint and next deadline. The coordinator independently +checks databaseWorkPending, so an open/uncertain transaction retains its activity +hold until transaction reconciliation establishes settlement. Paused timers then +use the existing passive-maintenance path without changing the idle generation. +Resume processes the retained first session without another producer signal. +No provider behavior or external sleep eligibility rule changes. + +The strengthened real-PostgreSQL test proves: an open write during drain retains +an activity hold; commit releases it; a further minute causes neither SQL nor +an idle-snapshot change; resume processes the one retained session exactly once. + +- /tmp/sleep21-drain-before.log: original worker fails, active is 1 instead of 0. +- /tmp/sleep21-drain-tests.log: all 11 voice recovery lifecycle tests pass. +- /tmp/sleep21-drain-typecheck.log: production server typecheck passes. +- Dedicated migrated database: paperclip_sleep21_drain_20261025_test. + +Propagate this predicate into slice 22's extracted scoped-deadline worker and +all descendants, then rerun its existing consumers. The earlier tool-health, +Discord command, and receiver workers have the same predicate pattern; audit +and repair those at their own parent PRs as a separate propagation pass. Do not +report all stack drain behavior fixed from the voice test alone. The company +inbound cleanup/deletion policy, older E2E failure, current chat milestone audit, +and full original inventory remain outstanding. diff --git a/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md b/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md new file mode 100644 index 0000000000..66d0ad41a6 --- /dev/null +++ b/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md @@ -0,0 +1,297 @@ +# Voice maintenance sleep migration audit + +Baseline: receiver parent #17572, head 643f4766227e15006b684d2cc7bea7ddd5cc2147. +Branch: codex/sleep-21-voice-session-recovery. +Status: implementation and local integration/lifecycle verification in progress; unpublished. + +## Separate the reasons before changing the timer + +The app invokes `voice.reconcile()` from the remaining one-second chat lane. +The public service is created through `chat-channels/speko/runtime.ts`; the +implementation is `voice/voice-session-service.ts`. Its reconcile method hides +five distinct obligations. Each needs its own PR, preserving the others until +its producer and deadline contracts are implemented. + +| Slice | Existing behavior | Proposed owner and trigger | +| --- | --- | --- | +| Live session recovery (this branch) | Reads experimental settings, scans live sessions for expiry, ending, disabled endpoints/connections, generation mismatch and 30-second active-state checks | Voice session owner; committed lifecycle/authorization changes plus real expiry, cleanup and active-call inspection deadlines | +| Reply push | `pushVoiceReplies` scans for provider delivery work | Existing reply/session records, committed source changes, pending provider delivery and claim/retry deadlines | +| Inbound admission expiry and provider cleanup | Scans guest/approval/denied/expired calls past expiresAt, inspects/ends exact provider sessions | Inbound owner; admission/lifecycle/decision changes, saved expiry and unresolved end retries | +| Inbound history retention | Deletes denied/expired/ended rows without sessionId or intakeIssueId once both expiresAt and updatedAt are seven days old | Retention owner; deadline max(expiresAt, updatedAt) + seven days, mutation invalidation and startup reconstruction | +| Call reports | Fetches reports for terminal sessions with provider IDs within 24 hours of endedAt; follows nextCheckAt backoff | Report owner; terminal/lifecycle/report events and saved report recheck deadlines, preserving late recording/transcript behavior | + +Removing one invocation of the outer reconcile method would drop several owners. +First expose the live-session operation independently, register it with the +existing coordinator, and leave the other four calls in their existing lane. +The outer wrapper can disappear only after the remaining owners migrate. + +## Live-session source of truth and invariants + +Existing chat_voice_sessions, chat_endpoints, tool_connections, access/bindings, +and instance settings define eligibility and current authority. No new table is +needed. Current live states are reserved, creating, creation_unknown, connecting, +active and ending. Terminal states are ended, failed and expired. + +The current pass uses 25 oldest updatedAt rows. A session qualifies when expired, +ending, disabled, generation-mismatched, or when a connecting/active session has +not been inspected for 30 seconds. Active-session checks validate current task, +caller and guest authority, then inspect the exact saved providerSessionId to +recover a hangup without a final browser callback. A live call is concrete work; +its bounded provider-state inspection is not empty-instance discovery. + +Unknown creation remains reserved until expiresAt plus two minutes of teardown +grace. The provider maximum duration is the final bound. Never replay creation +because its response was lost. Ending retains exact provider cleanup debt until +termination is confirmed. An error records cleanup_pending and updatedAt. +Preserve these business boundaries and distinguish a database error from a real +revocation before changing recovery behavior. + +An inactive instance with no live sessions must stop this owner's settings and +session queries after reconstruction. Retained terminal sessions alone must not +create a live-session deadline. Report work for those rows belongs to its own +slice and stays active under its existing contract for now. + +## Required producer audit + +- voice-session-store.reserve, including inbound and guest binding paths; +- service start claim, successful provider binding, unknown creation, failed + creation, concurrent end, and failure to persist provider identity; +- closeProvider and user end; inspect and tool callbacks that change session state; +- inbound lifecycle, guest intake, approval/denial and end paths that write sessions; +- signed lifecycle callbacks and report callbacks (separate saved report debt); +- endpoint/connection pause, removal, generation or credential changes; +- feature disable, membership/access revocation, task reassignment, binding and + environment changes used by authorizeSession; +- restart/resume and the supported single active writer contract. + +Use existing transaction ownership. Add intent before the write; transaction +outcome uncertainty remains the shared database layer's responsibility. Prefer +session/endpoint/company scopes rather than refreshing all sessions per mutation. +Authorization stays with the voice store; provider requests stay with the Speko +provider. Shared scheduling consumes normalized work identities and deadlines. + +## Required proof + +Reproduce recurring SQL in no-session and retained-terminal-history states before +removal. Test first real session creation after quiet; nested transaction commit +and rollback; uncertain creation without duplicate provider calls; expiry/grace; +ending cleanup failure and retry; active hangup recovery; revoked membership, +changed task binding and feature disable; guest isolation; multiple-company +fairness and backlog; failed scoped reads; mutation during sweep; drain/resume; +shutdown joining admitted provider work. Keep all four adjacent obligations and +their existing tests running. Inspect `voice-session-store.integration.test.ts` +and current producer fixtures before inventing new test contracts. + +The receiver parent has 218 selected passing tests, workspace typecheck/build, +and is published/attached. CI run 40173962833 is live on its exact head; Greptile, +security and Paperclip review are pending. This is not stack completion or merge +permission. The full original ledger, #17490 failure and final master +reconciliation remain required. + + +## Initial implementation checkpoint + +Live-session recovery now has a voice-owned paged deadline reader and a scoped +coordinator worker. The periodic wrapper retains reply, inbound, and report +maintenance. Session writes signal their existing transactions; standalone service +updates use one small transaction helper. Configuration, feature-disable and +voice tool/access removal paths signal relevant scopes. Active sessions retain +30-second inspection, unknown creation retains the saved expiry plus two-minute +grace, and no call-creation action exists in recovery. Authority lookup failures +retry instead of treating an unavailable database as confirmed revocation. + +Four initial actual-service cases pass: first start after quiet and retirement +on missed hangup, membership revocation, and feature disable; unknown creation +expires at its saved grace deadline without another provider call. Server types +pass after fixing the database projection of the internal runtime generation. +The startup wiring regression failed before removal. This is a checkpoint only: +full producer, lifecycle, race, backlog, and consumer evidence is still required. + +Receiver #17572 has Greptile 5/5 and no failed checks, but Paperclip Review 4/5 +requests suppressing rediscovery after discovery-initiated stops. Its optional +full-refresh coalescing observation will also be addressed in the owning parent +before this branch is published. Preserve all unfinished voice changes in this +commit while repairing that parent. + + +## Expanded lifecycle checkpoint + +Receiver review repairs are published in #17572 at `1604a9fcba` and merged into +this child. The receiver CI run 40175242445 remains live, with no failing checks +at inspection. The voice worker now also coalesces pending scopes under a full +refresh and retains its backoff when narrower signals arrive. + +The complete existing voice session integration suite passes all 93 tests on a +fresh disposable PostgreSQL database. This includes the four initial real-service +producer cases and the adjacent reply, inbound, callback and report consumers. +Eight new recovery-owner integration tests pass: configured terminal history, +outer rollback/commit, first write during drain, transient authority failure +without ending a valid call, restart with the same saved deadline, mutation during +full refresh, a 105-session paged backlog with one failed provider, and shutdown +joining admitted inspection. Their provider operations are controlled; real +service authorization/producer coverage remains in the 93-case suite. + +The 16 startup wiring tests also pass. Initial new-suite failures were fixture +errors (missing channel purpose) and a silence assertion starting before the +committed cleanup_pending refresh had settled. Both were corrected; the final +retry check proves no SQL during the actual backoff. No production failure was +concealed or treated as a successful run. + +Outstanding before publication: complete producer/authorization-change audit, +current-state races, cross-company fairness and failed-scope coverage, affected +shared producer consumers, full workspace verification and remote gates. The +other four voice duties and every remaining original stack item remain required. + + +## Producer and race audit completion + +See [the live-session contract](../idle-voice-session-recovery.md) for the complete +producer mapping, existing transaction ownership and retained active authorization +inspection contract. All production session writers are covered. Phone preferences, +membership and task authority remain checked on the existing active-call deadline +and on use; no new recurring empty inventory scan replaces their absence of signals. + +A late provider binding between grace-period discovery and expiration was reproduced: +the previous update marked a connecting call expired despite its new provider ID. +The expiration update now requires provider identity to remain absent. The regression +proves the known call stays live for exact-provider cleanup. All 11 recovery lifecycle +cases pass, including company fairness and failed-scope fencing. + +Expanded real-service coverage passes for endpoint pause, generic voice connection +disable, task reassignment and signed guest lifecycle settlement. Shared consumer +verification passes 469 tool-access, 18 access-removal, and 207 settings/provider +cases. Workspace typechecking passes. A fresh final voice/receiver/startup selection +and the full local test command are running; build/publication/remote gates remain. +No full local matrix result is claimed from targeted tests. Initial fixture errors +and the reproduced provider-binding race remain recorded above and in test logs. + +## Voice session inventory amplification repair, October 27 + +The voice branch normally integrates receiver parent #17572 at `841d451164` +through `f8d7dd2d14`. The initial single-duty scope is 21 files, with no database +changes. The health-observation routing repair adds one related production path, +`tool-access/catalog/health.ts`, because an observation must not claim that voice +configuration changed. The existing routing helper owns receiver and command +fanout; the voice service owns eligibility and deadline comparison. + +A real-API regression with 205 future live calls reproduced 1,230 selected call +rows across five metadata updates and one health observation. The row count +measures executed database selects, rather than timer calls or helper invocations. +After repair these operations read zero call rows, preserve the exact saved next +deadline, and issue no voice wake for the health-only operation. A separate real +SDK health case confirms receiver and command topics still settle after commit. + +The background path now uses one inventory page of at most 100 lean call headers +per refresh. Parent state is probed separately, in bounded pages, and unchanged +state keeps existing deadlines. Unknown or genuinely changed parent state has a +bounded reconstruction. Foreground reconciliation keeps its complete scan API. +The worker retains a generation and seen IDs across pages, processes at most 25 +scopes per pass, rechecks call state before provider effects, and preserves failed +scope/call backoff. It resets partial state when the same service starts a new +worker. No per-worker table, migration, extra scheduler, or route recovery wrapper +is introduced. + +Real database regressions cover all 205 deadlines after paging and partial-scan +restart, a new session behind the cursor, a feature change between pages, and +connection disablement settling all 205 calls. Guarded production mutations +reproduce the page-size violation, lost global generation (cleanup delayed past +its prompt-recovery assertion), and health-only wake. Each original file is +restored after the mutation. These checks supplement the existing outer-commit, +rollback, drain, current-authority, late-provider-binding and shutdown cases. + +Both newly supported external fixture paths skipped migrations. Empty-database +runs failed with missing instance_settings before the worker or provider action. +The worker and complete voice-store fixtures now apply pending migrations before +creating the client and use the shared PostgreSQL bootstrap budget. The initial +amplification fixture also omitted its required Speko provider config, then used +the wrong public health method name; those failures preceded the actual 1,230-row +baseline. Initial quiet checks counted the observer's previous calls; measurement +is now stopped before an independent quiet-window spy is created. The quiet +assertion itself is unchanged. The first typecheck exposed timing-only rows being +used for full provider operations; the timing-header type stays separate from full +rows at that operation boundary. All failures remain recorded as such. + +Current targeted acceptance passes 725 distinct cases: 18 voice lifecycle, 94 +shared coordinator/batching/startup/hostname, 97 complete voice-store, 469 +tool-access, and 47 health/receiver cases. Production typecheck and server build +pass. The 18 lifecycle cases replace the earlier 16-case result; these counts do +not add repeated runs. Full suites ran unless a narrower selection is explicitly +named above. No full local repository matrix, actual host suspension, Cloud wake, +multi-writer support, or savings is inferred from this worker migration. The original remaining-duty inventory stays +in scope. Current exact-head reviews, security, CI, and parent landing are still +required before merging #17576. + + +## Landed-foundation acceptance, October 27 + +The voice repair now integrates receiver parent `6e88ca0201`, including Discord +`547fdf2139` and master `f8563d2507`. Exact-patch comparison preserves the repaired +22-file scope and master's shared task-drain and native-checkpoint queue names. +No database file is part of this slice. + +All 725 distinct targeted cases pass again: 112 complete voice/shared/startup +cases, 97 complete voice-store cases, 469 complete tool-access cases, and 47 +complete health/receiver cases. Server production typecheck and build pass. +Voice worker/store and health/receiver fixtures use separate initially empty +self-migrating databases. Tool access uses a separate migrated clone. One initial +tool-access invocation supplied the generic fixture variable instead of the +suite-specific variable: 23 cases passed and 446 skipped. It was not a complete +suite result or an assertion failure. The corrected invocation executes and +passes all 469. Tests are not counted twice. Expected controlled failures and +listener warnings remain in logs. Only documentation changes follow these checks. +Fresh current-head reviews, CI and parent landing remain required. No full local +matrix, provider acceptance, external wake or deployed sleep result is claimed. + + +## Trusted-policy checkout compatibility, October 27 + +CI run `40632057899` on `712f3862b5` failed policy because the trusted workflow +had acquired `scripts/check-query-error-rendering.mjs` from newer master, while +the checked-out PR head lacked that script. The build job separately received +a runner shutdown signal during runtime bundling, then reported cancellation. +The underlying shutdown cause is not established. The E2E aggregate reports +`E2E_SHARDS_RESULT=success` but fails on `POLICY_RESULT=failure`; verification +also fails on policy and build. Those four failed gates are not passing results. + +A normal merge of receiver `ab39569250` brings the landed UI and policy tooling +into the checkout. Every server/package production and test file remains +byte-identical to the 725-case tested `712f3862b5` source. The policy checker now +executes successfully, and its complete Node test file passes locally. No sleep +implementation, test assertion or workflow gate is weakened. Fresh CI must prove +the full build and aggregate gates; the interrupted build is not relabeled as a +pass. Fresh Ponytail and Greptile acceptance still precede any merge. + + +## Wildcard failure and deterministic generation acceptance, October 28 + +Current-head CI `40633667141` fails server shard 1 at the feature-change test's +34,050 ms result against its 25,000 ms bound; verification fails as an aggregate. +This is an actual required assertion failure, distinct from the older interrupted +build and missing policy script. The old test mixed automatically advancing fake +time with real database latency. It now disables automatic clock advancement, +drives 20 explicit scheduler turns, waits for each pass without advancing the +clock, and requires every invalidated call to close before the unrelated periodic +authority deadline. Removing generation replay still fails with missing closures. +The test no longer benchmarks host/database speed or raises its business bound. + +A real committed-session regression also reproduces wildcard starvation: while +one broad reconstruction backs off, a newly ending exact call is not closed. +The worker now retains the broad retry but considers ready exact hints and cached +due calls independently. Each due call still reloads state and authorization in +its owning check before effects. Failed exact scopes retain their own fence; +the existing stale-cache failure regression still passes. Removing wildcard +isolation fails the new exact-call assertion. Broad reconstruction becomes quiet +after its successful retry, with no continuing SQL afterward. + +The full-row execution lookup and lean inventory lookup now share the live-state, +scope and cursor predicate. Their projections remain intentionally different. +An initial edit script rejected the header cursor name before applying the timing +and predicate changes; the intermediate 19-case result covers only the earlier +worker repair. It is not the final repair evidence. The corrected source passes +all 19 lifecycle cases and 191 complete voice-store/shared/startup cases: 210 +distinct targeted cases. Production server typecheck and build pass. Both guarded +regressions fail at their intended effect assertion and restore source afterward. +The fresh voice-store database self-migrates. Existing controlled failure logs +and signal-listener warnings remain recorded. The older 725-case acceptance +belongs to its stated revision and is not added to this count. Fresh current-head +reviews and all CI remain required before merging after receiver discovery. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index d715145e5f..32498f80ad 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -78,10 +78,17 @@ describe("createChatReconciliationCoordinator", () => { 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.reconcile()"); + expect(maintenance).toContain("chatChannels.voice.reconcileMaintenance()"); expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); }); + it("separates live voice recovery from the remaining periodic maintenance", () => { + const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); + expect(source).toContain("chatChannels.voice.registerRecoveryWork(deliveryWork,"); + expect(source).toContain("chatChannels.voice.reconcileMaintenance()"); + expect(source).not.toContain("chatChannels.voice.reconcile()"); + }); + it("wires GitHub recovery into its independent lane", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); expect(source).toContain( diff --git a/server/src/__tests__/voice-recovery-work.test.ts b/server/src/__tests__/voice-recovery-work.test.ts new file mode 100644 index 0000000000..f664754b7b --- /dev/null +++ b/server/src/__tests__/voice-recovery-work.test.ts @@ -0,0 +1,356 @@ +import { HttpError } from "../errors.js"; +import { toolAccessService } from "../services/tool-access.js"; +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, agents, chatConversations, chatEndpoints, chatVoiceSessions, companies, createDb, issues, subscribeDatabaseWork, 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 { instanceSettingsService } from "../services/instance-settings.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { voiceSessionRecovery } from "../services/voice/voice-session-recovery.js"; +import { registerVoiceRecoveryWork } from "../services/voice/voice-recovery-work.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { notifyVoiceRecovery, withVoiceSessionChange } from "../services/voice/voice-recovery-notifications.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("voice session recovery 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 authorize = vi.fn(async (_session: typeof chatVoiceSessions.$inferSelect) => {}), inspect = vi.fn(async (_session: typeof chatVoiceSessions.$inferSelect): Promise<{ endedAt?: unknown }> => ({ endedAt: true })); + const close = vi.fn(async (session: typeof chatVoiceSessions.$inferSelect) => { + await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id))); + }); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("voice-recovery-"); + if (external) await applyPendingMigrations(external); + db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true }); + }, 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", config: { provider: "speko" }, 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; authorize.mockClear(); inspect.mockReset().mockResolvedValue({ endedAt: true }); close.mockClear(); + ({ 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: "active", 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 = voiceSessionRecovery(db, { authorize, inspect, close })) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + await registerVoiceRecoveryWork(coordinator, service, () => enabled).ready; + return service; + } + async function settled() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); }, { timeout: 5000 }); + } + async function quiet(ms = 120_000) { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(ms); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + } + async function futureInventory(count: number) { + const row = await session({ updatedAt: new Date(), expiresAt: new Date(Date.now() + 3_600_000) }); + await db.insert(chatVoiceSessions).values(Array.from({ length: count - 1 }, () => { + const id = randomUUID(); + return { ...row, id, callerId: randomUUID(), providerSessionId: id, idempotencyKey: id, requestFingerprint: id }; + })); + return row.updatedAt.getTime() + 30_000; + } + function voiceReads() { + const counts: number[] = [], select = db.select.bind(db); + const measurement = 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; + }); + return { counts, stop: () => measurement.mockRestore() }; + } + async function futureSettled(at: number) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBe(at); + }, { timeout: 5000 }); + } + it("bounds startup pages and retains every saved deadline", async () => { + const due = await futureInventory(205), reads = voiceReads(), counts = reads.counts; + await start(); + expect(counts).toHaveLength(1); expect(counts[0]).toBe(100); + await futureSettled(due); + expect(counts.reduce((sum, count) => sum + count, 0)).toBe(205); + expect(Math.max(...counts)).toBe(100); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(new Set(inspect.mock.calls.map(([row]) => row.id)).size).toBe(205); + reads.stop(); await quiet(); + }); + it("reconstructs every page when the same service restarts after a partial scan", async () => { + const due = await futureInventory(205), service = voiceSessionRecovery(db, { authorize, inspect, close }); + await start(service); expect(inspect).not.toHaveBeenCalled(); await coordinator!.stop(); + await start(service); await futureSettled(due); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(new Set(inspect.mock.calls.map(([row]) => row.id)).size).toBe(205); + await quiet(); + }); + it("retains a new session behind the cursor during paged reconstruction", async () => { + const due = await futureInventory(205), service = voiceSessionRecovery(db, { authorize, inspect, close }); + const refresh = service.refresh; let inserted = false, newcomer = ""; + service.refresh = async (...args) => { + const result = await refresh(...args); + if (!args[0] && !result.complete && !inserted) { + inserted = true; newcomer = (await session({ id: "00000000-0000-4000-8000-000000000001" })).id; + } + return result; + }; + await start(service); await futureSettled(due); + expect(inspect.mock.calls.map(([row]) => row.id)).toEqual([newcomer]); + expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, newcomer)))[0].state).toBe("ended"); + await quiet(5000); + }); + it("reapplies a feature change that arrives between inventory pages", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + await futureInventory(205); + const settings = instanceSettingsService(db), service = voiceSessionRecovery(db, { authorize, inspect, close }); + const refresh = service.refresh; let changed = false; + service.refresh = async (...args) => { + const result = await refresh(...args); + if (!args[0] && !result.complete && !changed) { + changed = true; await settings.updateExperimental({ enableChatConnectors: false }); + } + return result; + }; + try { + await start(service); + 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(new Set(close.mock.calls.map(([row]) => row.id)).size).toBe(205); + await settled(); + await quiet(); + } finally { + await coordinator?.stop(); coordinator = undefined; + await settings.updateExperimental({ enableChatConnectors: true }); + } + }); + it("pages actual connection disablement and settles every cached live call", async () => { + const due = await futureInventory(205); await start(); await futureSettled(due); + const [endpoint] = await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, endpointId)); + const reads = voiceReads(); + authorize.mockRejectedValue(new HttpError(403, "Connection disabled")); + try { + await toolAccessService(db).updateConnection(endpoint.connectionId, { enabled: false }); + await settled(); + expect(new Set(close.mock.calls.map(([row]) => row.id)).size).toBe(205); + expect(inspect).not.toHaveBeenCalled(); expect(Math.max(...reads.counts)).toBe(100); + reads.stop(); await quiet(); + } finally { authorize.mockResolvedValue(undefined); reads.stop(); } + }); + it("does not reread live calls after repeated real connection metadata and health writes", async () => { + const due = await futureInventory(205); + await start(); await futureSettled(due); + const reads = voiceReads(), counts = reads.counts; + const [endpoint] = await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, endpointId)); + const tools = toolAccessService(db); + for (let index = 0; index < 5; index++) { + await tools.updateConnection(endpoint.connectionId, { name: `Metadata ${index}` }); + await futureSettled(due); + } + let observations = 0; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.voiceSessionRecovery, event => { if (event === "settled") observations++; }); + try { + await expect(tools.checkHealth(endpoint.connectionId)).rejects.toMatchObject({ status: 422 }); + await futureSettled(due); + expect(counts.reduce((sum, count) => sum + count, 0)).toBe(0); + expect(observations).toBe(0); + expect(inspect).not.toHaveBeenCalled(); reads.stop(); await quiet(5000); + } finally { unsubscribe(); } + }); + it("preserves receiver and command wakes for SDK health observations", async () => { + const id = randomUUID(); + await db.insert(toolConnections).values({ id, companyId, applicationId: (await db.select().from(toolApplications).where(eq(toolApplications.companyId, companyId)))[0].id, + name: "SDK observation", uid: id, transport: "chat_sdk", connectionPurpose: "channel", config: { provider: "discord" }, enabled: true, status: "active" }); + const seen = new Set(); + const topics = [DELIVERY_QUEUES.chatReceivers, DELIVERY_QUEUES.chatCommandRegistration, DELIVERY_QUEUES.voiceSessionRecovery]; + const stops = topics.map(topic => subscribeDatabaseWork(db, topic, event => { if (event === "settled") seen.add(topic); })); + try { + await expect(toolAccessService(db).checkHealth(id)).rejects.toMatchObject({ status: 422 }); + expect(seen).toEqual(new Set([DELIVERY_QUEUES.chatReceivers, DELIVERY_QUEUES.chatCommandRegistration])); + } finally { for (const stop of stops) stop(); } + }); + it("leaves configured terminal history quiet", async () => { + await session({ state: "ended", endedAt: new Date() }); await start(); await settled(); await quiet(); expect(inspect).not.toHaveBeenCalled(); + }); + it("waits for outer commit and ignores a rolled-back nested producer", async () => { + await start(); await settled(); + await expect(db.transaction(async tx => { await session({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(60_000); expect(inspect).not.toHaveBeenCalled(); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await settled(); await quiet(); + await db.transaction(async tx => { await session({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(60_000); expect(inspect).not.toHaveBeenCalled(); }); + await settled(); expect(inspect).toHaveBeenCalledOnce(); await quiet(); + }); + it("preserves the first session during drain without passive database reads", async () => { + await start(); await settled(); enabled = false; + await db.transaction(async tx => { + await session({}, tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(60_000); + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + expect(inspect).not.toHaveBeenCalled(); + }); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }); + const drained = idleWorkSnapshot(); + await quiet(60_000); expect(idleWorkSnapshot()).toEqual(drained); expect(inspect).not.toHaveBeenCalled(); + enabled = true; await settled(); expect(inspect).toHaveBeenCalledOnce(); await quiet(); + }); + it("retries unavailable authority without ending a valid provider call", async () => { + const row = await session(); authorize.mockRejectedValueOnce(new Error("authority database unavailable")); + await start(); expect(errors).toHaveLength(1); expect(close).not.toHaveBeenCalled(); expect(inspect).not.toHaveBeenCalled(); + // Persisting cleanup_pending emits a committed hint; let that refresh settle first. + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now() + 20_000); + await quiet(10_000); await settled(); + expect(inspect).toHaveBeenCalledOnce(); expect(close).not.toHaveBeenCalled(); + expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, row.id)))[0].state).toBe("ended"); + errors.length = 0; await quiet(); + }); + it("reconstructs a saved session deadline after restart", async () => { + const due = Date.now() + 60_000; + await session({ updatedAt: new Date(due - 30_000) }); await start(); + expect(coordinator!.nextWakeAt()).toBe(due); await coordinator!.stop(); + await start(); expect(coordinator!.nextWakeAt()).toBe(due); await quiet(10_000); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); expect(inspect).toHaveBeenCalledOnce(); await quiet(); + }); + it("preserves a session committed during a full refresh", async () => { + const service = voiceSessionRecovery(db, { authorize, inspect, close }), original = service.refresh; + await start(service); await settled(); let inserted = false; + service.refresh = async (...args) => { const rows = await original(...args); if (!inserted) { inserted = true; await session(); } return rows; }; + await db.transaction(tx => notifyVoiceRecovery(tx)); await settled(); expect(inspect).toHaveBeenCalledOnce(); await quiet(); + }); + it("pages saved sessions and drains healthy work while one provider fails", async () => { + const failed = await session(); + for (let index = 0; index < 104; index++) await session(); + let broken = true; + inspect.mockImplementation(async row => { + if (row.id === failed.id && broken) throw new Error("one provider unavailable"); + return { endedAt: true }; + }); + await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + const rows = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.companyId, companyId)); + expect(rows.filter(row => row.state === "ended")).toHaveLength(104); + }, { timeout: 5000 }); + expect(inspect.mock.calls.filter(([row]) => row.id === failed.id)).toHaveLength(1); + expect(errors).toHaveLength(1); + broken = false; await settled(); errors.length = 0; await quiet(); + }); + it("includes another company in the first batch despite a large older backlog", async () => { + for (let index = 0; index < 30; index++) await session({ updatedAt: new Date(Date.now() - 120_000) }); + const other = await fixture(); + const row = await session(other); + await start(); + expect(inspect.mock.calls.slice(0, 25).some(([candidate]) => candidate.id === row.id)).toBe(true); + await settled(); expect(inspect).toHaveBeenCalledTimes(31); await quiet(); + }); + it("keeps exact hints and cached due calls moving while wildcard reconstruction backs off", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + const cached = await session({ updatedAt: new Date(Date.now() - 20_000) }); + const dueAt = cached.updatedAt.getTime() + 30_000; + const service = await start(), refresh = service.refresh; + service.refresh = async (scope, signal) => { + if (!scope) throw new Error("wildcard unavailable"); + return refresh(scope, signal); + }; + await db.transaction(tx => notifyVoiceRecovery(tx)); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(errors).toHaveLength(1), { interval: 0 }); + const exact = await session({ state: "ending" }); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(close).toHaveBeenCalledWith(expect.objectContaining({ id: exact.id })), { interval: 0 }); + await vi.advanceTimersByTimeAsync(Math.max(0, dueAt - Date.now())); + await vi.waitFor(() => expect(inspect).toHaveBeenCalledWith(expect.objectContaining({ id: cached.id })), { interval: 0 }); + expect(errors).toHaveLength(1); + service.refresh = refresh; errors.length = 0; + await vi.advanceTimersByTimeAsync(Math.max(0, coordinator!.nextWakeAt()! - Date.now())); + await settled(); await quiet(); + }); + it("does not inspect stale cached work until a failed scope refresh succeeds", async () => { + const row = await session({ updatedAt: new Date(Date.now() - 20_000) }); + const service = voiceSessionRecovery(db, { authorize, inspect, close }), original = service.refresh; + await start(service); service.refresh = async () => { throw new Error("scope unavailable"); }; + await db.transaction(tx => notifyVoiceRecovery(tx, { kind: "session", id: row.id })); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors).toHaveLength(1); }); + await quiet(15_000); expect(inspect).not.toHaveBeenCalled(); + service.refresh = original; await settled(); expect(inspect).toHaveBeenCalledOnce(); errors.length = 0; await quiet(); + }); + it("does not expire a provider identity bound after the grace-period read", async () => { + const row = await session({ state: "creation_unknown", providerSessionId: null, + createdAt: new Date(Date.now() - 240_000), expiresAt: new Date(Date.now() - 180_000) }); + const service = voiceSessionRecovery(db, { authorize, inspect, close }); + const providerSessionId = randomUUID(), transact = db.transaction.bind(db); + const transaction = vi.spyOn(db, "transaction").mockImplementationOnce(async (callback, options) => { + await transact(async tx => { + await notifyVoiceRecovery(tx, { kind: "session", id: row.id }); + await tx.update(chatVoiceSessions).set({ providerSessionId, state: "connecting", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, row.id)); + }); + return transact(callback, options); + }); + try { await service.check(row.id); } finally { transaction.mockRestore(); } + const [bound] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, row.id)); + expect(bound).toMatchObject({ state: "connecting", providerSessionId }); + await service.check(row.id); + expect(close).toHaveBeenCalledWith(expect.objectContaining({ id: row.id, providerSessionId })); + expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, row.id)))[0].state).toBe("ended"); + }); + it("joins admitted provider inspection at shutdown", async () => { + await start(); await settled(); + let finish!: () => void; const gate = new Promise(resolve => { finish = resolve; }); + inspect.mockImplementationOnce(async () => { await gate; return { endedAt: true }; }); + await session(); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(inspect).toHaveBeenCalledOnce(); }); + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(5000); expect(stopped).toBe(false); + finish(); await stopping; expect(stopped).toBe(true); + }); +}); diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 4b1d0dbf17..e5a7f53d7e 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -1,3 +1,7 @@ +import { toolAccessService } from "../services/tool-access.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import express from "express"; import request from "supertest"; import { errorHandler } from "../middleware/error-handler.js"; @@ -11,7 +15,7 @@ import { VOICE_RESULT_NOTIFICATION } from "@paperclipai/shared"; import { createHash, createHmac, randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { chatActions, instanceUserRoles, principalPermissionGrants, environments, chatVoiceInboundCalls, chatVoicePhoneLines, chatVoiceReports, heartbeatRuns, toolProfileBindings, chatVoiceCallbacks, agentWakeupRequests, issueThreadInteractions, issueQuestionResponseDeliveries, issueComments, agents, chatConversations, chatDeliveries, chatEndpoints, chatPublications, chatVoiceReplies, chatVoiceSessions, chatVoiceToolCalls, companies, companyMemberships, companySecrets, connectionGrants, connectionGrantMembers, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, issues, toolApplications, toolConnections, toolConnectionInstalls } from "@paperclipai/db"; +import { applyPendingMigrations, chatActions, instanceUserRoles, principalPermissionGrants, environments, chatVoiceInboundCalls, chatVoicePhoneLines, chatVoiceReports, heartbeatRuns, toolProfileBindings, chatVoiceCallbacks, agentWakeupRequests, issueThreadInteractions, issueQuestionResponseDeliveries, issueComments, agents, chatConversations, chatDeliveries, chatEndpoints, chatPublications, chatVoiceReplies, chatVoiceSessions, chatVoiceToolCalls, companies, companyMemberships, companySecrets, connectionGrants, connectionGrantMembers, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, issues, toolApplications, toolConnections, toolConnectionInstalls } from "@paperclipai/db"; import { configureSpekoSessionTools } from "../services/voice/speko-tool-setup.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { chatChannelService } from "../services/chat-channels.js"; @@ -28,15 +32,17 @@ import { buildPaperclipRuntimeMcpServers } from "../services/heartbeat.js"; import { createToolGatewayService } from "../services/tool-gateway.js"; import { initializeRunIdentity } from "../services/run-identity.js"; import { syncSpekoVoiceTools, spekoToolsForSession, executeSpekoVoiceTool } from "../services/voice/speko-agent-tools.js"; -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("durable Speko voice sessions", () => { let db: ReturnType; - let database: Awaited>; + let database: Awaited> | undefined; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-speko-store-"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-speko-store-"); + if (external) await applyPendingMigrations(external); + db = createDb(external ?? database!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); - afterAll(async () => { await database?.cleanup(); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); async function fixture() { const companyId = randomUUID(), endpointId = randomUUID(), agentId = randomUUID(), applicationId = randomUUID(), connectionId = randomUUID(), issueId = randomUUID(), conversationId = randomUUID(), callerId = randomUUID(); @@ -67,6 +73,93 @@ const support = await getEmbeddedPostgresTestSupport(); return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication }; } + it.each(["hangup", "revocation", "feature disable", "endpoint pause", "connection disable", "task reassignment"] as const)("wakes live recovery from a real start and settles after %s", async reason => { + await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true }); + const f = await nativeServiceFixture(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const settle = async (terminal = false) => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + if (terminal) expect(coordinator.nextWakeAt()).toBeNull(); + else expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now()); + }); + const quiet = async () => { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + }; + try { + await f.service.registerRecoveryWork(coordinator, () => true).ready; + await settle(true); await quiet(); + const started = await f.service.start(f.input); await settle(); + expect(f.transport.createBrowserSession).toHaveBeenCalledOnce(); + expect(f.transport.inspectSession).not.toHaveBeenCalled(); + if (reason === "hangup") f.transport.inspectSession.mockResolvedValue({ status: "ended", endedAt: new Date().toISOString() }); + if (reason === "revocation") await db.update(companyMemberships).set({ status: "removed" }).where(eq(companyMemberships.principalId, f.callerId)); + if (reason === "feature disable") await instanceSettingsService(db).updateExperimental({ enableChatConnectors: false }); + if (reason === "endpoint pause") { + const chat = chatChannelService(db, { heartbeat: { wakeup: vi.fn() } }); + try { await chat.configure(f.endpointId, { action: "pause" }, f.callerId); } + finally { await chat.shutdown(); } + } + if (reason === "connection disable") await toolAccessService(db).updateConnection(f.connectionId, { enabled: false }); + if (reason === "task reassignment") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId)); + if (["hangup", "revocation", "task reassignment"].includes(reason)) await vi.advanceTimersByTimeAsync(31_000); + await settle(true); + const [row] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, started.session.id)); + expect(row.state).toBe("ended"); + if (reason !== "hangup") expect(f.transport.endSession).toHaveBeenCalledOnce(); + await quiet(); expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); f.service.unregisterAgentTools(); vi.useRealTimers(); } + }); + + it("retires uncertain creation at its saved grace deadline without creating another provider call", async () => { + await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true }); + const f = await nativeServiceFixture(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await f.service.registerRecoveryWork(coordinator, () => true).ready; + f.transport.createBrowserSession.mockRejectedValue(new SpekoProviderError("provider_unavailable", true)); + await expect(f.service.start({ ...f.input, maxDurationSeconds: 30 })).rejects.toMatchObject({ status: 409 }); + const row = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.idempotencyKey, f.input.idempotencyKey)).then(rows => rows[0]); + const deadline = row.expiresAt.getTime() + 120_001; + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBe(deadline); }); + const reads = vi.spyOn(db, "select"); + try { await vi.advanceTimersByTimeAsync(100_000); expect(reads).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); } + await vi.advanceTimersByTimeAsync(deadline - Date.now()); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); }); + expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, row.id)))[0].state).toBe("expired"); + expect(f.transport.createBrowserSession).toHaveBeenCalledOnce(); expect(f.transport.endSession).not.toHaveBeenCalled(); + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); f.service.unregisterAgentTools(); vi.useRealTimers(); } + }); + + it("recovers a real guest session and settles from its signed lifecycle event", async () => { + const f = await inboundFixture(true); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const settle = async () => vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }); + try { + await f.service.registerRecoveryWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + await f.tool("submit_request", { text: "Help draft a public question" }); + await settle(); expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now()); + await vi.advanceTimersByTimeAsync(31_000); await settle(); + expect(f.transport.inspectSession).toHaveBeenCalledWith(f.event.session_id); + expect(f.transport.endSession).not.toHaveBeenCalled(); + const ended = f.sign({ type: "call.status", call_id: f.event.call_id, status: "ended" }); + await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers); + await settle(); expect(coordinator.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); f.service.unregisterAgentTools(); vi.useRealTimers(); } + }); + it("reports signing-secret presence for interrupted setup without returning credential refs", async () => { const f = await fixture(); const chat = chatChannelService(db, {heartbeat: {wakeup: vi.fn()}}); diff --git a/server/src/app.ts b/server/src/app.ts index 1c95c88a45..3700918dba 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1298,6 +1298,7 @@ export async function createApp( } registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); + chatChannels.voice.registerRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); @@ -1305,7 +1306,7 @@ export async function createApp( await chatChannels.processPublicationMaintenance(); }; const chatReconciliation = createChatReconciliationCoordinator({ - reconcileVoiceSessions: () => chatChannels.voice.reconcile(), + 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/access.ts b/server/src/services/access.ts index f23a0ec82e..7cd68a1438 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyVoiceRecovery } from "./voice/voice-recovery-notifications.js"; import { notifyChatEndpointWork } from "./chat-work-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { and, eq, inArray, ne, notInArray, sql } from "drizzle-orm"; @@ -328,6 +329,7 @@ export function accessService(db: Db) { && grant.credentialSecretRefs.every((ref) => !connectionSecretIdsToRemove.has(ref.secretId)), ); if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id); + if (connection.transport === "voice") await notifyVoiceRecovery(tx, { kind: "connection", id: connection.id }); if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, { companyId, id: connection.id }); await tx.update(toolConnections).set({ credentialRefs, diff --git a/server/src/services/chat-channels/configuration-work-notifications.ts b/server/src/services/chat-channels/configuration-work-notifications.ts index 5b592e896e..e8a357daab 100644 --- a/server/src/services/chat-channels/configuration-work-notifications.ts +++ b/server/src/services/chat-channels/configuration-work-notifications.ts @@ -1,8 +1,10 @@ +import { notifyVoiceRecovery } from "../voice/voice-recovery-notifications.js"; import { notifyChatReceiver, notifyChatConnectionReceiver, notifyChatCompanyReceivers } from "./runtime/receiver-work-notifications.js"; import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js"; export async function notifyChatEndpointConfiguration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { + if (endpoint.provider === "speko") await notifyVoiceRecovery(transaction, { kind: "endpoint", id: endpoint.id }); await notifyChatReceiver(transaction, endpoint); if (endpoint.provider === "discord") { await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); @@ -10,6 +12,11 @@ export async function notifyChatEndpointConfiguration(transaction: object, endpo } export async function notifyChatConnectionConfiguration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { + if (transport === "chat_sdk" || transport === "voice") await notifyVoiceRecovery(transaction, { kind: "connection", id: connectionId }); + await notifyChatConnectionObservation(transaction, connectionId, transport); +} + +export async function notifyChatConnectionObservation(transaction: object, connectionId: string, transport: ToolConnectionTransport) { await notifyChatConnectionReceiver(transaction, connectionId, transport); if (transport === "chat_sdk") { await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId }); @@ -17,6 +24,7 @@ export async function notifyChatConnectionConfiguration(transaction: object, con } export async function notifyChatCompanyConfiguration(transaction: object, companyId: string) { + await notifyVoiceRecovery(transaction, { kind: "company", id: companyId }); await notifyChatCompanyReceivers(transaction, companyId); await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId }); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 8ff37d9b96..4b58afa888 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -44,6 +44,7 @@ export const DELIVERY_QUEUES = { chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", chatReceivers: "chat-receivers", + voiceSessionRecovery: "voice-session-recovery", chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs", diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 7808541872..47fbccb8b1 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyVoiceRecovery } from "./voice/voice-recovery-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import type { Db } from "@paperclipai/db"; @@ -614,6 +615,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); + if (patch.enableChatConnectors !== undefined) await notifyVoiceRecovery(tx); await notifyToolConnectionHealth(tx, "*"); return tx .update(instanceSettings) diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index 8c2a82fb80..a4f1753592 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -1,3 +1,4 @@ +import { notifyVoiceRecovery } from "./voice/voice-recovery-notifications.js"; import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js"; import { connectionHealthSchedule } from "./tool-access/catalog/health-schedule.js"; import type { ConnectionManagementContext } from "./connection-management.js"; @@ -4672,6 +4673,7 @@ export function toolAccessService( return rows; }) : await writeToolConnectionHealthChange(db, existing, async tx => { + if (existing.transport === "voice") await notifyVoiceRecovery(tx, { kind: "connection", id: existing.id }); const rows = await write(tx); await syncCredentialBindings(rows[0]!, [], tx); await createToolRuntimeSlots(tx, runtimeSupervisor).ensureRuntimeSlot(rows[0]!); diff --git a/server/src/services/tool-access/catalog/health.ts b/server/src/services/tool-access/catalog/health.ts index 63567464a5..9de76a44c6 100644 --- a/server/src/services/tool-access/catalog/health.ts +++ b/server/src/services/tool-access/catalog/health.ts @@ -1,4 +1,4 @@ -import { notifyChatConnectionConfiguration } from "../../chat-channels/configuration-work-notifications.js"; +import { notifyChatConnectionObservation } from "../../chat-channels/configuration-work-notifications.js"; import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectionActorUserId } from "../actor.js"; import { assertSupportedConnection } from "../connections/setup-policy.js"; @@ -50,7 +50,7 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d ) { const now = new Date(); const [updated] = await writeToolConnectionHealthChange(db, connection, async tx => { - await notifyChatConnectionConfiguration(tx, connection.id, connection.transport); + await notifyChatConnectionObservation(tx, connection.id, connection.transport); return tx.update(toolConnections) .set({ healthStatus: status, diff --git a/server/src/services/tool-access/connections/removal.ts b/server/src/services/tool-access/connections/removal.ts index 6c4c61f494..b241998e37 100644 --- a/server/src/services/tool-access/connections/removal.ts +++ b/server/src/services/tool-access/connections/removal.ts @@ -1,3 +1,4 @@ +import { notifyVoiceRecovery } from "../../voice/voice-recovery-notifications.js"; import { notifyChatEndpointWork } from "../../chat-work-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { toConnection } from "./projection.js"; @@ -541,6 +542,7 @@ async function removeConnection( const archived = await db.transaction(async (tx) => { if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id); + if (connection.transport === "voice") await notifyVoiceRecovery(tx, { kind: "connection", id: connection.id }); await notifyToolConnectionHealth(tx, connection); await tx.select({ id: toolApplications.id }).from(toolApplications) .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId))) diff --git a/server/src/services/voice/voice-inbound-service.ts b/server/src/services/voice/voice-inbound-service.ts index d59755ee84..5d7f2c1a65 100644 --- a/server/src/services/voice/voice-inbound-service.ts +++ b/server/src/services/voice/voice-inbound-service.ts @@ -1,3 +1,4 @@ +import { notifyVoiceRecovery } from "./voice-recovery-notifications.js"; import { logger } from "../../middleware/logger.js"; import { createHash, createHmac, randomInt, randomUUID, timingSafeEqual } from "node:crypto"; import { and, asc, desc, eq, gte, inArray, lte, isNull, sql } from "drizzle-orm"; @@ -150,6 +151,7 @@ export function voiceInboundService(db: Db, options: { } if (event.type === "call.report" || event.status && ["ended", "failed"].includes(event.status)) { if (session) { + await notifyVoiceRecovery(tx, { kind: "session", id: session.id }); await tx.update(chatVoiceSessions).set({state: event.status === "failed" ? "failed" : "ended", endedAt: session.endedAt ?? new Date(), updatedAt: new Date()}).where(eq(chatVoiceSessions.id, session.id)); await tx.insert(chatVoiceReports).values({companyId: row.companyId, sessionId: session.id}).onConflictDoUpdate({target: chatVoiceReports.sessionId, set: {nextCheckAt: new Date()}}); } @@ -184,7 +186,9 @@ export function voiceInboundService(db: Db, options: { const task = await issueSvc.create(row.companyId, {id: taskId, title: "Incoming phone conversation", description: "Conversation with an unverified phone caller. The agent runs under a task-scoped low-trust policy. You are on a live phone call and must attempt to answer as quickly as possible. Immediately post a brief, caller-safe task comment explaining what you are doing, then find the answer and post the result on this task so it can be spoken to the caller. Continue working after the initial update; do not treat it as the final answer. Preserve the task-scoped low-trust policy and ordinary permissions.", status: "todo", priority: "medium", assigneeAgentId: row.assignedAgentId, executionWorkspaceSettings: {mode: "isolated_workspace", workspaceStrategy: {type: "cloud_sandbox"}, ...(line.lowTrustEnvironmentId ? {environmentId: line.lowTrustEnvironmentId} : {})}, originKind: "chat_channel", originId: row.id, idempotencyKey: `guest-intake:${call.id}`, sourceTrust: {preset: LOW_TRUST_REVIEW_PRESET, disposition: "quarantined", sourceIssueId: taskId}, executionPolicy: {mode: "normal", commentRequired: true, stages: [], reviewPreset, authorizationPolicy: {trustPreset: LOW_TRUST_REVIEW_PRESET, reviewPreset, trustBoundary: {mode: LOW_TRUST_REVIEW_PRESET, companyId: row.companyId, rootIssueId: taskId, issueIds: [taskId], allowedAgentIds: [row.assignedAgentId], allowedToolClasses: [], allowedSecretBindingIds: []}}}}, tx); const [resource] = await tx.insert(chatEndpointResources).values({companyId: row.companyId, endpointId: row.id, type: "direct_message", providerResourceId: threadId, label: "Incoming phone conversation", enabled: true, availability: "available"}).returning(); await tx.insert(chatConversations).values({id: conversationId, companyId: row.companyId, endpointId: row.id, resourceId: resource.id, issueId: task.id, externalConversationId: threadId, externalThreadId: threadId, externalLabel: "Incoming phone conversation", isDirectMessage: true, communicationGuidance: [buildChatCommunicationGuidance({ provider: "speko", isDirectMessage: true, communicationInstructions: row.communicationInstructions }), "This conversation is its own low-trust task. Ask clarifications as brief task comments; the caller answers through submit_request follow-ups. Do not use ask_user_questions or create a protected human-input wait. Use only context supplied in this conversation; do not browse other company tasks, secrets or private tools. Spoken consent does not grant governed authority."].join("\n\n")}); - const [session] = await tx.insert(chatVoiceSessions).values({companyId: row.companyId, endpointId: row.id, conversationId, issueId: task.id, assignedAgentId: row.assignedAgentId, callerId: `guest:${call.id}`, callerAuthority: "guest_intake", mode: "inbound_phone", state: "active", generation: call.generation, credentialFingerprint: call.credentialFingerprint, providerSessionId: call.providerSessionId, toolTokenHash: call.toolTokenHash, idempotencyKey: `guest:${call.id}`, requestFingerprint: call.requestFingerprint, createdAt: call.createdAt, expiresAt: call.expiresAt}).returning({id: chatVoiceSessions.id}); + const sessionId = randomUUID(); + await notifyVoiceRecovery(tx, { kind: "session", id: sessionId }); + const [session] = await tx.insert(chatVoiceSessions).values({id: sessionId, companyId: row.companyId, endpointId: row.id, conversationId, issueId: task.id, assignedAgentId: row.assignedAgentId, callerId: `guest:${call.id}`, callerAuthority: "guest_intake", mode: "inbound_phone", state: "active", generation: call.generation, credentialFingerprint: call.credentialFingerprint, providerSessionId: call.providerSessionId, toolTokenHash: call.toolTokenHash, idempotencyKey: `guest:${call.id}`, requestFingerprint: call.requestFingerprint, createdAt: call.createdAt, expiresAt: call.expiresAt}).returning({id: chatVoiceSessions.id}); await tx.update(chatVoiceInboundCalls).set({intakeIssueId: task.id, sessionId: session.id, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); await tx.insert(activityLog).values({companyId: row.companyId, actorType: "system", actorId: "speko-inbound", action: "voice.low_trust_call.started", entityType: "issue", entityId: task.id, details: {callId: call.id, sessionId: session.id}}); } @@ -228,7 +232,10 @@ export function voiceInboundService(db: Db, options: { if (input.issueId) await store.authorizeTask(tx, companyId, input.issueId, caller); const [changed] = await tx.update(chatVoiceInboundCalls).set({state: input.approve ? "approving" : row.state === "guest_intake" ? "ended" : "denied", approvedByUserId: caller.id, callerAuthority: caller.authority, requestedIssueId: input.issueId ?? row.intakeIssueId ?? null, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, row.id)).returning(); await tx.insert(activityLog).values({companyId, actorType: "user", actorId: caller.id, action: input.approve ? "voice.inbound.approval_started" : row.state === "guest_intake" ? "voice.low_trust_call.ended" : "voice.inbound.denied", entityType: "voice_inbound_call", entityId: row.id}); - if (!input.approve && row.state === "guest_intake" && row.sessionId) await tx.update(chatVoiceSessions).set({state: "ending", updatedAt: new Date()}).where(and(eq(chatVoiceSessions.companyId, companyId), eq(chatVoiceSessions.id, row.sessionId))); + if (!input.approve && row.state === "guest_intake" && row.sessionId) { + await notifyVoiceRecovery(tx, { kind: "session", id: row.sessionId }); + await tx.update(chatVoiceSessions).set({state: "ending", updatedAt: new Date()}).where(and(eq(chatVoiceSessions.companyId, companyId), eq(chatVoiceSessions.id, row.sessionId))); + } return changed; }); if (admission.state === "approved") return projection(admission); diff --git a/server/src/services/voice/voice-recovery-notifications.ts b/server/src/services/voice/voice-recovery-notifications.ts new file mode 100644 index 0000000000..c814f221e1 --- /dev/null +++ b/server/src/services/voice/voice-recovery-notifications.ts @@ -0,0 +1,20 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; + +export type VoiceRecoveryScope = { kind: "session" | "endpoint" | "connection" | "company"; id: string }; +export function voiceRecoveryScope(hint: string): VoiceRecoveryScope | undefined { + if (hint === "*") return undefined; + const [kind, id, extra] = hint.split(":"); + if (!id || extra !== undefined || !["session", "endpoint", "connection", "company"].includes(kind)) + throw new Error("Invalid voice recovery scope"); + return { kind: kind as VoiceRecoveryScope["kind"], id }; +} +export function notifyVoiceRecovery(tx: object, scope?: VoiceRecoveryScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceSessionRecovery, scope ? `${scope.kind}:${scope.id}` : "*"); +} +export function withVoiceSessionChange(db: Db, id: string, write: (tx: Parameters[0]>[0]) => PromiseLike): Promise { + return db.transaction(async tx => { + await notifyVoiceRecovery(tx, { kind: "session", id }); + return write(tx); + }); +} diff --git a/server/src/services/voice/voice-recovery-work.ts b/server/src/services/voice/voice-recovery-work.ts new file mode 100644 index 0000000000..9b98e0cb13 --- /dev/null +++ b/server/src/services/voice/voice-recovery-work.ts @@ -0,0 +1,106 @@ +import { takeCompanyWorkBatch } from "../company-work-batch.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { VoiceRecoveryEntry, voiceSessionRecovery } from "./voice-session-recovery.js"; +import { voiceRecoveryScope, type VoiceRecoveryScope } from "./voice-recovery-notifications.js"; + +function matches(entry: VoiceRecoveryEntry, scope?: VoiceRecoveryScope) { + return !scope || (scope.kind === "session" ? entry.id : scope.kind === "endpoint" ? entry.endpointId : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id; +} +export function registerVoiceRecoveryWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + service.reset(); + const dirty = new Set(["*"]); + const versions = new Map(); + const scans = new Map }>(); + const entries = new Map(); + const failures = new Map(); + const scopeFailures = new Map(); + let earliest = 0; + let lastCompany: string | null = null; + function failed(map: typeof failures, id: string) { + const count = Math.min((map.get(id)?.count ?? 0) + 1, 5); + map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) }); + } + function dirtyScopes() { + return [...dirty].filter(hint => hint !== "*" || !scopeFailures.has(hint)) + .filter(hint => !scans.has(hint) || scopeFailures.has(hint)) + .map(hint => { try { return voiceRecoveryScope(hint); } catch { return undefined; } }); + } + function next() { + let nextAt: number | null = null; + const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; + for (const hint of dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); + const unavailable = dirtyScopes(); + for (const entry of entries.values()) if (!unavailable.some(scope => matches(entry, scope))) + take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); + return nextAt; + } + const worker = coordinator.register(DELIVERY_QUEUES.voiceSessionRecovery, { + retryMs: 30_000, + pausePassiveMaintenance: true, + onSettled: hints => { for (const hint of hints ?? ["*"]) { + dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1); + } }, + run: async signal => { + if (!canRun() || Date.now() < earliest) return; + earliest = Date.now() + 1000; + const errors: unknown[] = []; + const ready = [...dirty].filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); + const requested = ready.includes("*") + ? ["*", ...ready.filter(hint => hint !== "*")].slice(0, 25) + : ready.slice(0, 25); + for (const hint of requested) dirty.delete(hint); + for (let index = 0; index < requested.length; index++) { + if (signal.aborted || !canRun()) { for (const hint of requested.slice(index)) dirty.add(hint); return; } + const hint = requested[index]; + let scope: VoiceRecoveryScope | undefined; + let decoded = false; + try { + scope = voiceRecoveryScope(hint); decoded = true; + const scan = scans.get(hint) ?? { version: versions.get(hint) ?? 0, seen: new Set() }; + scans.set(hint, scan); + const result = await service.refresh(scope, signal); + for (const entry of result.rows ?? []) { + entries.set(entry.id, entry); + for (const [other, progress] of scans) if (matches(entry, voiceRecoveryScope(other))) progress.seen.add(entry.id); + } + if (!result.complete) { dirty.add(hint); continue; } + if (result.rows !== null) for (const entry of entries.values()) if (matches(entry, scope) && !scan.seen.has(entry.id)) { + entries.delete(entry.id); failures.delete(entry.id); + } + scans.delete(hint); + if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint); + if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint); + } catch (error) { + const retryHint = decoded ? hint : "*"; + if (!decoded) { scans.delete(hint); versions.delete(hint); } + dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error); + } + } + const unavailable = dirtyScopes(); + const due = [...entries.values()].filter(entry => !unavailable.some(scope => matches(entry, scope)) + && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) + .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)); + const batch = takeCompanyWorkBatch(due, 25, lastCompany); lastCompany = batch.lastCompany; + for (const entry of batch.entries) { + if (signal.aborted || !canRun()) break; + try { + const current = await service.check(entry.id, signal); + if (current) entries.set(entry.id, current); else entries.delete(entry.id); + failures.delete(entry.id); + } catch (error) { failed(failures, entry.id); errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, "Voice session recovery could not settle"); + }, + // Retain scopes/deadlines for resume without retaining a settled write's + // idle hold during drain. The coordinator still fences unresolved XIDs. + hasPending: async () => canRun() && dirty.size > 0, + nextRunAt: next, + nextRunAtOnError: next, + }); + return worker; +} diff --git a/server/src/services/voice/voice-reply-push.ts b/server/src/services/voice/voice-reply-push.ts index 6d8eb2da88..a17e256a7a 100644 --- a/server/src/services/voice/voice-reply-push.ts +++ b/server/src/services/voice/voice-reply-push.ts @@ -1,3 +1,4 @@ +import { notifyVoiceRecovery } from "./voice-recovery-notifications.js"; import { createHash } from "node:crypto"; import { and, asc, eq, gt, inArray, isNull, lt, sql } from "drizzle-orm"; import { chatActions, chatEndpoints, chatPublications, chatVoiceReplies, chatVoiceSessions, type Db } from "@paperclipai/db"; @@ -114,6 +115,7 @@ export async function pushVoiceReplies(db: Db, options: { : providerError?.httpStatus === 429 ? "retry" : "failed"; await db.transaction(async tx => { 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 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-session-recovery.ts b/server/src/services/voice/voice-session-recovery.ts new file mode 100644 index 0000000000..bf43afdf21 --- /dev/null +++ b/server/src/services/voice/voice-session-recovery.ts @@ -0,0 +1,164 @@ +import { and, eq, gt, inArray, isNull, sql } from "drizzle-orm"; +import { chatEndpoints, chatVoiceSessions, toolConnections, type Db } from "@paperclipai/db"; +import { HttpError } from "../../errors.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { withVoiceSessionChange, type VoiceRecoveryScope } from "./voice-recovery-notifications.js"; + +export const LIVE_VOICE_STATES = ["reserved", "creating", "creation_unknown", "connecting", "active", "ending"] as const; +type Session = typeof chatVoiceSessions.$inferSelect; +export type VoiceRecoveryEntry = { id: string; companyId: string; endpointId: string; connectionId: string; dueAt: number }; +type Candidate> = { session: T; connectionId: string; endpointStatus: string; enabled: boolean; generation: number }; + +export function voiceSessionRecoveryAt(row: Candidate, featureEnabled: boolean): number { + const session = row.session; + if (!session.providerSessionId) return session.expiresAt.getTime() + 120_001; + const changed = !featureEnabled || !row.enabled || !["active", "verifying"].includes(row.endpointStatus) + || session.generation !== row.generation || session.state === "ending"; + if (changed) return session.errorCode === "cleanup_pending" ? session.updatedAt.getTime() + 30_000 : 0; + if (["connecting", "active"].includes(session.state)) return Math.min(session.expiresAt.getTime(), session.updatedAt.getTime() + 30_000); + return session.expiresAt.getTime(); +} + +export function voiceSessionRecovery(db: Db, operations: { + authorize(session: Session): Promise; + inspect(session: Session): Promise<{ endedAt?: unknown } | null>; + close(session: Session): Promise; +}) { + function sessionConditions(scope?: VoiceRecoveryScope, cursor?: string) { + return and(inArray(chatVoiceSessions.state, [...LIVE_VOICE_STATES]), + scope?.kind === "session" ? eq(chatVoiceSessions.id, scope.id) : undefined, + scope?.kind === "endpoint" ? eq(chatVoiceSessions.endpointId, scope.id) : undefined, + scope?.kind === "connection" ? eq(toolConnections.id, scope.id) : undefined, + scope?.kind === "company" ? eq(chatVoiceSessions.companyId, scope.id) : undefined, + cursor ? gt(chatVoiceSessions.id, cursor) : undefined); + } + async function candidates(scope?: VoiceRecoveryScope, signal?: AbortSignal) { + const rows: Candidate[] = []; + let cursor: string | undefined; + for (;;) { + signal?.throwIfAborted(); + const page = await db.select({ session: chatVoiceSessions, connectionId: toolConnections.id, + endpointStatus: chatEndpoints.status, enabled: toolConnections.enabled, generation: sql`coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0)` }) + .from(chatVoiceSessions) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceSessions.endpointId), eq(chatEndpoints.companyId, chatVoiceSessions.companyId))) + .innerJoin(toolConnections, and(eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatVoiceSessions.companyId))) + .where(sessionConditions(scope, cursor)) + .orderBy(chatVoiceSessions.id).limit(100); + rows.push(...page); + if (page.length < 100) return rows; + cursor = page[page.length - 1].session.id; + } + } + async function load(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise { + const rows = await candidates(scope, signal); + if (!rows.length) return []; + const enabled = (await instanceSettingsService(db).getExperimental()).enableChatConnectors; + return rows.map(row => ({ id: row.session.id, companyId: row.session.companyId, endpointId: row.session.endpointId, + connectionId: row.connectionId, dueAt: voiceSessionRecoveryAt(row, enabled) })); + } + type Parent = { id: string; companyId: string; connectionId: string; signature: string | null }; + type Scan = { phase: "parents" | "sessions"; cursor?: string; parents: Map }; + const parents = new Map(); + const scans = new Map(); + function scopeKey(scope?: VoiceRecoveryScope) { return scope ? `${scope.kind}:${scope.id}` : "*"; } + function parentMatches(row: Parent, scope?: VoiceRecoveryScope) { + return !scope || (scope.kind === "endpoint" ? row.id : scope.kind === "connection" ? row.connectionId : row.companyId) === scope.id; + } + function remember(target: Map, row: Parent) { + const prior = target.get(row.id); + target.set(row.id, prior && prior.signature !== row.signature ? { ...row, signature: null } : row); + } + async function refresh(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise<{ rows: VoiceRecoveryEntry[] | null; complete: boolean }> { + const key = scopeKey(scope); + const scan = scans.get(key) ?? { phase: scope && scope.kind !== "session" ? "parents" as const : "sessions" as const, parents: new Map() }; + signal?.throwIfAborted(); + if (scan.phase === "parents") { + const page = await db.select({ id: chatEndpoints.id, companyId: chatEndpoints.companyId, connectionId: toolConnections.id, + endpointStatus: chatEndpoints.status, enabled: toolConnections.enabled, + generation: sql`coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0)` }) + .from(chatEndpoints).innerJoin(toolConnections, and(eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatEndpoints.companyId))) + .where(and(scope?.kind === "endpoint" ? eq(chatEndpoints.id, scope.id) : undefined, + scope?.kind === "connection" ? eq(toolConnections.id, scope.id) : undefined, + scope?.kind === "company" ? eq(chatEndpoints.companyId, scope.id) : undefined, + scan.cursor ? gt(chatEndpoints.id, scan.cursor) : undefined)) + .orderBy(chatEndpoints.id).limit(100); + const enabled = page.length ? (await instanceSettingsService(db).getExperimental()).enableChatConnectors : false; + const next = new Map(scan.parents); + for (const row of page) remember(next, { id: row.id, companyId: row.companyId, connectionId: row.connectionId, + signature: JSON.stringify([row.connectionId, row.endpointStatus, row.enabled, row.generation, enabled]) }); + signal?.throwIfAborted(); + if (page.length === 100) { + scans.set(key, { phase: "parents", cursor: page[page.length - 1].id, parents: next }); + return { rows: [], complete: false }; + } + const previous = [...parents.values()].filter(row => parentMatches(row, scope)); + if (previous.length === next.size && previous.every(row => row.signature !== null && next.get(row.id)?.signature === row.signature)) { + scans.delete(key); return { rows: null, complete: true }; + } + scans.set(key, { phase: "sessions", parents: next }); + return { rows: [], complete: false }; + } + const page = await db.select({ session: { id: chatVoiceSessions.id, companyId: chatVoiceSessions.companyId, + endpointId: chatVoiceSessions.endpointId, state: chatVoiceSessions.state, providerSessionId: chatVoiceSessions.providerSessionId, + expiresAt: chatVoiceSessions.expiresAt, updatedAt: chatVoiceSessions.updatedAt, generation: chatVoiceSessions.generation, + errorCode: chatVoiceSessions.errorCode }, connectionId: toolConnections.id, + endpointStatus: chatEndpoints.status, enabled: toolConnections.enabled, + generation: sql`coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0)` }) + .from(chatVoiceSessions) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceSessions.endpointId), eq(chatEndpoints.companyId, chatVoiceSessions.companyId))) + .innerJoin(toolConnections, and(eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatVoiceSessions.companyId))) + .where(sessionConditions(scope, scan.cursor)) + .orderBy(chatVoiceSessions.id).limit(100); + const enabled = page.length ? (await instanceSettingsService(db).getExperimental()).enableChatConnectors : false; + const next = new Map(scan.parents); + for (const row of page) remember(next, { id: row.session.endpointId, companyId: row.session.companyId, connectionId: row.connectionId, + signature: JSON.stringify([row.connectionId, row.endpointStatus, row.enabled, row.generation, enabled]) }); + const rows = page.map(row => ({ id: row.session.id, companyId: row.session.companyId, endpointId: row.session.endpointId, + connectionId: row.connectionId, dueAt: voiceSessionRecoveryAt(row, enabled) })); + signal?.throwIfAborted(); + const complete = page.length < 100; + if (!complete) scans.set(key, { phase: "sessions", cursor: page[page.length - 1].session.id, parents: next }); + else { + scans.delete(key); + if (scope?.kind !== "session") { + for (const row of parents.values()) if (parentMatches(row, scope)) parents.delete(row.id); + for (const [id, row] of next) parents.set(id, row); + } + } + return { rows, complete }; + } + function reset() { scans.clear(); parents.clear(); } + async function check(id: string, signal?: AbortSignal): Promise { + const scope = { kind: "session", id } as const; + const [row] = await candidates(scope, signal); + if (!row) return null; + const enabled = (await instanceSettingsService(db).getExperimental()).enableChatConnectors; + if (voiceSessionRecoveryAt(row, enabled) <= Date.now()) { + signal?.throwIfAborted(); + const session = row.session; + try { + let authorized = false; + if (enabled && ["connecting", "active"].includes(session.state) && session.expiresAt.getTime() > Date.now()) { + try { await operations.authorize(session); authorized = true; } + catch (error) { if (!(error instanceof HttpError && [403, 404].includes(error.status))) throw error; } + } + if (authorized) { + const state = await operations.inspect(session); + await withVoiceSessionChange(db, id, tx => tx.update(chatVoiceSessions).set({ + ...(state?.endedAt ? { state: "ended" as const, endedAt: new Date(), errorCode: null } : {}), updatedAt: new Date(), + }).where(and(eq(chatVoiceSessions.id, id), inArray(chatVoiceSessions.state, ["connecting", "active"])))); + } else if (session.providerSessionId) await operations.close(session); + else if (session.expiresAt.getTime() + 120_000 < Date.now()) { + await withVoiceSessionChange(db, id, tx => tx.update(chatVoiceSessions).set({ state: "expired", endedAt: new Date(), updatedAt: new Date() }) + .where(and(eq(chatVoiceSessions.id, id), isNull(chatVoiceSessions.providerSessionId), inArray(chatVoiceSessions.state, [...LIVE_VOICE_STATES])))); + } + } catch (error) { + await withVoiceSessionChange(db, id, tx => tx.update(chatVoiceSessions).set({ errorCode: "cleanup_pending", updatedAt: new Date() }) + .where(and(eq(chatVoiceSessions.id, id), inArray(chatVoiceSessions.state, [...LIVE_VOICE_STATES])))); + throw error; + } + } + return (await load(scope, signal))[0] ?? null; + } + return { load, refresh, reset, check }; +} diff --git a/server/src/services/voice/voice-session-service.ts b/server/src/services/voice/voice-session-service.ts index 59b99c6616..954299bac1 100644 --- a/server/src/services/voice/voice-session-service.ts +++ b/server/src/services/voice/voice-session-service.ts @@ -1,3 +1,7 @@ +import { voiceSessionRecovery } from "./voice-session-recovery.js"; +import { registerVoiceRecoveryWork } from "./voice-recovery-work.js"; +import { notifyVoiceRecovery, withVoiceSessionChange } from "./voice-recovery-notifications.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { logger } from "../../middleware/logger.js"; import { captureVoiceDeliveryDiagnostics } from "./voice-delivery-diagnostics.js"; import { pushVoiceReplies } from "./voice-reply-push.js"; @@ -92,7 +96,7 @@ export function voiceSessionService(db: Db, options: { // browser response must inspect/end this attempt, never blindly create again. if (!reserved.created || input.mode === "inbound_phone") return { session: serializeVoiceSession(reserved.session) }; const session = reserved.session; - const [claim] = await db.update(chatVoiceSessions).set({ state: "creating", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "reserved"))).returning({ id: chatVoiceSessions.id }); + const [claim] = await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ state: "creating", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "reserved"))).returning({ id: chatVoiceSessions.id })); if (!claim) throw conflict("The call was ended before Speko connected"); let client: ReturnType | undefined; let created: { sessionId: string; transportToken?: string; transportUrl?: string }; @@ -107,13 +111,13 @@ export function voiceSessionService(db: Db, options: { : await client.createBrowserSession({ ...shared, systemPrompt: VOICE_PROMPT }); } catch (error) { const known = error instanceof SpekoProviderError && !error.outcomeUnknown || error instanceof HttpError && [403, 404].includes(error.status); - await db.update(chatVoiceSessions).set({ state: known ? "failed" : "creation_unknown", errorCode: error instanceof SpekoProviderError ? error.code : "provider_unavailable", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "creating"))); + await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ state: known ? "failed" : "creation_unknown", errorCode: error instanceof SpekoProviderError ? error.code : "provider_unavailable", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "creating")))); if (error instanceof SpekoProviderError && error.code === "insufficient_credits") throw conflict("Speko needs credits before starting a call. Add credits in Speko, then try again.", {code: "voice_credits_required", sessionId: session.id}); throw conflict(known ? "Speko could not start the call. Check connection status before trying again." : "Speko may have started this call. Inspect this attempt before starting another.", { code: known ? "voice_creation_failed" : "voice_creation_unknown", sessionId: session.id }); } let bound: VoiceSessionRow | undefined; try { - [bound] = await db.update(chatVoiceSessions).set({ providerSessionId: created.sessionId, state: "connecting", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "creating"))).returning(); + [bound] = await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ providerSessionId: created.sessionId, state: "connecting", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "creating"))).returning()); } catch (error) { // If the provider started but persistence failed, stop that exact call. // Never return media or retry creation on an uncertain database outcome. @@ -122,7 +126,7 @@ export function voiceSessionService(db: Db, options: { } if (!bound) { // A concurrent end must still learn the provider identity for reconciliation. - await db.update(chatVoiceSessions).set({ providerSessionId: created.sessionId, state: "ending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); + await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ providerSessionId: created.sessionId, state: "ending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id))); await closeProvider({ ...session, providerSessionId: created.sessionId, state: "ending" }); throw conflict("The call was ended while Speko was connecting"); } @@ -141,6 +145,7 @@ export function voiceSessionService(db: Db, options: { async function closeProviderOnce(session: VoiceSessionRow) { if (terminalStates.some(state => state === session.state)) return session; await db.transaction(async (tx) => { + await notifyVoiceRecovery(tx, { kind: "session", id: session.id }); const [changed] = await tx.update(chatVoiceSessions).set({ state: "ending", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), inArray(chatVoiceSessions.state, [...liveStates].filter((state) => state !== "ending")))).returning({ id: chatVoiceSessions.id }); if (changed) await tx.insert(activityLog).values({ companyId: session.companyId, actorType: "system", actorId: "voice-session-service", action: "voice.session.ending", entityType: "voice_session", entityId: session.id, details: { issueId: session.issueId } }); }); @@ -156,11 +161,11 @@ export function voiceSessionService(db: Db, options: { try { confirmed = Boolean((await client.inspectSession(session.providerSessionId)).endedAt); } catch (error) { if (!(error instanceof SpekoProviderError)) throw error; } } - await db.update(chatVoiceSessions).set(confirmed + await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set(confirmed ? { state: "ended", endedAt: new Date(), errorCode: null, updatedAt: new Date() } - : { errorCode: "cleanup_pending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); + : { errorCode: "cleanup_pending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id))); } else if (session.state === "reserved") { - await db.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); + await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id))); } return db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, session.id)).then((rows) => rows[0]); } @@ -170,7 +175,7 @@ export function voiceSessionService(db: Db, options: { const endpoint = await endpointFor(companyId, session.endpointId), credentials = await options.credentials(endpoint); const state = await provider(credentials.apiKey).inspectSession(session.providerSessionId); if (state.endedAt) { - [session] = await db.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, sessionId)).returning(); + [session] = await withVoiceSessionChange(db, sessionId, tx => tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, sessionId)).returning()); } } return serializeVoiceSession(session); @@ -199,51 +204,30 @@ export function voiceSessionService(db: Db, options: { if (verified.envelope.tool === "answer_question") options.onQuestionAnswered(verified.envelope.args.interactionId); return session.mode === "inbound_phone" ? {...result, authorization: session.authority === "guest_intake" ? "guest_intake" : "approved"} : result; } - async function reconcile(limit = 25) { - // Provider maximum duration is the final bound even if creation response was - // lost. Unknown attempts remain reserved until that bound plus teardown grace. - const enabled = (await instanceSettingsService(db).getExperimental()).enableChatConnectors; - const rows = await db.select({ session: chatVoiceSessions }).from(chatVoiceSessions) - .innerJoin(chatEndpoints, eq(chatEndpoints.id, chatVoiceSessions.endpointId)) - .innerJoin(toolConnections, eq(toolConnections.id, chatEndpoints.connectionId)) - .where(and(inArray(chatVoiceSessions.state, [...liveStates]), or( - enabled ? undefined : sql`true`, - lte(chatVoiceSessions.expiresAt, new Date()), eq(chatVoiceSessions.state, "ending"), - notInArray(chatEndpoints.status, ["active", "verifying"]), eq(toolConnections.enabled, false), - sql`${chatVoiceSessions.generation} <> coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0)`, - // Recheck membership and bindings periodically even without callbacks. - and(inArray(chatVoiceSessions.state, ["connecting", "active"]), lte(chatVoiceSessions.updatedAt, new Date(Date.now() - 30_000))), - ))).orderBy(asc(chatVoiceSessions.updatedAt)).limit(limit); - for (const { session } of rows) { - try { - let authorized = false; - if (enabled && ["connecting", "active"].includes(session.state) && session.callerAuthority === "guest_intake" && session.expiresAt.getTime() > Date.now()) { - authorized = await inbound.authorizeGuest(session.id); - } else if (enabled && ["connecting", "active"].includes(session.state)) { - try { await db.transaction((tx) => store.authorizeSession(tx, session.companyId, session.id)); authorized = true; } catch { /* revoked, changed binding, or expired */ } - } - if (authorized) { - // A caller can hang up without a final browser request. Poll provider - // state as well as permissions so a missed callback releases the task. - const endpoint = await endpointFor(session.companyId, session.endpointId); - const credentials = await options.credentials(endpoint); - const state = session.providerSessionId ? await provider(credentials.apiKey).inspectSession(session.providerSessionId) : null; - await db.update(chatVoiceSessions).set({ - ...(state?.endedAt ? { state: "ended", endedAt: new Date(), errorCode: null } : {}), - updatedAt: new Date(), - }).where(and(eq(chatVoiceSessions.id, session.id), inArray(chatVoiceSessions.state, ["connecting", "active"]))); - continue; - } - if (session.providerSessionId) await closeProvider(session); - else if (session.expiresAt.getTime() + 120_000 < Date.now()) await db.update(chatVoiceSessions).set({ state: "expired", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); - } catch { - await db.update(chatVoiceSessions).set({ errorCode: "cleanup_pending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); - } - } + const recovery = voiceSessionRecovery(db, { + authorize: async session => { + await db.transaction(tx => store.authorizeSession(tx, session.companyId, session.id)); + }, + inspect: async session => { + const endpoint = await endpointFor(session.companyId, session.endpointId); + const credentials = await options.credentials(endpoint); + return session.providerSessionId ? provider(credentials.apiKey).inspectSession(session.providerSessionId) : null; + }, + close: closeProvider, + }); + function registerRecoveryWork(coordinator: ReturnType, canRun: () => boolean) { + return registerVoiceRecoveryWork(coordinator, recovery, canRun); + } + async function reconcileMaintenance(limit = 25) { await pushReplies(limit); await inbound.reconcile(limit); await reconcileReports(limit); - return rows.length; + } + async function reconcile(limit = 25) { + const due = (await recovery.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 due) await recovery.check(row.id).catch(() => undefined); + await reconcileMaintenance(limit); + return due.length; } async function pushReplies(limit = 25, companyId?: string, endpointId?: string) { return pushVoiceReplies(db, {store, credentials: options.credentials, provider, limit, companyId, endpointId}); @@ -355,7 +339,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, pushReplies, inbound, notification: store.notification, callbackPreference, saveCallbackPreference, history, callDetail, unregisterAgentTools }; + return { start, inspect, end, tool, reconcile, reconcileMaintenance, registerRecoveryWork, pushReplies, inbound, notification: store.notification, callbackPreference, saveCallbackPreference, history, callDetail, unregisterAgentTools }; } export type VoiceSessionService = ReturnType; diff --git a/server/src/services/voice/voice-session-store.ts b/server/src/services/voice/voice-session-store.ts index 7a0ce846ac..9ab8379a22 100644 --- a/server/src/services/voice/voice-session-store.ts +++ b/server/src/services/voice/voice-session-store.ts @@ -1,3 +1,4 @@ +import { notifyVoiceRecovery } from "./voice-recovery-notifications.js"; import { notifyChatDeliveryWork } from "../chat-work-notifications.js"; import { environmentService } from "../environments.js"; import { logger } from "../../middleware/logger.js"; @@ -192,13 +193,16 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n if (!["connecting", "active"].includes(guest.state) || guest.expiresAt.getTime() <= Date.now() || guest.generation !== context.generation || guest.toolTokenHash !== voiceTokenHash(input.toolToken ?? "")) throw forbidden("Guest call is no longer available for approval"); const task = await authorizeTask(tx, input.companyId, binding.issueId, input.caller); if (task.assigneeAgentId !== context.endpoint.assignedAgentId) throw conflict("The selected agent must be assigned to this task"); + await notifyVoiceRecovery(tx, { kind: "session", id: guest.id }); const [promoted] = await tx.update(chatVoiceSessions).set({...binding, callerId: input.caller.id, callerAuthority: input.caller.authority, approvedByUserId: input.caller.id, idempotencyKey: input.idempotencyKey, requestFingerprint: input.requestFingerprint, updatedAt: new Date()}).where(eq(chatVoiceSessions.id, guest.id)).returning(); await tx.insert(activityLog).values({companyId: input.companyId, actorType: "user", actorId: input.caller.id, action: "voice.guest.private_call_approved", entityType: "voice_session", entityId: guest.id, details: {issueId: binding.issueId}}); return {session: promoted, created: false as const}; } } const token = input.toolToken ?? randomBytes(32).toString("base64url"), createdAt = new Date(); - const [session] = await tx.insert(chatVoiceSessions).values({ id: randomUUID(), companyId: input.companyId, endpointId: input.endpointId, ...binding, assignedAgentId: context.endpoint.assignedAgentId, + const sessionId = randomUUID(); + await notifyVoiceRecovery(tx, { kind: "session", id: sessionId }); + const [session] = await tx.insert(chatVoiceSessions).values({ id: sessionId, companyId: input.companyId, endpointId: input.endpointId, ...binding, assignedAgentId: context.endpoint.assignedAgentId, callerId: input.caller.id, callerAuthority: input.caller.authority, mode: input.mode ?? "browser", state: input.providerSessionId ? "connecting" : "reserved", providerSessionId: input.providerSessionId ?? null, approvedByUserId: input.mode === "inbound_phone" ? input.caller.id : null, generation: context.generation, credentialFingerprint: context.fingerprint, toolTokenHash: voiceTokenHash(token), idempotencyKey: input.idempotencyKey, requestFingerprint: input.requestFingerprint, createdAt, expiresAt: new Date(createdAt.getTime() + input.maxDurationSeconds * 1000) }).returning(); await tx.insert(activityLog).values({ companyId: input.companyId, actorType: "user", actorId: input.caller.id, action: "voice.session.requested", entityType: "voice_session", entityId: session.id, details: { endpointId: input.endpointId, issueId: session.issueId, mode: input.mode ?? "browser" } }); @@ -269,6 +273,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n await tx.insert(chatVoiceReplies).values({ companyId: session.companyId, sessionId: session.id, publicationId: publication.id, cursor: ++cursor }); } if (cursor !== session.replyCursor) { + await notifyVoiceRecovery(tx, { kind: "session", id: session.id }); await tx.update(chatVoiceSessions).set({ replyCursor: cursor, updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); session.replyCursor = cursor; } @@ -290,6 +295,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n if (existing) return; const cursor = session.replyCursor + 1; await tx.insert(chatVoiceReplies).values({ companyId: session.companyId, sessionId: session.id, publicationId: previous.publicationId, cursor, deliveredAt: new Date() }); + await notifyVoiceRecovery(tx, { kind: "session", id: session.id }); await tx.update(chatVoiceSessions).set({ replyCursor: cursor, updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); session.replyCursor = cursor; } @@ -327,6 +333,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n const context = await authorizeSession(tx, input.companyId, input.sessionId), { session } = context; if (session.endpointId !== input.endpointId || session.providerSessionId !== input.envelope.session_id || !timingSafeEqual(Buffer.from(session.toolTokenHash, "hex"), Buffer.from(voiceTokenHash(input.token), "hex"))) throw forbidden("Voice tool scope does not match this session"); if (session.state === "connecting") { + await notifyVoiceRecovery(tx, { kind: "session", id: session.id }); await tx.update(chatVoiceSessions).set({ state: "active", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); await tx.insert(activityLog).values({ companyId: session.companyId, actorType: "system", actorId: "voice-session-service", action: "voice.session.connected", entityType: "voice_session", entityId: session.id, details: { endpointId: session.endpointId, issueId: session.issueId } }); } @@ -463,6 +470,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n if (priorCall?.blocked) return; const cursor = session.replyCursor + 1; await tx.insert(chatVoiceReplies).values({ companyId, sessionId: session.id, publicationId: publication.id, cursor }); + await notifyVoiceRecovery(tx, { kind: "session", id: session.id }); await tx.update(chatVoiceSessions).set({ replyCursor: cursor, updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); return {sessionId: session.id, providerSessionId: session.providerSessionId, replyCursor: cursor, mode: session.mode}; }).then(queued => {