paperclipai/paperclip
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-07 06:16:06 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

fix(server): schedule live voice recovery from session changes and deadlines (#17576)

Browse Source 
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Voice calls retain durable session state and provider identities.
> - The chat maintenance timer scans live voice sessions even when no
call exists.
> - Session writes, saved expiry and active-call inspection deadlines
identify real work.
> - This pull request moves live-session recovery to the shared
coordinator.
> - Idle session recovery becomes quiet without losing exact-call
cleanup or current authorization checks.

## Linked Issues or Issue Description

Refs #17572. This PR is stacked on receiver discovery. It owns only live
voice session recovery. Reply delivery, inbound admission cleanup,
inbound history retention and call reports remain separate migration
slices.

**What happened?**
The one-second chat maintenance lane reads voice settings and scans live
sessions even when an instance has no calls or only terminal history.

**What did you expect?**
Reconstruct live calls on startup. Wake on committed session or
configuration changes. Inspect active calls and retry cleanup at actual
deadlines. Stop querying when no live obligation remains.

**Steps to reproduce**
Start an instance with no live voice sessions. Let startup settle. The
old maintenance method continues to read settings and session inventory
every chat maintenance pass.

## What Changed

- Add a voice-owned bounded deadline reader and scoped worker on the
existing coordinator. Retain concurrent writes across pages and reset
partial scans on restart. No tables or migrations are added.
- Separate generic connection health observations from configuration
changes. Existing receiver and command consumers retain their
notifications. Voice eligibility and provider operations remain in the
voice owner.
- Signal existing session transactions and configuration/feature
changes. Keep provider calls outside database transactions. Standalone
session writes use one small transaction helper.
- Preserve saved expiry, the two-minute unknown-creation grace,
active-call authorization and missed-hangup inspection, and
exact-provider cleanup. Recovery never creates a provider call.
- Retry transient authority lookup failures instead of treating them as
confirmed revocation.
- Fence expiration against a provider identity bound after discovery. A
late known call remains eligible for cleanup.
- Preserve work through rollback, drain, restart, concurrent refresh,
backoff and shutdown. Page inventory and rotate bounded batches across
companies.
- Leave the four other voice maintenance duties in the periodic lane.
- Document producer coverage and the supported writer contract in
`doc/idle-voice-session-recovery.md` and the dated audit.

## Verification

Published head: `f0b2e3cc22a18e7d08b10d4fc02aae4a19a52f30`. Receiver
#17572 has landed in master at
`dce004259394486f6361d0c0dd5a33a5272acc0a`. This normal merge reconciles
squash ancestry; the complete 22-file voice recovery patch is
byte-for-byte identical to the reviewed and tested `ad79c7e805` patch,
while all unrelated master changes are preserved. The PR now targets
master. Fresh exact-head CI and Greptile are running; interrupted
Ponytail turns are not completed verdicts under the maintainer’s
clarified merge policy, and an actual failing verdict still blocks.

- **210 distinct targeted cases pass on the current repair**: 19
complete recovery lifecycle cases and 191 complete voice-store,
coordinator, batching, startup and hostname cases. Production server
typecheck and build pass. Later edits record evidence only.
- CI `40633667141` on prior head `1b6323c164` failed the feature-change
assertion with `34050 < 25000`; its verification aggregate also failed.
The test now disables automatically advancing fake time, drives 20
controlled scheduler turns and requires all 205 calls to close before an
unrelated inspection deadline can mask a lost wake. Removing generation
replay fails with missing closures. The repair does not benchmark
database/host speed or raise the elapsed-time bound.
- A real committed exact-call regression reproduces wildcard-backoff
starvation. The worker keeps the broad retry but processes independent
exact hints and cached due calls. Each check reloads current state and
authority. Failed exact scopes remain fenced. Removing wildcard
isolation fails the intended assertion; a successful broad retry then
leaves the worker SQL-quiet. Full-row and lean queries now share the
state/scope/cursor predicate.
- Earlier acceptance on `712f3862b5` passed 725 distinct targeted cases.
Its measurements remain useful history: five real metadata writes and
one health check over 205 future calls selected 1,230 rows before the
inventory repair and zero afterward, without changing the saved
deadline. Those older test counts are not added to the current 210-case
count.
- Both external voice fixtures self-migrate; the current full
voice-store run uses a separate initially empty database. Earlier
fixture/method/observer failures, source mutations and partial edit
attempts remain in the [current
audit](https://github.com/paperclipai/paperclip/blob/ad79c7e805367fc7e2df3ea53c942ca6c7d27365/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md#wildcard-failure-and-deterministic-generation-acceptance-october-28).
Source is restored after guarded mutations.
- Older CI `40632057899` failed because its trusted policy script was
absent from the checkout, while build separately received a runner
shutdown signal. E2E shards succeeded but their aggregate failed on
policy. The current checkout includes the landed policy tooling and its
checker/tests pass locally. The shutdown cause is unproven; neither old
failed run is counted as passing. Earlier full-local attempts stopped on
embedded PostgreSQL setup failures and did not pass.
- Fresh exact-head Greptile, security and all CI must pass before merge.
A completed failing Ponytail verdict blocks; an interrupted turn without
a verdict does not. No full current local repository matrix, live
provider acceptance, multi-writer support, Cloud wake, host suspension
or savings is claimed.

## Risks

- Process-local hints cover the existing active writer and shared
clients. Independent SQL writers or another replica need explicit wake
integration.
- Membership, task binding, phone preferences and other authority inputs
remain checked on the existing 30-second active-call inspection deadline
and on use. This is not an instant cleanup guarantee for every authority
mutation.
- Unknown creation retains its existing maximum-duration grace. Cleanup
failures retain retries. A provider request is never replayed merely
because its creation response is unknown.
- Genuine configuration changes retain bounded reconstruction. Metadata
writes can probe parent eligibility, while health-only observations
leave voice recovery quiet. Paging state is process-local and
reconstructs from durable rows after restart.
- Active calls, the other voice maintenance owners and connector
transports remain real obligations. This does not expand Cloud alarm
coverage, relax sleep eligibility or prove deployed suspension.
- The receiver and Discord parents have landed. Current-master
compatibility and exact-head CI and Greptile remain merge gates.

## Model Used

OpenAI GPT-6 via Codex, with repository inspection, reasoning, code
execution and test tools. The runtime does not expose a more specific
deployment identifier or context-window size.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either linked existing issues or described the issue in-PR
following the relevant issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links
- [x] My branch name describes the change and contains no internal
ticket identifier
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <[email protected]>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-28 07:38:48 -05:00
1 parent dce0042593
commit e1ba77f067
22 files changed
+1403 -64
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/idle-voice-session-recovery.md
+184
View File
@@ -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.
doc/plans/2026-10-24-recurring-server-sleep-stack.md
+59
View File
@@ -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.
doc/plans/2026-10-25-voice-drain-repair.md
+34
View File
@@ -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.
doc/plans/2026-10-25-voice-maintenance-sleep-audit.md
+297
View File
@@ -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.
server/src/__tests__/app-private-hostname-gate.test.ts
+8 -1
View File
@@ -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(
server/src/__tests__/voice-recovery-work.test.ts
+356
View File
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>, companyId: string, endpointId: string, agentId: string, issueId: string, conversationId: string;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | 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<typeof chatVoiceSessions.$inferInsert> = {}, 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<string>();
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<void>(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);
});
});
server/src/__tests__/voice-session-store.integration.test.ts
+99 -6
View File
@@ -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<typeof createDb>;
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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()}});
server/src/app.ts
+2 -1
View File
@@ -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,
server/src/services/access.ts
+2
View File
@@ -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,
server/src/services/chat-channels/configuration-work-notifications.ts
+8
View File
@@ -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 });
}
server/src/services/delivery-work-notifications.ts
+1
View File
@@ -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",
server/src/services/instance-settings.ts
+2
View File
@@ -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)
server/src/services/tool-access.ts
+2
View File
@@ -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]!);
server/src/services/tool-access/catalog/health.ts
+2 -2
View File
@@ -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,
server/src/services/tool-access/connections/removal.ts
+2
View File
@@ -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)))
server/src/services/voice/voice-inbound-service.ts
+9 -2
View File
@@ -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);
server/src/services/voice/voice-recovery-notifications.ts
+20
View File
@@ -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<T>(db: Db, id: string, write: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => PromiseLike<T>): Promise<T> {
return db.transaction(async tx => {
await notifyVoiceRecovery(tx, { kind: "session", id });
return write(tx);
});
}
server/src/services/voice/voice-recovery-work.ts
+106
View File
@@ -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<typeof createDeliveryWorkCoordinator>,
service: ReturnType<typeof voiceSessionRecovery>,
canRun: () => boolean,
) {
service.reset();
const dirty = new Set<string>(["*"]);
const versions = new Map<string, number>();
const scans = new Map<string, { version: number; seen: Set<string> }>();
const entries = new Map<string, VoiceRecoveryEntry>();
const failures = new Map<string, { count: number; retryAt: number }>();
const scopeFailures = new Map<string, { count: number; retryAt: number }>();
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<string>() };
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;
}
server/src/services/voice/voice-reply-push.ts
+2
View File
@@ -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));
});
server/src/services/voice/voice-session-recovery.ts
+164
View File
@@ -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<T = Pick<Session, "id" | "companyId" | "endpointId" | "providerSessionId" | "expiresAt" | "updatedAt" | "generation" | "state" | "errorCode">> = { 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<void>;
inspect(session: Session): Promise<{ endedAt?: unknown } | null>;
close(session: Session): Promise<unknown>;
}) {
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<Session>[] = [];
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<number>`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<VoiceRecoveryEntry[]> {
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<string, Parent> };
const parents = new Map<string, Parent>();
const scans = new Map<string, Scan>();
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<string, Parent>, 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<string, Parent>() };
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<number>`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<number>`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<VoiceRecoveryEntry | null> {
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 };
}
server/src/services/voice/voice-session-service.ts
+35 -51
View File
@@ -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<typeof createSpekoProvider> | 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<typeof createDeliveryWorkCoordinator>, 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<typeof voiceSessionService>;
server/src/services/voice/voice-session-store.ts
+9 -1
View File
@@ -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 => {