fix(server): schedule inbound voice cleanup from durable obligations (#17580)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Incoming voice calls need approval, expiry, and safe provider
cleanup.
> - The remaining voice maintenance tick scans admissions every second,
including idle endpoints.
> - Admission status does not prove that the provider call has ended.
> - Existing admission rows can retain the cleanup owner, confirmation,
and retry deadline.
> - This pull request replaces the cleanup scan with committed changes
and actual deadlines.
> - Idle cleanup stops querying while unfinished calls keep a
recoverable path to termination.

## Linked Issues or Issue Description

**What existing behavior does this improve?**

Inbound voice admission cleanup. Its reply-delivery parent #17578 has
landed; this PR targets master. Live-session recovery is in #17576. It
preserves separate ownership for those services.

**Current behavior**

The voice maintenance tick scans expired admissions even when no cleanup
is due. A session can commit before the separate approved-admission
write. Expiry can then terminate that authorized call. A rejected guest
call can be labeled ended before provider termination succeeds.
Seven-day retention can delete that unfinished cleanup.

**Proposed behavior**

Wake cleanup from committed admission and session changes. Schedule the
original expiry or saved retry. Claim current state under the admission
lock before provider I/O. Commit approval with session reservation. Keep
unconfirmed termination durable and transfer bound calls to live-session
recovery.

**Reason and benefit**

Configured idle endpoints and confirmed history cause no recurring
admission cleanup SQL or provider requests. Approval, retries, crash
recovery, and cleanup remain supported. History retention and call
reports stay separate migration items.

## What Changed

- Recover terminal and rejected legacy admissions whose session binding
was lost. Copy known termination or bind a proven matching live session
for cleanup without granting approval.
- Register inbound cleanup with the shared coordinator and scoped
deadline worker. Yield after each 100-row scope page and reset cursors
on registration. Remove only cleanup from periodic voice maintenance.
Keep complete manual reconciliation.
- Add commit-aware notifications at admission, approval, rejection,
guest binding, pending-tool expiry, lifecycle confirmation, and session
termination writes.
- Commit authorized session binding and admission approval together.
Recheck cleanup ownership under the same admission lock. Recover older
interrupted bindings only from matching session evidence.
- Add nullable cleanup confirmation, next-attempt, and provider-account
fields to the existing admission table. No new queue table is added.
Generated migration 0396 backfills only provable historical termination
in committed 1,000-row keyset batches and adds a concurrent partial
index.
- Save the retry before provider contact. An unconfirmed end receipt
retains cleanup. Bound calls remain owned by live-session recovery.
- Verify the signed provider account before using a rotated key for
unbound cleanup. A different account cannot inspect or end the saved
call. Legacy admissions require their original credential fingerprint.
- Prevent history pruning and company deletion from erasing unfinished
cleanup. Company deletion locks endpoints before the voice owner's
check, matching ingress serialization.
- Document producer coverage, ownership, deadlines, verification, and
remaining sleep limits.

## Verification

Published head: `a4e7ccf976fc5793a8914e4e4861aa080d43a4da`. The reply
parent #17578 has landed. A normal merge integrates actual master
`e7ea2e2961863171956e70fb05d3d1f3790f7909`, preserves its unrelated
changes, and regenerates this slice’s migration after landed migration
0395. Fresh exact-head reviews and CI are required.

- The previous Ponytail 3/5 finding is repaired: legacy terminal
admissions no longer enter approval-only repair, and rejected admissions
with valid matching live sessions retain their rejection and persist the
binding needed for session-scoped cleanup confirmation. Five regressions
failed with ownership conflicts before the repair. Five
conflicting-session regressions prevent an unrelated leg from being
claimed. No cleanup operation grants approval or starts a provider call.
- **24 complete cleanup lifecycle cases and 122 complete voice-store
integration cases pass** on the repair. After actual-master integration
and migration regeneration, the complete 24-case cleanup suite passes
again on a new empty PostgreSQL 17 database, and the database migration
integration test passes separately. The cleanup owner, voice-store test
and admission schema are byte-identical to that tested repair; the
cleanup fixture only changes the migration filename. Migration
numbering/safety, database build, production server typecheck and build
pass.
- Migration `0396_tan_paladin.sql` was generated with `pnpm
db:generate`. Every landed SQL migration and the complete parent journal
prefix are unchanged. Generated snapshots retain exactly 0392–0396,
pruning 0391. The reviewed cleanup SQL, concurrent idempotent index and
committed 1,000-row historical backfill are retained byte for byte under
the new generated filename, registered for top-level execution. The real
application migrator boots a fresh database, repairs an interrupted
concurrent index, serializes replays and verifies eligible-history
backfill across actual transactions.
- The preceding published head
`48d8356e73308579470c182e759eead6e83173d2` received Greptile 5/5 but
failed CI server shard 9/12: one existing connected MCP rate-limit test
expected two real-clock calls to share a fixed 60-second bucket. A
controlled window rollover reproduced its exact failure. That existing
fixture now freezes only `Date` around the same-window 429 assertion,
verifies acceptance after advancing one window, and restores the real
clock in `finally`. All 92 gateway acceptance cases pass. Production
rate limiting, timeouts and assertions are unchanged. This adds one
existing CI fixture to the slice; the original failed run remains
recorded.
- The first regression invocation skipped the suite because it used the
wrong database variable; that is not a pass. The corrected invocation
reproduced five ownership conflicts. The first fixed cleanup run passed
22 and failed two new low-level deletion test calls on a
retained-session foreign key; the regressions now check the cleanup
deletion guard at its owning boundary, with production deletion ordering
unchanged. The final full suite passes all 24.
- The first complete voice-store invocation passed 121 and timed out on
one transcript case at the unchanged 15-second limit. That case passed
in isolation on a fresh database. One subsequent complete invocation,
with builds finished, passed all 122 without source, assertion or
timeout changes. The underlying timeout cause is unproven; the original
failure remains recorded. A new PostgreSQL startup also failed from
exhausted host semaphore capacity; fresh databases on the already-owned
server were used without removing shared IPC resources.
- Prior source `b8b12199b9` passed 266 distinct cases and carried three
guarded mutation proofs; those older results are not added to this
repair’s case count. Its CI run 40641168596 failed E2E shard 3 when
legacy cancellation hit a one-second PostgreSQL lock timeout. GitHub
initially rejected a retry while the run was active, then accepted one
targeted retry after completion; attempt 2 passed on that same old head.
The lock holder and underlying cause remain unproven. The unchanged Node
policy checker’s older failure on ignored generated Copilot manifests
and passing tracked-input mirror are also retained in the audit; no
workflow, checker, vendor manifest or ignore rule was modified.
- [Current repair and failure
record](https://github.com/paperclipai/paperclip/blob/a4e7ccf976fc5793a8914e4e4861aa080d43a4da/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md#legacy-terminal-admissions-missing-their-session-binding--october-28).
Current-head CI and Greptile 5/5 gate merge. A completed failing
Ponytail verdict must be fixed; missing or interrupted reviews do not
block, and active pending review waits are capped at 30 minutes under
the maintainer’s explicit instruction. No new full local repository
suite, real provider, Cloud alarm, host/database suspension or savings
claim is made.

## Risks

- The migration distinguishes admission status from physical
termination. Legacy ended guest rejections without proof remain cleanup
obligations. Missing original credentials can keep those old obligations
unresolved.
- Future writers must use the existing notification boundary.
Independent SQL writers and multiple processes need their own wake
contract.
- Pending inbound cleanup now blocks company deletion instead of
allowing cascade deletion to lose the provider obligation. Existing
voice-history deletion restrictions remain. This deliberately includes
an active anonymous inbound ring and an old call whose termination
cannot be verified. An operator must stop ingress and settle the call
before deletion; this slice adds no abandonment override and never
treats missing credentials as proof of termination. This is the retained
cleanup ownership policy, not a claim that every deletion attempt can
complete immediately.
- History retention and report processing still run in the remaining
maintenance lane. This PR does not extend Cloud wake coverage or claim
measured host/database suspension.

## Model Used

OpenAI Codex (GPT-6), with reasoning and code execution. The exact
serving model ID and context window were not exposed.

## 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 (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-28 08:30:15 -05:00
1 parent 6ce0909db4
commit 71b448b9cb
23 files changed
+1715 -57

No files matched your search

+111
View File
@@ -0,0 +1,111 @@
# Inbound voice cleanup without idle scans
Inbound admission cleanup runs from committed changes and saved deadlines. It
uses the shared delivery coordinator and scoped deadline worker. The recurring
voice maintenance lane now owns only history retention and reports, which remain
separate migrations. Manual reconciliation remains available for diagnostics.
## Ownership and durable state
An unbound admission owns the existing provider call until approval or guest
intake binds a voice session. The live-session recovery service then owns physical
termination. Admission cleanup never also ends that bound call. It can request
ending through the session row and copy the owner's confirmed terminal result.
Approval now commits the admission binding with the authorized session insert or
guest promotion. Expiry cannot see an old approving admission after that session
commits. Cleanup also locks and rechecks the admission before claiming it; a
candidate from an earlier scan is not permission to terminate a call. Both paths
lock admission before session. Provider requests run outside SQL transactions.
Older interrupted handoffs are repaired only when the existing session matches
the approval actor, authority, request key, task, token, generation and credential
identity. Conflicting ownership retains an error/retry instead of terminating a
call or inventing authority.
The migration adds three nullable fields to `chat_voice_inbound_calls`, not a new
queue table:
- `cleanupConfirmedAt` separates admission status from physical termination.
An ended or denied admission can still have an unconfirmed provider leg.
- `cleanupNextAttemptAt` survives crashes and restart. The claim saves the retry
before contacting the provider.
- `providerAccountId` records the account authenticated by the signed pre-call.
Cleanup can use a rotated key after verifying its account. A key for another
account cannot inspect or terminate this call. Older admissions lack that
evidence and require their original credential fingerprint.
The migration is replayable. It backfills only historical ended rows whose old
writer or signed lifecycle audit proves termination. It leaves old guest-denial
rows uncertain when their ended label did not prove physical cleanup. Those rows
receive a real check rather than disappearing as completed history. The partial
index covers unconfirmed admission identities. Startup reads at most 100 rows
per scope refresh and reconstructs deadlines across pages. The historical
backfill commits keyset batches of at most 1,000 rows through the migration
runner. Migration 0395 preserves the landed checkpoint lineage through 0394.
## Deadlines and quiet state
Rejection returns after its decision transaction commits. Its notification wakes
the registered cleanup worker; the HTTP request does not wait for provider account
verification, inspection, or hangup. The saved denial immediately rejects further
unauthorized call input while physical termination remains tracked separately.
Pending unbound admissions retain their original two-minute approval or ten-minute
guest-intake expiry. Rejection is immediately due. A cleanup claim saves a
30-second retry before provider I/O. An unconfirmed end receipt keeps that retry;
inspection, a confirmed end receipt, or a signed terminal callback clears it.
Unexpected failures also receive the shared worker's bounded failure backoff.
Neither a timeout nor an end request by itself proves termination.
Bound live calls leave no admission cleanup deadline. Their existing live-session
worker owns expiry, authorization checks and provider cleanup. Its terminal write
signals admission settlement. Healthy passive provider inspections do not wake
this admission worker. Confirmed history and configured-but-idle endpoints leave
no recurring admission SQL or provider requests.
Feature disable does not discard an outstanding call. Cleanup continues with its
saved identity. Missing credentials or account mismatches retain debt; they do not
permit use of a replacement account. No recovery path creates or redials a call.
## Producer audit
| Producer | Commit boundary and wake |
| --- | --- |
| Signed pre-call admission | Endpoint-scoped intent before insert in the quota transaction |
| Approval or rejection | Admission-scoped intent before the locked state write |
| Authorized session insert or guest promotion | Admission binding and intent inside the reservation transaction |
| Guest intake task/session binding | Session-scoped voice change before insert; admission binding shares the transaction |
| Pending tool detects expiry | Admission intent before the expired-state write |
| Signed terminal lifecycle/report callback | Session and admission intents before confirmed termination writes |
| Live-session termination, inspection or no-resource expiry | Existing voice-change boundary also signals inbound settlement |
| Reply provider confirms the call ended | Inbound session intent before the terminal session write |
| Configuration, authority or connection changes | Existing scoped voice-change composition |
| Cleanup claim, legacy handoff repair and confirmation | Admission intent in the owning transaction |
Notifications are process-local hints. Existing database rows remain truth. The
shared transaction layer delays hints until outer commit and retains uncertainty.
Other replicas and independent SQL writers require an explicit integration.
History retention cannot delete unconfirmed cleanup. Company deletion locks its
endpoints, matching ingress, then asks the voice owner to prove inbound cleanup is
settled. It rejects deletion while a provider obligation remains. This prevents
cascade deletion from erasing the evidence needed to finish cleanup. Existing
voice-history foreign-key restrictions otherwise remain unchanged.
## Verification and limits
Real PostgreSQL tests cover confirmed-history quiet, exact expiry, restart,
transaction rollback/outer commit, stale discovery, old interrupted handoffs,
bound-session ownership, unconfirmed requests, failures, fair paged backlog,
drain/resume, shutdown, retention protection, and deletion protection. Production
service tests exercise signed ingress, rejection, account mismatch and rotation,
feature disable, and atomic ordinary/guest approval. See the dated
[inbound cleanup audit](plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md) for
commands, counts and preserved failures. The current-parent acceptance records
266 distinct passing cases and the guarded batching, restart and approval
regressions. Earlier counts in that audit belong to their stated revisions.
These changes do not migrate history retention or call reports, extend Cloud's
external wake coverage, relax sleep eligibility for live calls, or prove actual
host/database suspension. A registered in-process deadline alone cannot wake
stopped compute.
@@ -1557,6 +1557,38 @@ fixture failures and local full-suite infrastructure limitations. Publication an
latest-head review/check acceptance remain outstanding. Inbound admission cleanup,
inbound history retention, and call reports remain separate next slices.
Voice reply delivery is [#17578](https://github.com/paperclipai/paperclip/pull/17578),
head 439d0a86eb75c129c30742a8cc24bf18e2a6ebf3, stacked on #17576. Local selected
checks pass; remote acceptance is pending. Slice 23 starts the inbound admission
cleanup audit in `2026-10-25-voice-inbound-cleanup-sleep-audit.md`. It explicitly
separates retention and reports and requires proof of approval/cleanup handoff,
termination confirmation, and protection against deleting unfinished cleanup.
Voice reply review repairs are published at #17578 head `90caaf3fa9`: shared scoped
worker, actual authority-restoration wiring regression, 191 overlapping selected
consumer tests and workspace types/build passing. Removing the real notification
reproduces the lost reply. Both review threads are resolved; fresh CI/review remain.
Inbound cleanup slice 23 now makes approval/session handoff atomic, after a real
expiry regression reproduced termination of an already authorized call. All 110
voice integration cases and server typechecking pass. This slice is not published:
locked cleanup claims, legacy handoff recovery, bound-guest ownership, termination
confirmation, retention protection and scheduler/producer migration remain required.
The full original inventory, older #17490 E2E failure, and stack/master reconciliation
remain outstanding. No stack PR has been merged by this work.
Slice 23 inbound cleanup now uses committed admission/session changes and actual
expiry/retry deadlines. The admission row separately records physical confirmation
and saved provider account. Locked claims, legacy handoff repair, bound-call owner
handoff, retention/deletion protection and safe same-account rotation are verified.
All 205 selected tests, workspace types and build pass; publication and remote
acceptance remain pending. History retention and call reports remain separate next
slices. Parent #17578 at 90caaf3fa9 has no pending/failed checks and exact-head
Greptile 5/5; the second Paperclip reviewer stopped before completion. The original
remaining inventory, #17490 E2E failure and final reconciliation remain in scope.
## Acceptance follow-up after the plugin parent merged (October 27)
Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at
@@ -1658,3 +1690,16 @@ budget after repair. The [reply audit](2026-10-25-voice-reply-delivery-sleep-aud
records restart, denial-cursor and guarded-mutation coverage. All 449 current
targeted cases, server typecheck/build and the policy check pass on parent
`ad79c7e805`. Fresh exact-head review, CI and parent landing remain required.
## Inbound cleanup current-parent acceptance, October 28
Inbound cleanup #17580 now includes reply parent `d2c0a7de8c` and uses its existing
paged worker contract. Startup yields after 100 rows per scope refresh, and a
same-owner restart reconstructs all 105 saved fixture deadlines. Migration 0395
was generated on the parent's 0394 lineage, preserves all historical SQL/journal
entries, and commits historical backfill in batches of at most 1,000. The
[inbound cleanup audit](2026-10-25-voice-inbound-cleanup-sleep-audit.md#current-parent-acceptance-and-bounded-discovery-october-28)
records 266 passing cases, negative probes, migration lineage, preserved failures,
and limits. Exact-head review, CI, and the preceding stack still must pass before
landing. History retention and call reports remain separate duties.
@@ -0,0 +1,419 @@
# Inbound voice admission cleanup sleep audit
Original baseline: voice reply PR #17578, head 439d0a86eb75c129c30742a8cc24bf18e2a6ebf3.
Current integrated parent: d2c0a7de8c3e9cac8382b1268f3f5d883a2067b3.
Branch: codex/sleep-23-voice-inbound-cleanup.
Status: cleanup implementation and local verification complete; publication and remote acceptance pending.
## One reason in this slice
Remove recurring discovery of expired inbound admissions and unfinished provider
termination. Keep inbound history retention and call reports as separate PRs.
The current `voiceInboundService.reconcile` combines a seven-day delete with a
25-row expiry/provider scan. `voiceSessionService.reconcileMaintenance` invokes
it from the remaining one-second chat lane.
Existing `chat_voice_inbound_calls` rows own admission state, expiresAt, updatedAt,
providerSessionId, generation, credential fingerprint, requested issue, caller,
and optional bound voice session/intake task. Existing voice sessions own the
bound call's execution and current authorization. Do not add a table per timer.
Decide and document the handoff between those existing owners before coding.
## Current behavior and supported producers
- Signed `call.pre_call` inserts an awaiting-approval admission with a two-minute
expiry, or a guest-intake admission with a ten-minute expiry. The endpoint lock
serializes the quota check and insertion. Repeated provider identities are
deduplicated; changed replay content is rejected.
- `pendingTool` can mark an expired admission as expired. Guest submit_request
creates a low-trust task and active voice session, then saves the session/task
binding in the same transaction. These writes must wake the appropriate owner.
- `decide` locks the admission and validates current caller, endpoint, credential,
approval code, expiry, and task authority. It writes approving or denied/ended.
Provider inspection and session start occur afterward. Final approved binding
is currently a standalone update guarded only by state=approving.
- Signed lifecycle/report callbacks lock admission before session and mark the
admission ended. Preserve that order and the callback receipt's replay fence.
- The expiry sweep scans guest_intake, awaiting_approval, approving, denied, and
expired rows only once their expiresAt is past. It inspects the exact saved
provider call, marks ended if inspection confirms termination, otherwise marks
expired and requests termination. Errors only update updatedAt.
- No production admission writers were found outside voice-inbound-service.ts.
Session handoff/approval recovery must also inspect voice-session-store.reserve.
## Correctness questions that the migration must resolve
1. **Discovery is not an execution claim.** A row read by cleanup can change before
provider I/O. Recheck/claim current due state under the owning row lock. Never
overwrite a newer approved or terminal state from a stale candidate snapshot.
2. **Approval handoff has a crash window.** Inbound start commits an authorized
session with the existing provider ID and returns without creating a new call.
The admission's final approved update follows in a different transaction.
Expiry or a crash can occur between these writes. Determine whether existing
binding evidence should complete handoff or retain cleanup; do not end a valid
handed-off call based on an older approving snapshot. Test the exact race.
3. **Termination requested is not termination confirmed.** Speko endSession returns
confirmed=false for status=ending and true only for already_ended. Keep cleanup
debt until an inspect/callback/end receipt establishes termination. Never create
or redial a provider call in recovery.
4. **Guest denial currently writes ended before provider confirmation.** Without a
bound session, a failed end request can lose its cleanup obligation. With a
session, live-session recovery also owns ending. Preserve one explicit owner
for unfinished termination; do not treat a terminal UI label as physical proof.
5. **Bound guest calls have two current scanners.** Admission expiry and live-session
recovery can inspect/end the same exact call. Define the handoff and admission
settlement contract so a scheduler migration does not create two uncoordinated
provider workers for one resource.
6. **Retention currently precedes cleanup.** The seven-day delete can remove denied
or expired rows with no session/task binding before provider termination is
proved. Keep history retention separate, but do not allow it to erase newly
explicit unfinished cleanup debt. Record any required narrow predicate change
with this slice and keep the retention scheduling migration in its own PR.
7. **Configuration and credentials.** Cleanup must remain possible after feature
disable and endpoint/credential changes. Do not use a replacement credential
to act on an unrelated provider call. Read the existing provider/account
identity and deletion contracts before selecting recovery scope.
## Intended scheduler contract
The inbound owner exposes normalized identities and actual deadlines to the
existing delivery coordinator. The initial deadline comes from saved admission
expiry; explicit rejection requests immediate cleanup through the same owner.
Unconfirmed termination or errors retain a bounded retry. Completed history and
no admissions leave no cleanup deadline. Startup reconstructs saved obligations.
Notifications precede writes in existing transaction ownership and cover all
producer paths above. No provider request is made inside a SQL transaction.
Preserve scope coalescing, per-obligation failure isolation, fair bounded batches,
startup paging, drain/resume, concurrent changes during a sweep, and shutdown
joining in-flight work. Keep public/manual reconciliation compatible while
separating the retention operation from the scheduled cleanup operation.
## Required proof
- No configuration, configured idle, and confirmed terminal history become quiet.
- Real signed pre-call after quiet registers its original expiry.
- First guest intake, approval/denial, pending-tool expiry, and lifecycle callbacks
signal their existing transactions; outer rollback and commit behave correctly.
- Approval completes during cleanup discovery without stale termination or state
overwrite. A crash between session binding and final approval remains recoverable.
- End receipts with confirmed=false, provider failure, missing callbacks, and restart
retain the exact-call obligation without redial or silent deletion.
- Bound guest termination has one explicit owner and settles admission history.
- Retention cannot erase unfinished provider cleanup.
- Backlog, failure isolation, drain/resume, and shutdown behave as claimed.
- Existing full voice, reply, and call-report cases continue to pass. Use separate
PostgreSQL databases for suites that retain fixtures and global-idle suites.
## Parent and full-stack obligations
#17578 is published and attached. Local evidence covers 786 selected tests,
workspace typecheck and build. CI run 40179014816 and PR review 40179014437 were
live at publication; no check failure was present at the latest inspection.
Latest-head checks, Greptile score and review findings still require acceptance.
No stack PR is authorized for merge by the active goal. The remaining original
ledger, older #17490 failure, and final master reconciliation remain required.
## Atomic handoff checkpoint
The parent reply review repairs are published at `90caaf3fa929141a13cc17f54d6be846d6e93217`
and merged into this branch. Both voice workers now share the scoped deadline
mechanism; the next inbound worker should reuse it rather than copy a third loop.
Parent CI run 40180022971 is live; no failure was present at the latest inspection.
Its old 439d head passed CI/Greptile, but fresh-head acceptance remains required.
A real-service regression paused after inbound session reservation committed but
before its result reached `decide`, then advanced beyond the admission expiry.
The original sweep called provider termination on that authorized session. This
failed in `/tmp/sleep23-handoff-before.log`. The reservation now writes admission
state=approved and sessionId in the same transaction as insertion or guest
promotion. `decide` reads that binding after start, rather than creating it in a
second transaction. No provider call is moved inside the transaction.
Both ordinary admission and guest-to-private promotion are covered. All 110 voice
integration cases pass on a fresh PostgreSQL database, and server typechecking
passes. An initial full run passed 109 and failed the guest test's assertion of
no termination calls globally: advancing ten minutes also expired unrelated
retained fixture admissions. The assertion now checks the exact provider call
whose handoff is under test, preserving legitimate unrelated cleanup. The full
rerun passes. No broad workspace/remote verification is claimed for this child.
This is a prerequisite, not the complete cleanup migration. Still required:
- Recheck and claim a discovered admission under its row lock before provider I/O;
atomic new handoff alone does not protect a stale discovery snapshot.
- Reconstruct legacy interrupted handoffs written before this atomic change.
- Define one termination owner for bound guest sessions, and settlement signals.
- Preserve unconfirmed/failed termination debt and credential/account isolation.
- Prevent history retention from deleting unfinished cleanup.
- Wire all producer notifications, actual deadlines, scoped worker registration,
lifecycle tests and remove only the cleanup portion of periodic reconciliation.
- Keep history retention and call reports as separate forthcoming PRs.
## Cleanup implementation and local acceptance
The worker is now registered in app startup. `reconcileMaintenance` no longer
runs inbound cleanup; it retains history pruning and reports for separate PRs.
See [the implementation contract](../idle-voice-inbound-cleanup.md) for the complete
producer table, ownership, credential rotation and remaining host-sleep limits.
The previous checkpoint's remaining implementation items are now handled:
locked current-state claims; matching legacy handoff repair; bound-session cleanup
ownership; durable termination confirmation/retry; account isolation; retention
protection; notifications and shared scheduler registration. A company-delete
participant check also prevents cascades from erasing unfinished cleanup and locks
endpoints to serialize against signed admission ingress.
The existing admission row needs separate authorization and cleanup state. Three
nullable fields record confirmation, retry and the signed provider account.
Migration 0392_lying_trish_tilby.sql was generated with `pnpm db:generate`, made
idempotent, and given a concurrent partial index. It backfills only provable old
termination. The initial two-field unpublished generation was replaced by this
single migration after account-rotation analysis. Five newest snapshots remain;
the generated deletion of 0383 is intentional.
Local evidence:
- `voice-inbound-cleanup.test.ts`: 13 tests pass using a separate PostgreSQL DB.
Includes migration backfill/replay, company deletion and shutdown joining.
- Full voice integration and app wiring: 129 tests pass on a fresh DB (113 voice,
16 app wiring). Signed ingress wakes an idle worker; a feature-disabled instance
retains an unconfirmed end; a replacement account cannot issue cleanup; a key
for the saved account can. Rejection failure survives history pruning.
- Company deletion and server startup consumers: 52 tests pass on their isolated DB.
- Parent live-session recovery: 11 tests pass on its separate DB.
- Total selected coverage: 205 passing cases across these nonoverlapping suites.
- `pnpm -r typecheck` and `pnpm build` pass on the current implementation.
- Migration safety/numbering checks passed for the initial generated migration;
rerun on the final regenerated name before publication. Initial application and
replay succeeded on real PostgreSQL. Final generation also applied cleanly;
its backfill is exercised twice by the lifecycle test.
Earlier handoff regression and fixture assertion failures remain recorded above.
The repository-wide local test command still has the previously recorded embedded
PostgreSQL host limitation; this is not a claim that that command completed. Fresh
remote CI and review are required before accepting the PR.
Parent #17578 at 90caaf3fa9 has all checks terminal with no failures and Greptile
5/5 on that exact head. Both requested findings are fixed/resolved. The fresh
Paperclip reviewer stopped before completing its turn; do not report a new 5/5
assessment from that reviewer. The full original migration ledger, #17490 failure,
and final stack/master reconciliation remain outstanding.
## Migration runner repair after first CI
The initial head 4b4bf2d8af failed database setup. Greptile identified that the
concurrent 0388 index was missing from the migration runner's top-level set.
CI production bundle boot and Runner static integration logs independently show
`CREATE INDEX CONCURRENTLY cannot run inside a transaction block`. The earlier
SQL application/replay evidence bypassed this application runner and did not
establish that fresh setup or upgrades could work.
Added a real-PostgreSQL migration regression that creates a disposable database,
boots all migrations through `applyPendingMigrations`, then simulates an invalid
concurrent index and replays through two concurrent application migrators. It
failed with the exact transaction error before the fix. Registering 0388 in the
existing top-level set makes fresh boot, invalid-index repair, single journal
record, and repeat replay pass. No migration SQL/history was rewritten.
Database typecheck and migration numbering/safety checks also pass. Local logs:
`/tmp/sleep23-migrator-before.log`, `/tmp/sleep23-migrator-after.log`, and
`/tmp/sleep23-migrator-types.log`. Fresh remote checks remain required; the
original CI failure is not classified as a flake.
## Denial request latency review repair
Review finding 4370459974 correctly identified that denial awaited account
verification, provider inspection and endSession, despite already committing the
rejection and its cleanup notification. The request now returns the saved denial;
the registered worker owns physical termination and persists its retry. No detached
promise or second dispatch path is added. Per-attempt account verification stays
with the provider boundary, protecting rotated credentials from a different
account. This patch does not cache provider authority indefinitely.
A new real-producer regression holds verifyAgent unresolved, proves the committed
denial returns, checks that cleanup is still unconfirmed with a saved retry, then
releases the provider and verifies confirmed termination. It failed before the
fix (`/tmp/sleep23-denial-before.log`) and passed after
(`/tmp/sleep23-denial-after.log`). Existing explicit one-shot tests now invoke
that one-shot path rather than relying on the HTTP decision to do cleanup.
All 130 full voice and app wiring cases pass on a fresh migrated PostgreSQL DB;
production server typecheck passes. Logs: `/tmp/sleep23-denial-acceptance.log`
and `/tmp/sleep23-denial-types.log`.
Finding 4370459800 about company deletion remains separately unresolved. Missing
credentials and account mismatch retain cleanup debt. There is a tested recovery
path through restoring the original account, but a permanently unavailable
provider needs a deliberate ownership/product policy. A retry cap, elapsed time,
or ambiguous 404 is not evidence that a billable call ended. Do not manufacture
cleanup confirmation or silently discard the obligation to unblock deletion.
## Retrying an interrupted approval (2026-10-27)
Older versions could commit a private session or guest promotion before writing
the admission binding. Retrying approval then found the existing session and
returned before the new atomic binding helper ran. The inbound API rejected the
still-approving admission with “This call ended during approval.”
Reservation now validates the inbound admission before the existing-session
return. It uses the existing session authorizer to check current liveness,
expiry, generation, credentials, caller and task access, then checks the exact
admission/session identity before binding in the same transaction. Lock order
remains endpoint, connection, admission, session. A matching provider session ID
alone cannot establish approval. Already approved idempotent calls remain valid
and do not rewrite the binding. Browser and outbound reservation behavior stays
with the existing path. This adds no table, timer or recovery wrapper.
Both direct and promoted-guest retry regressions failed through the real inbound
API before the correction. They now return the same session without starting or
ending another provider call. Ended, expired and token-changed sessions are
rejected without binding. Existing atomic handoff and approval-authority cases
also pass. An initial negative expiry fixture violated the database timestamp
constraint; it was corrected to represent a valid session that has expired.
All 132 cases across the complete voice-session and inbound-cleanup suites pass
on separate fresh databases, and server typechecking passes. Downstream
integration and latest-head CI remain required. The separate company-deletion
cleanup findings are not resolved by this correction.
## Fairness-test clock follow-up (2026-10-27)
CI's backlog test observed two provider failures where it expected one. A controlled
30 ms provider delay reproduced the same failure: the assertion callback advanced
fake time by one second on every retry, in addition to Vitest's ordinary clock
advancement, reaching the failed provider's retry while the original backlog was
still settling through PostgreSQL. Removing that extra clock jump makes the same
controlled case pass. The five-second assertion budget and all product code remain
unchanged. With the temporary provider delay removed, all 13 lifecycle tests pass.
This repairs test clock ownership; it does not weaken provider retries or delivery
fairness. Current-head CI remains required.
## Current-parent acceptance and bounded discovery, October 28
This revision normally merges reply parent `d2c0a7de8c3e9cac8382b1268f3f5d883a2067b3`.
It keeps the parent's paged shared worker, failure isolation, denial restoration,
and publication-scoped reply signals. It adds no new shared scheduler contract.
Domain SQL and termination decisions remain in the inbound voice owner.
The previous cleanup `load` read every page before yielding. An actual 105-row
pending-admission fixture failed its first-pass budget: 105 selected rows instead
of 100. The registered worker now reads one page per scope refresh and returns
`{ rows, complete }` through the parent's existing contract. The explicit one-shot
`load` still returns complete inventory for its foreground callers. Each query is
limited to 100 rows. Refresh cursors advance only after a successful, non-aborted
read. Registration resets them when the same service owner restarts.
The new lifecycle case stops after the first page of 105 future admissions,
re-registers the same service, reconstructs the original expiry, and confirms
all 105 terminations. It checks no SQL or provider calls before the deadline and
no recurring work after settlement. Removing the registration cursor reset fails
at the real restarted read (5 rows instead of 100). These measurements cover
selected rows, not physical PostgreSQL tuple visits or deployed resource cost.
The fresh external fixture first failed because it assumed the admission schema
already existed. It now runs the application migrator before setup, with the
standard embedded-database setup budget. This repairs setup ownership; it does
not skip any cleanup or authority assertions.
### Migration lineage and bounded historical backfill
The parent includes native checkpoint migration 0394. The old, unmerged inbound
0392 migration cannot overwrite that history. The complete parent migration tree
and client were restored, then the existing schema edit was regenerated with
`pnpm db:generate`. The new migration is `0395_married_the_hand.sql`. Its generated
snapshot links the parent snapshot. All 393 historical SQL files and all prior
journal entries are byte-for-byte preserved. The five retained snapshots are
0391 through 0395; the generator's required deletion of 0390 is included.
The nullable columns and partial index remain replayable. The concurrent index
and batch backfill use the application's existing top-level migration protocol.
The original unbounded historical UPDATE is now an indexed keyset scan of at
most 1,000 unconfirmed admission IDs per transaction. It commits each batch.
The first page includes the nil UUID. Each batch retains the original ended-state
and same-company signed termination-evidence predicate. It does not confirm a
call from its label alone. No new queue table or temporary production index is
introduced. The existing activity entity index supports the evidence lookup.
The real migrator regression creates and drops its own database. It covers fresh
migration, invalid concurrent-index repair, two simultaneous replays, and a single
migration receipt. It confirms 1,007 eligible historical rows across multiple
actual transaction IDs, with at most 1,000 updates in one transaction. Denied,
pending, expired, wrong-company and wrong-action evidence stay unconfirmed. A
prior confirmation remains unchanged. Reapplying the receipt adds no updates.
The transaction probe exists only in that disposable test database. Restoring
the old unbounded UPDATE makes the test fail: one transaction instead of at least
two. Snapshot/schema drift acceptance also passes.
### Final local evidence
266 distinct cases pass on this combined source:
- 122 complete voice-session integration cases, using an initially empty test DB.
- 142 cases across nine complete inbound cleanup, voice recovery, company removal,
shared deadline/coordinator/batching, startup and hostname suites. Cleanup and
parent recovery use separate initially empty DBs. Inbound cleanup uses its
dedicated DB after the fresh-setup repair; all 14 lifecycle cases pass.
- Two complete database migration and schema-snapshot cases.
Database and server typechecking and builds pass. Migration numbering/safety
checks pass. The complete voice suite retains both interrupted-approval retry
regressions. Removing only the existing-session binding makes those real API
requests reject instead of resolve; the source was restored after the probe.
The bounded-backfill and restart probes were also restored before final acceptance.
Commands use `pnpm --filter @paperclipai/server exec vitest run` for the ten named
server suites, `pnpm --filter @paperclipai/db exec vitest run` for the two database
cases, and each affected package's `typecheck` and `build` scripts. Evidence logs:
`/tmp/sleep23-final-voice-consumers.log`,
`/tmp/sleep23-final-lifecycle-consumers.log`,
`/tmp/sleep23-current-migration-acceptance.log`, and
`/tmp/sleep23-current-migration-lineage-proof.json`. The three guarded negative
probes have `/tmp/sleep23-*-mutant-red.log` logs. Earlier failed fixture setup,
unbounded inventory, and migration failures remain recorded above and in
`/tmp/sleep23-empty-fixture-red.log` and `/tmp/sleep23-page-budget-red.log`.
The older 130/132/205-case and full-workspace claims apply only to their stated
revisions. No new full local repository suite, live provider, Cloud alarm,
instance/database suspension, or savings claim is made. Latest-head review,
all CI, and parent landing remain required. Company deletion still refuses to
erase unconfirmed cleanup; an audited tombstone or override requires a separate
product and ownership decision.
The module-boundary check passes. The full local Node policy scan reports two
ignored, generated Copilot distribution manifests without an engine field after
the successful package builds. Neither file is tracked or changed by this PR.
The unchanged checker passes on a temporary copy of all its tracked source inputs
(all manifests, Dockerfiles, workflows and required policy fragments). The initial
full-tree failure remains `/tmp/sleep23-final-policy.log`; the source-input check
and its complete input list are `/tmp/sleep23-final-tracked-policy.log` and
`/tmp/sleep23-tracked-policy-proof.json`. No checker, workflow, generated vendor
manifest, or ignore rule was changed to obtain this result. Remote CI remains a
separate gate.
### Legacy terminal admissions missing their session binding — October 28
Ponytail found that an old interrupted approval could leave an admission `ended` or `expired` with a NULL session binding even though a matching provider session exists. The approval-only repair rejected that state forever, leaving cleanup unconfirmed and the company-deletion guard armed. Five new regressions reproduced the ownership conflict on the reviewed source before the repair.
Cleanup now handles confirmed terminal sessions before approval repair. It copies the existing termination timestamp without granting approval or issuing another provider request. Rejected admissions with a matching active inbound session retain their rejection, persist the missing binding and hand the leg to live recovery. Persisting the binding also lets the subsequent session-scoped confirmation wake find the admission. A conflicting existing binding, mode, idempotency key, generation, credential fingerprint or tool-token digest cannot authorize that handoff. Five negative regressions retain cleanup debt and leave the unrelated session active.
The complete cleanup lifecycle suite passes 24 cases, including quiet after settlement, current ownership, rejected-state preservation, matching terminal confirmation and conflicting identities. The complete voice-store integration suite passes 122 cases on a separate initially empty PostgreSQL 17 database. These are 146 distinct current cases; the isolated transcript probe is not counted again. Production server typecheck and build pass. No local full-repository, real provider, Cloud wake or host-suspension claim is made.
Failure history: the first regression invocation set the generic database variable rather than this suite's explicit variable and skipped all cases; it is not a pass. The corrected invocation executed and failed all five intended ownership regressions. The first repaired full cleanup run passed 22 and failed two new test calls to low-level chat deletion because retained voice sessions still have endpoint foreign keys. The tests now exercise the owning cleanup deletion guard (409 before cleanup, settled after confirmation); production deletion ordering is unchanged. Starting another disposable PostgreSQL server failed because the host's semaphore capacity was exhausted. The already-owned PostgreSQL server was verified available and reused with new isolated test databases; no shared IPC resources were removed.
The first complete voice-store run passed 121 cases and timed out at the unchanged 15-second limit in `retains both spoken sides once on the public call task and preserves quarantine`. Its isolated fresh-database probe passed. One subsequent complete run, with builds finished, passed all 122 cases without source or timeout changes. Recorded transform time was 59.09 seconds for the failed full invocation and 34.65 seconds for the passing full invocation; this does not establish the timeout's underlying cause. The earlier timeout remains part of the verification record.
### Landed-parent integration and regenerated migration — October 28
Reply delivery #17578 landed at `e93732c3d49f513845543a6c5d61054950969923`. This normal merge uses actual master `e7ea2e2961863171956e70fb05d3d1f3790f7909`, including the unrelated eval ingestion change. The cleanup owner, voice-store consumer suite and inbound admission schema are byte-identical to tested repair `7e4588fa8c`; the cleanup fixture changes only its migration filename. All unrelated master changes remain present.
Master acquired `0395_tranquil_firestar.sql` while this stack was open. Its SQL, every other landed SQL file and the parent journal prefix are preserved. Running `pnpm db:generate` produced `0396_tan_paladin.sql` and its new snapshot. The previously reviewed idempotent cleanup SQL, concurrent index and bounded historical backfill are retained byte for byte under that generated filename; application registration and both migration fixtures now reference 0396. Exactly snapshots 0392 through 0396 remain, with 0391 pruned by the existing command.
On the combined source, database build and numbering/safety checks pass. The complete 24-case cleanup lifecycle suite passes on a new empty database. The separate database migration integration case passes, including fresh application boot, interrupted concurrent-index repair, serialized replay and the real transaction-bounded backfill. Production server typecheck and build pass again after integration. The prior 122-case voice-store pass is carried only for its unchanged consumer and cleanup source; it was not rerun merely for the new migration filename. These checks do not establish deployed sleep or provider behavior. Fresh published-head CI and reviews remain separate merge gates.
### CI connected-tool rate-limit clock repair — October 28
Published head `48d8356e73` received exact-head Greptile 5/5. Its replacement CI run 40692550133 failed server shard 9/12 (job 122830940435): 1,610 cases passed, two skipped, and the connected MCP policy/approval/rate-limit acceptance case failed with `Expected connected MCP call to be rate limited`. This is a separate real failed check, not the cancelled duplicate workflow run 40692549975 created during retargeting.
The policy owner uses fixed 60-second buckets. The existing acceptance test performed its first and second calls against the real clock, so a legitimate window change could make the second call allowed. A controlled minute rollover between those calls reproduced the exact failing assertion on unchanged production code; the diagnostic source was restored. The CI log does not directly record the two bucket timestamps, so this reproduction establishes the test's unsafe timing assumption without claiming an independently captured clock trace for the CI attempt.
The existing acceptance case now freezes only `Date` around the rate-limit assertions, preserves the second-call 429 and persisted invocation assertions, then advances one window and requires a new call to succeed. A `finally` restores the real clock; network and timeout timers remain real. The complete 92-case gateway acceptance suite passes on a dedicated database. A subsequent whitespace-only correction removed an introduced blank-line indentation and changes no tokens. No production limiter, policy, credential, timeout, workflow or assertion was weakened. This additional existing test fixture is the only new source path outside the cleanup slice and fixes the observed CI gate. Fresh CI and exact-head review are required again after publication.
+1
View File
@@ -547,6 +547,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
"0392_small_frightful_four.sql",
"0393_open_grandmaster.sql",
"0395_tranquil_firestar.sql",
"0396_tan_paladin.sql",
"0384_funny_adam_warlock.sql",
@@ -0,0 +1,31 @@
ALTER TABLE "chat_voice_inbound_calls" ADD COLUMN IF NOT EXISTS "provider_account_id" text;--> statement-breakpoint
ALTER TABLE "chat_voice_inbound_calls" ADD COLUMN IF NOT EXISTS "cleanup_confirmed_at" timestamp with time zone;--> statement-breakpoint
ALTER TABLE "chat_voice_inbound_calls" ADD COLUMN IF NOT EXISTS "cleanup_next_attempt_at" timestamp with time zone;--> statement-breakpoint
CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_voice_inbound_cleanup_idx" ON "chat_voice_inbound_calls" USING btree ("id") WHERE "chat_voice_inbound_calls"."cleanup_confirmed_at" is null;
--> statement-breakpoint
DO $$
DECLARE cursor_id uuid; batch_ids uuid[];
BEGIN
LOOP
IF cursor_id IS NULL THEN
SELECT array_agg(id ORDER BY id) INTO batch_ids FROM (
SELECT id FROM chat_voice_inbound_calls WHERE cleanup_confirmed_at IS NULL ORDER BY id LIMIT 1000
) batch;
ELSE
SELECT array_agg(id ORDER BY id) INTO batch_ids FROM (
SELECT id FROM chat_voice_inbound_calls WHERE cleanup_confirmed_at IS NULL AND id > cursor_id ORDER BY id LIMIT 1000
) batch;
END IF;
EXIT WHEN batch_ids IS NULL;
UPDATE chat_voice_inbound_calls c
SET cleanup_confirmed_at = c.updated_at
WHERE c.id = ANY(batch_ids) AND c.cleanup_confirmed_at IS NULL AND c.state = 'ended'
AND (c.approved_by_user_id IS NULL OR EXISTS (
SELECT 1 FROM activity_log a
WHERE a.company_id = c.company_id AND a.entity_type = 'voice_inbound_call'
AND a.entity_id = c.id::text AND a.action = 'voice.call.ended'
));
cursor_id := batch_ids[array_length(batch_ids, 1)];
COMMIT;
END LOOP;
END $$;
@@ -1,6 +1,6 @@
{
"id": "e080fc1b-af8e-4012-8ce5-3b5065b65143",
"prevId": "f38ffbf3-c0ce-435e-aad8-9e5afa6b9417",
"id": "be87a5d8-4164-4d95-89ae-b895701fa21b",
"prevId": "66fe02b3-6672-4c19-b447-57acad43bf16",
"version": "7",
"dialect": "postgresql",
"tables": {
@@ -1350,6 +1350,34 @@
"concurrently": false,
"method": "btree",
"with": {}
},
"activity_log_company_user_issue_actor_idx": {
"name": "activity_log_company_user_issue_actor_idx",
"columns": [
{
"expression": "company_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "actor_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "entity_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"activity_log\".\"actor_type\" = 'user' AND \"activity_log\".\"entity_type\" = 'issue'",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
@@ -5721,6 +5749,28 @@
}
},
"indexes": {
"approvals_pending_retention_idx": {
"name": "approvals_pending_retention_idx",
"columns": [
{
"expression": "updated_at",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"approvals\".\"status\" = 'pending'",
"concurrently": false,
"method": "btree",
"with": {}
},
"approvals_company_status_type_idx": {
"name": "approvals_company_status_type_idx",
"columns": [
@@ -14138,6 +14188,24 @@
"primaryKey": false,
"notNull": false
},
"provider_account_id": {
"name": "provider_account_id",
"type": "text",
"primaryKey": false,
"notNull": false
},
"cleanup_confirmed_at": {
"name": "cleanup_confirmed_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"cleanup_next_attempt_at": {
"name": "cleanup_next_attempt_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"expires_at": {
"name": "expires_at",
"type": "timestamp with time zone",
@@ -14160,6 +14228,22 @@
}
},
"indexes": {
"chat_voice_inbound_cleanup_idx": {
"name": "chat_voice_inbound_cleanup_idx",
"columns": [
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_voice_inbound_expiry_idx": {
"name": "chat_voice_inbound_expiry_idx",
"columns": [
@@ -41923,6 +42007,28 @@
}
},
"indexes": {
"join_requests_pending_retention_idx": {
"name": "join_requests_pending_retention_idx",
"columns": [
{
"expression": "updated_at",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"join_requests\".\"status\" = 'pending_approval'",
"concurrently": false,
"method": "btree",
"with": {}
},
"join_requests_invite_unique_idx": {
"name": "join_requests_invite_unique_idx",
"columns": [
@@ -42943,6 +43049,99 @@
"checkConstraints": {},
"isRLSEnabled": false
},
"public.native_runner_checkpoints": {
"name": "native_runner_checkpoints",
"schema": "",
"columns": {
"company_id": {
"name": "company_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"run_id": {
"name": "run_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"issue_id": {
"name": "issue_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"part": {
"name": "part",
"type": "text",
"primaryKey": false,
"notNull": true
},
"material": {
"name": "material",
"type": "jsonb",
"primaryKey": false,
"notNull": true
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"native_runner_checkpoints_headers_idx": {
"name": "native_runner_checkpoints_headers_idx",
"columns": [
{
"expression": "run_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"native_runner_checkpoints\".\"part\" = 'header'",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk": {
"name": "native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk",
"tableFrom": "native_runner_checkpoints",
"tableTo": "heartbeat_runs",
"columnsFrom": [
"company_id",
"issue_id",
"run_id"
],
"columnsTo": [
"company_id",
"native_issue_id",
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {
"native_runner_checkpoints_run_id_part_pk": {
"name": "native_runner_checkpoints_run_id_part_pk",
"columns": [
"run_id",
"part"
]
}
},
"uniqueConstraints": {},
"policies": {},
"checkConstraints": {},
"isRLSEnabled": false
},
"public.pipeline_automation_executions": {
"name": "pipeline_automation_executions",
"schema": "",
+7
View File
@@ -2759,6 +2759,13 @@
"when": 1793159398533,
"tag": "0395_tranquil_firestar",
"breakpoints": true
},
{
"idx": 396,
"version": "7",
"when": 1793191988458,
"tag": "0396_tan_paladin",
"breakpoints": true
}
]
}
@@ -27,6 +27,9 @@ export const chatVoiceInboundCalls = pgTable("chat_voice_inbound_calls", {
approvedByUserId: text("approved_by_user_id"), callerAuthority: text("caller_authority"),
intakeIssueId: uuid("intake_issue_id"),
requestedIssueId: uuid("requested_issue_id"), sessionId: uuid("session_id"),
providerAccountId: text("provider_account_id"),
cleanupConfirmedAt: timestamp("cleanup_confirmed_at", {withTimezone: true}),
cleanupNextAttemptAt: timestamp("cleanup_next_attempt_at", {withTimezone: true}),
expiresAt: timestamp("expires_at", {withTimezone: true}).notNull(),
createdAt: timestamp("created_at", {withTimezone: true}).notNull().defaultNow(),
updatedAt: timestamp("updated_at", {withTimezone: true}).notNull().defaultNow(),
@@ -36,6 +39,7 @@ export const chatVoiceInboundCalls = pgTable("chat_voice_inbound_calls", {
foreignKey({columns: [t.companyId, t.sessionId], foreignColumns: [chatVoiceSessions.companyId, chatVoiceSessions.id], name: "chat_voice_inbound_session_fk"}),
foreignKey({columns: [t.companyId, t.intakeIssueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_inbound_intake_fk"}),
foreignKey({columns: [t.companyId, t.requestedIssueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_inbound_requested_issue_fk"}),
index("chat_voice_inbound_cleanup_idx").on(t.id).where(sql`${t.cleanupConfirmedAt} is null`),
index("chat_voice_inbound_expiry_idx").on(t.state, t.expiresAt),
check("chat_voice_inbound_state_check", sql`${t.state} in ('guest_intake', 'awaiting_approval', 'approving', 'approved', 'denied', 'expired', 'ended')`),
check("chat_voice_inbound_token_check", sql`${t.toolTokenHash} ~ '^[a-f0-9]{64}$'`),
@@ -0,0 +1,120 @@
import { createHash, randomUUID } from "node:crypto";
import { readFile } from "node:fs/promises";
import postgres from "postgres";
import { describe, expect, it } from "vitest";
import { applyPendingMigrations } from "./client.js";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js";
const external = process.env.PAPERCLIP_MIGRATION_TEST_DATABASE_URL;
if (external && !new URL(external).pathname.endsWith("_test")) throw new Error("Migration test URL must name a test database");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
const migrationFile = "0396_tan_paladin.sql";
const indexName = "chat_voice_inbound_cleanup_idx";
(support.supported ? describe : describe.skip)("inbound voice cleanup migration", () => {
it("boots through the application migrator and repairs an interrupted index on concurrent replay", async () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let admin: ReturnType<typeof postgres> | undefined;
let databaseName: string | undefined;
let connectionString: string;
let client: ReturnType<typeof postgres> | undefined;
try {
if (external) {
admin = postgres(external, { max: 1, onnotice: () => {} });
databaseName = `voice_migration_${randomUUID().replaceAll("-", "")}_test`;
await admin.unsafe(`CREATE DATABASE "${databaseName}"`);
const url = new URL(external); url.pathname = `/${databaseName}`;
connectionString = url.toString();
await applyPendingMigrations(connectionString);
} else {
temporary = await startEmbeddedPostgresTestDatabase("voice-inbound-migration-");
connectionString = temporary.connectionString;
}
client = postgres(connectionString, { max: 1, onnotice: () => {} });
const columns = await client`SELECT column_name FROM information_schema.columns
WHERE table_name = 'chat_voice_inbound_calls'`;
expect(columns.map(row => row.column_name)).toEqual(expect.arrayContaining([
"cleanup_confirmed_at", "cleanup_next_attempt_at", "provider_account_id",
]));
const content = await readFile(new URL(`./migrations/${migrationFile}`, import.meta.url), "utf8");
const hash = createHash("sha256").update(content).digest("hex");
await client`DELETE FROM drizzle.__drizzle_migrations WHERE hash = ${hash}`;
await client.unsafe(`DROP INDEX ${indexName}`);
const [company, otherCompany] = await client`INSERT INTO companies (name, issue_prefix)
VALUES ('First', 'VM1'), ('Second', 'VM2') RETURNING id`;
const agentId = randomUUID(), applicationId = randomUUID(), connectionId = randomUUID(), endpointId = randomUUID();
await client`INSERT INTO agents (id, company_id, name, adapter_type) VALUES (${agentId}, ${company.id}, 'Migration voice', 'paperclip_runner')`;
await client`INSERT INTO tool_applications (id, company_id, name, type) VALUES (${applicationId}, ${company.id}, 'Voice', 'chat')`;
await client`INSERT INTO tool_connections (id, company_id, application_id, name, uid, transport, connection_purpose)
VALUES (${connectionId}, ${company.id}, ${applicationId}, 'Voice', ${connectionId}, 'voice', 'channel')`;
await client`INSERT INTO chat_endpoints (id, company_id, connection_id, assigned_agent_id, provider, public_id, status)
VALUES (${endpointId}, ${company.id}, ${connectionId}, ${agentId}, 'speko', ${endpointId}, 'active')`;
await client`WITH candidates AS (SELECT gen_random_uuid() AS id FROM generate_series(1, 1005))
INSERT INTO chat_voice_inbound_calls (id, company_id, endpoint_id, provider_session_id, state,
approval_code, generation, credential_fingerprint, tool_token_hash, request_fingerprint, expires_at)
SELECT id, ${company.id}, ${endpointId}, 'migration:' || id::text, 'ended', '123456', 1,
'migration', repeat('a', 64), id::text, now() FROM candidates`;
const nilId = '00000000-0000-0000-0000-000000000000', signed = randomUUID();
const denied = randomUUID(), wrongCompany = randomUUID(), wrongAction = randomUUID();
const pending = randomUUID(), expired = randomUUID(), confirmed = randomUUID();
const confirmedAt = new Date('2020-01-01T00:00:00Z');
const rows = [
{ id: nilId, state: 'ended', actor: null }, { id: signed, state: 'ended', actor: 'approver' },
{ id: denied, state: 'ended', actor: 'denier' }, { id: wrongCompany, state: 'ended', actor: 'approver' },
{ id: wrongAction, state: 'ended', actor: 'approver' }, { id: pending, state: 'awaiting_approval', actor: null },
{ id: expired, state: 'expired', actor: null }, { id: confirmed, state: 'ended', actor: null },
].map(row => ({ id: row.id, company_id: company.id, endpoint_id: endpointId,
provider_session_id: `migration:${row.id}`, state: row.state, approved_by_user_id: row.actor,
approval_code: '123456', generation: 1, credential_fingerprint: 'migration', tool_token_hash: 'a'.repeat(64),
request_fingerprint: row.id, expires_at: new Date(), cleanup_confirmed_at: row.id === confirmed ? confirmedAt : null }));
await client`INSERT INTO chat_voice_inbound_calls ${client(rows, 'id', 'company_id', 'endpoint_id',
'provider_session_id', 'state', 'approved_by_user_id', 'approval_code', 'generation',
'credential_fingerprint', 'tool_token_hash', 'request_fingerprint', 'expires_at', 'cleanup_confirmed_at')}`;
await client`INSERT INTO activity_log (company_id, actor_id, action, entity_type, entity_id)
VALUES (${company.id}, 'migration', 'voice.call.ended', 'voice_inbound_call', ${signed}),
(${otherCompany.id}, 'migration', 'voice.call.ended', 'voice_inbound_call', ${wrongCompany}),
(${company.id}, 'migration', 'voice.call.requested', 'voice_inbound_call', ${wrongAction})`;
await client.unsafe(`CREATE TABLE voice_cleanup_backfill_probe (call_id uuid NOT NULL, transaction_id bigint NOT NULL);
CREATE FUNCTION record_voice_cleanup_backfill() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN INSERT INTO voice_cleanup_backfill_probe VALUES (NEW.id, txid_current()); RETURN NEW; END $$;
CREATE TRIGGER record_voice_cleanup_backfill AFTER UPDATE ON chat_voice_inbound_calls
FOR EACH ROW WHEN (OLD.cleanup_confirmed_at IS NULL AND NEW.cleanup_confirmed_at IS NOT NULL)
EXECUTE FUNCTION record_voice_cleanup_backfill()`);
// A failed unique build deterministically leaves an invalid index under
// the migration's name, as cancellation does, without timing a cancel.
await expect(client.unsafe(`CREATE UNIQUE INDEX CONCURRENTLY ${indexName} ON companies ((true))`)).rejects.toThrow();
const [invalid] = await client`SELECT indisvalid FROM pg_index WHERE indexrelid = ${indexName}::regclass`;
expect(invalid.indisvalid).toBe(false);
await Promise.all([applyPendingMigrations(connectionString), applyPendingMigrations(connectionString)]);
const [rebuilt] = await client`SELECT indisvalid, pg_get_indexdef(indexrelid) AS definition
FROM pg_index WHERE indexrelid = ${indexName}::regclass`;
expect(rebuilt.indisvalid).toBe(true);
expect(rebuilt.definition).toContain("chat_voice_inbound_calls");
expect(rebuilt.definition).toContain("cleanup_confirmed_at IS NULL");
const [history] = await client`SELECT count(*)::int AS count FROM drizzle.__drizzle_migrations WHERE hash = ${hash}`;
expect(history.count).toBe(1);
const [backfill] = await client`SELECT count(*)::int AS count, count(DISTINCT transaction_id)::int AS transactions
FROM voice_cleanup_backfill_probe`;
expect(backfill.count).toBe(1007); expect(backfill.transactions).toBeGreaterThanOrEqual(2);
const [batch] = await client`SELECT max(size)::int AS size FROM (
SELECT count(*) AS size FROM voice_cleanup_backfill_probe GROUP BY transaction_id
) groups`;
expect(batch.size).toBeLessThanOrEqual(1000);
const proof = await client`SELECT id, cleanup_confirmed_at FROM chat_voice_inbound_calls
WHERE id = ANY(${[nilId, signed, denied, wrongCompany, wrongAction, pending, expired, confirmed]})`;
for (const id of [nilId, signed]) expect(proof.find(row => row.id === id)?.cleanup_confirmed_at).toBeInstanceOf(Date);
for (const id of [denied, wrongCompany, wrongAction, pending, expired])
expect(proof.find(row => row.id === id)?.cleanup_confirmed_at).toBeNull();
expect(proof.find(row => row.id === confirmed)?.cleanup_confirmed_at.getTime()).toBe(confirmedAt.getTime());
await applyPendingMigrations(connectionString);
const [unchanged] = await client`SELECT count(*)::int AS count FROM voice_cleanup_backfill_probe`;
expect(unchanged.count).toBe(1007);
} finally {
await client?.end();
if (admin && databaseName) await admin.unsafe(`DROP DATABASE "${databaseName}" WITH (FORCE)`);
await admin?.end();
await temporary?.cleanup();
}
}, 120_000);
});
@@ -86,6 +86,7 @@ describe("createChatReconciliationCoordinator", () => {
const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
expect(source).toContain("chatChannels.voice.registerRecoveryWork(deliveryWork,");
expect(source).toContain("chatChannels.voice.registerReplyWork(deliveryWork,");
expect(source).toContain("chatChannels.voice.inbound.registerCleanupWork(deliveryWork,");
expect(source).toContain("chatChannels.voice.reconcileMaintenance()");
expect(source).not.toContain("chatChannels.voice.reconcile()");
});
+40 -26
View File
@@ -3940,32 +3940,46 @@ rl.on("line", (line) => {
});
const rateToolName = (await gateway.listToolsForSession(session.token))
.find((tool) => tool.connectionId === rateTool.connection.id)!.name;
await expect(gateway.executeTool({
sessionToken: session.token,
tool: rateToolName,
parameters: { key: "rate", value: "first" },
})).resolves.toMatchObject({ status: "completed" });
await gateway.executeTool({
sessionToken: session.token,
tool: rateToolName,
parameters: { key: "rate", value: "second" },
}).then(
() => {
throw new Error("Expected connected MCP call to be rate limited");
},
(error) => expectGatewayError(error, 429, "rate_limited"),
);
const [rateLimitedInvocation] = await db
.select()
.from(toolInvocations)
.where(eq(toolInvocations.toolName, rateToolName))
.then((rows) => rows.filter((row) => row.status === "rate_limited"));
expect(rateLimitedInvocation).toMatchObject({
connectionId: rateTool.connection.id,
providerType: "mcp_remote_http",
applicationKey: "rate-app",
errorCode: "rate_limited",
});
const rateClock = Date.now();
vi.useFakeTimers({ toFake: ["Date"] });
vi.setSystemTime(rateClock);
try {
await expect(gateway.executeTool({
sessionToken: session.token,
tool: rateToolName,
parameters: { key: "rate", value: "first" },
})).resolves.toMatchObject({ status: "completed" });
await gateway.executeTool({
sessionToken: session.token,
tool: rateToolName,
parameters: { key: "rate", value: "second" },
}).then(
() => {
throw new Error("Expected connected MCP call to be rate limited");
},
(error) => expectGatewayError(error, 429, "rate_limited"),
);
const [rateLimitedInvocation] = await db
.select()
.from(toolInvocations)
.where(eq(toolInvocations.toolName, rateToolName))
.then((rows) => rows.filter((row) => row.status === "rate_limited"));
expect(rateLimitedInvocation).toMatchObject({
connectionId: rateTool.connection.id,
providerType: "mcp_remote_http",
applicationKey: "rate-app",
errorCode: "rate_limited",
});
vi.setSystemTime(rateClock + 60_000);
await expect(gateway.executeTool({
sessionToken: session.token,
tool: rateToolName,
parameters: { key: "rate", value: "next-window" },
})).resolves.toMatchObject({ status: "completed" });
} finally {
vi.useRealTimers();
}
const otherTool = await createRemoteMcpTool(db, otherCompany.id, {
applicationKey: "other-company-app",
@@ -0,0 +1,278 @@
import { readFile } from "node:fs/promises";
import { chatCompanyDeletion } from "../services/chat-channels/company-deletion.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, chatVoiceInboundCalls, chatVoiceSessions, companies, createDb, issues, toolApplications, toolConnections } from "@paperclipai/db";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
import { assertVoiceInboundCleanupSettled, voiceInboundCleanup } from "../services/voice/voice-inbound-cleanup.js";
import { notifyVoiceInbound } from "../services/voice/voice-inbound-notifications.js";
import { notifyVoiceChange } from "../services/voice/voice-recovery-notifications.js";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_INBOUND_TEST_DATABASE_URL");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("inbound voice cleanup 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[] = [], fixtureCompanies: string[] = [];
const terminate = vi.fn(async (_call: typeof chatVoiceInboundCalls.$inferSelect) => true);
beforeAll(async () => {
if (!external) temporary = await startEmbeddedPostgresTestDatabase("voice-inbound-cleanup-");
if (external) await applyPendingMigrations(external);
db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
async function fixture() {
const companyId = randomUUID(), endpointId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), conversationId = randomUUID();
fixtureCompanies.push(companyId);
const applicationId = randomUUID(), connectionId = randomUUID();
await db.insert(companies).values({ id: companyId, name: "Inbound cleanup", issuePrefix: `V${companyId.slice(0, 7)}` });
await db.insert(agents).values({ id: agentId, companyId, name: "Voice", adapterType: "paperclip_runner" });
await db.insert(issues).values({ id: issueId, companyId, title: "Voice", assigneeAgentId: agentId, status: "in_progress" });
await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Voice", type: "chat" });
await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Voice", uid: connectionId, transport: "voice", connectionPurpose: "channel", enabled: true, status: "active" });
await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, assignedAgentId: agentId, publicId: endpointId, provider: "speko", status: "active", setup: { step: "complete", runtimeGeneration: 1 } });
await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId, externalConversationId: conversationId, externalThreadId: conversationId, externalLabel: "Voice", isDirectMessage: true });
return { companyId, endpointId, agentId, issueId, conversationId };
}
beforeEach(async () => {
enabled = true; errors.length = 0; terminate.mockReset().mockResolvedValue(true);
({ companyId, endpointId, 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(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, companyId));
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 admission(patch: Partial<typeof chatVoiceInboundCalls.$inferInsert> = {}, writer = db) {
const id = randomUUID();
return writer.transaction(async tx => {
await notifyVoiceInbound(tx, { kind: "admission", id });
return (await tx.insert(chatVoiceInboundCalls).values({ id, companyId, endpointId, providerSessionId: randomUUID(), state: "awaiting_approval", approvalCode: "123456", generation: 1, credentialFingerprint: "test", toolTokenHash: "a".repeat(64), requestFingerprint: id, expiresAt: new Date(Date.now() - 1), ...patch }).returning())[0];
});
}
async function owner(call: typeof chatVoiceInboundCalls.$inferSelect, patch: Partial<typeof chatVoiceSessions.$inferInsert> = {}) {
return (await db.insert(chatVoiceSessions).values({ companyId, endpointId, conversationId, issueId, assignedAgentId: agentId, callerId: call.approvedByUserId ?? `guest:${call.id}`, callerAuthority: call.callerAuthority === "member" ? "member" : "guest_intake", approvedByUserId: call.approvedByUserId,
mode: "inbound_phone", state: "active", generation: call.generation, credentialFingerprint: call.credentialFingerprint, providerSessionId: call.providerSessionId, toolTokenHash: call.toolTokenHash, idempotencyKey: `inbound:${call.id}`, requestFingerprint: call.requestFingerprint, expiresAt: new Date(Date.now() + 600_000), ...patch }).returning())[0];
}
async function start(service = voiceInboundCleanup(db, terminate)) {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) });
await service.register(coordinator, () => 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(); }
}
it("bounds pending inventory pages and restores every deadline when the same owner restarts mid-scan", async () => {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false });
const expiry = Date.now() + 60_000;
for (let index = 0; index < 105; index++) await admission({ expiresAt: new Date(expiry) });
const counts: number[] = [], select = db.select.bind(db);
const reads = vi.spyOn(db, "select").mockImplementation((...args) => {
const selection = select(...args), from = selection.from.bind(selection);
vi.spyOn(selection, "from").mockImplementation(table => {
const query = from(table);
if (table === chatVoiceInboundCalls) {
const execute = query.execute.bind(query);
vi.spyOn(query, "execute").mockImplementation(async () => {
const rows = await execute(); counts.push(rows.length); return rows;
});
}
return query;
});
return selection;
});
try {
const service = await start();
expect(counts.reduce((sum, count) => sum + count, 0)).toBe(100);
await coordinator!.stop(); coordinator = undefined;
const beforeRestart = counts.length;
await start(service);
expect(counts.slice(beforeRestart).reduce((sum, count) => sum + count, 0)).toBe(100);
await vi.advanceTimersByTimeAsync(1000);
await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(expiry), { interval: 0 });
expect(Math.max(...counts)).toBeLessThanOrEqual(100); reads.mockRestore();
await quiet(1000); expect(terminate).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(expiry - Date.now()); await settled();
expect(terminate).toHaveBeenCalledTimes(105);
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, companyId)))
.every(call => call.cleanupConfirmedAt !== null)).toBe(true);
await quiet();
} finally { reads.mockRestore(); }
});
it("leaves configured idle and confirmed history quiet", async () => {
await admission({ state: "ended", cleanupConfirmedAt: new Date() });
await start(); await settled(); await quiet(); expect(terminate).not.toHaveBeenCalled();
});
it("prevents company deletion from erasing an unfinished provider cleanup", async () => {
const call = await admission();
await expect(db.transaction(tx => chatCompanyDeletion.deleteCompanyData(tx, companyId))).rejects.toMatchObject({ status: 409 });
expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id))).toHaveLength(1);
await start(); await settled();
await db.transaction(tx => chatCompanyDeletion.deleteCompanyData(tx, companyId));
expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id))).toHaveLength(0);
});
it("backfills only provable legacy termination and preserves rejected cleanup debt", async () => {
const confirmed = await admission({ state: "ended" });
const uncertain = await admission({ state: "ended", approvedByUserId: randomUUID(), callerAuthority: "member" });
const migration = await readFile(new URL("../../../packages/db/src/migrations/0396_tan_paladin.sql", import.meta.url), "utf8");
const backfill = migration.split("--> statement-breakpoint").at(-1)!;
await db.$client.unsafe(backfill); await db.$client.unsafe(backfill);
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, confirmed.id)))[0].cleanupConfirmedAt).not.toBeNull();
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, uncertain.id)))[0].cleanupConfirmedAt).toBeNull();
await start(); await settled(); expect(terminate).toHaveBeenCalledOnce();
expect(terminate.mock.calls[0][0].id).toBe(uncertain.id); await quiet();
});
it("joins provider cleanup at shutdown and saves a restart deadline for uncertainty", async () => {
const call = await admission(); let release!: (confirmed: boolean) => void;
terminate.mockImplementationOnce(() => new Promise<boolean>(resolve => { release = resolve; }));
const service = voiceInboundCleanup(db, terminate);
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
const worker = service.register(coordinator, () => true);
await vi.waitFor(() => expect(terminate).toHaveBeenCalledOnce());
let stopped = false;
const stopping = coordinator.stop().then(() => { stopped = true; });
try {
await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false);
release(false); await stopping; await worker.ready;
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0].cleanupNextAttemptAt).not.toBeNull();
await start(); await settled(); expect(terminate).toHaveBeenCalledTimes(2);
} finally { release(false); await stopping; }
});
it("waits for the original expiry and reconstructs it on restart", async () => {
const due = Date.now() + 60_000; await admission({ expiresAt: new Date(due) });
await start(); expect(coordinator!.nextWakeAt()).toBe(due); await quiet(10_000);
await coordinator!.stop(); await start(); expect(coordinator!.nextWakeAt()).toBe(due);
await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); expect(terminate).toHaveBeenCalledOnce(); await quiet();
});
it("waits for outer commit and ignores a rolled-back nested producer", async () => {
await start(); await settled();
await expect(db.transaction(async tx => { await admission({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(2000); expect(terminate).not.toHaveBeenCalled(); throw new Error("rollback"); })).rejects.toThrow("rollback");
await settled(); await quiet();
await db.transaction(async tx => { await admission({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(2000); expect(terminate).not.toHaveBeenCalled(); });
await settled(); expect(terminate).toHaveBeenCalledOnce(); await quiet();
});
it("retains unconfirmed termination across restart without early requests", async () => {
terminate.mockResolvedValueOnce(false);
const call = await admission(); await start();
const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id));
expect(pending.cleanupConfirmedAt).toBeNull(); expect(pending.cleanupNextAttemptAt).not.toBeNull();
await coordinator!.stop(); await start(); await quiet(10_000); expect(terminate).toHaveBeenCalledOnce();
await vi.advanceTimersByTimeAsync(pending.cleanupNextAttemptAt!.getTime() - Date.now()); await settled();
expect(terminate).toHaveBeenCalledTimes(2); await quiet();
});
it("rechecks ownership after discovery rather than ending a newly bound call", async () => {
const call = await admission({ state: "approving", approvedByUserId: randomUUID(), callerAuthority: "member", requestedIssueId: issueId });
const service = voiceInboundCleanup(db, terminate), [candidate] = await service.load();
const session = await owner(call);
await db.update(chatVoiceInboundCalls).set({ state: "approved", sessionId: session.id }).where(eq(chatVoiceInboundCalls.id, call.id));
expect(await service.check(candidate.id)).toBeNull(); expect(terminate).not.toHaveBeenCalled();
});
it.each([false, true])("repairs older interrupted handoffs (existing guest binding=%s)", async bound => {
const call = await admission({ state: "approving", approvedByUserId: randomUUID(), callerAuthority: "member", requestedIssueId: issueId });
const session = await owner(call);
if (bound) await db.update(chatVoiceInboundCalls).set({ sessionId: session.id }).where(eq(chatVoiceInboundCalls.id, call.id));
await start(); await settled(); expect(terminate).not.toHaveBeenCalled();
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0]).toMatchObject({ state: "approved", sessionId: session.id });
await quiet();
});
it("hands rejected bound calls to live recovery and settles from its confirmation", async () => {
const call = await admission({ state: "denied" }), session = await owner(call);
await db.update(chatVoiceInboundCalls).set({ sessionId: session.id }).where(eq(chatVoiceInboundCalls.id, call.id));
await start(); await settled(); expect(terminate).not.toHaveBeenCalled();
expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, session.id)))[0].state).toBe("ending");
await quiet();
await db.transaction(async tx => {
await notifyVoiceChange(tx, { kind: "session", id: session.id });
await tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
});
await settled();
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0].cleanupConfirmedAt).not.toBeNull();
await quiet(); expect(terminate).not.toHaveBeenCalled();
});
it.each(["ended", "expired"] as const)("settles terminal legacy %s admissions without a session binding", async state => {
const call = await admission({ state, approvedByUserId: randomUUID(), callerAuthority: "member", requestedIssueId: issueId });
const endedAt = new Date(Date.now() - 20_000);
await owner(call, { state: "ended", endedAt });
const service = voiceInboundCleanup(db, terminate);
await expect(assertVoiceInboundCleanupSettled(db, companyId)).rejects.toMatchObject({ status: 409 });
expect(await service.check(call.id)).toBeNull();
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0])
.toMatchObject({ state, cleanupConfirmedAt: endedAt, cleanupNextAttemptAt: null });
await start(service); await settled(); await quiet();
expect(terminate).not.toHaveBeenCalled();
await assertVoiceInboundCleanupSettled(db, companyId);
});
it.each(["denied", "expired", "ended"] as const)("hands rejected legacy %s admissions without a binding to live recovery", async state => {
const call = await admission({ state, approvedByUserId: randomUUID(), callerAuthority: "member", requestedIssueId: issueId });
const session = await owner(call);
const service = voiceInboundCleanup(db, terminate);
expect(await service.check(call.id)).toBeNull();
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0])
.toMatchObject({ state, sessionId: session.id, cleanupConfirmedAt: null });
expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, session.id)))[0].state).toBe("ending");
await start(service); await settled(); await quiet();
await db.transaction(async tx => {
await notifyVoiceChange(tx, { kind: "session", id: session.id });
await tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
});
await settled();
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0].cleanupConfirmedAt).not.toBeNull();
await quiet(); expect(terminate).not.toHaveBeenCalled();
});
it.each([
{ mode: "outbound_phone" as const },
{ idempotencyKey: "another-admission" },
{ generation: 2 },
{ credentialFingerprint: "another-credential" },
{ toolTokenHash: "b".repeat(64) },
])("retains rejected legacy cleanup debt when session evidence conflicts: %j", async patch => {
const call = await admission({ state: "expired" });
const session = await owner(call, patch);
await expect(voiceInboundCleanup(db, terminate).check(call.id)).rejects.toMatchObject({ status: 409 });
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0])
.toMatchObject({ state: "expired", sessionId: null, cleanupConfirmedAt: null });
expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, session.id)))[0].state).toBe("active");
expect(terminate).not.toHaveBeenCalled();
});
it("retains first work committed during drain without passive SQL", async () => {
await start(); await settled(); enabled = false;
await admission(); await quiet(60_000); expect(terminate).not.toHaveBeenCalled();
enabled = true; await settled(); expect(terminate).toHaveBeenCalledOnce(); await quiet();
});
it("pages backlog fairly while a failed provider retains its own retry", async () => {
const broken = await admission();
for (let i = 0; i < 104; i++) await admission();
const other = await fixture(), last = await admission({ companyId: other.companyId, endpointId: other.endpointId });
let failed = true;
terminate.mockImplementation(async call => { if (call.id === broken.id && failed) throw new Error("provider unavailable"); return true; });
await start();
expect(terminate.mock.calls.slice(0, 25).some(([call]) => call.id === last.id)).toBe(true);
// waitFor already advances fake time. Extra one-second jumps can reach
// the failed provider's retry while ordinary database writes are pending.
await vi.waitFor(() => {
expect(terminate.mock.calls.filter(([call]) => call.id !== broken.id)).toHaveLength(105);
}, { timeout: 5000 });
expect(errors).toHaveLength(1); failed = false;
await settled(); errors.length = 0; await quiet();
});
});
@@ -25,6 +25,7 @@ import { applyPendingMigrations, chatActions, instanceUserRoles, principalPermis
import { configureSpekoSessionTools } from "../services/voice/speko-tool-setup.js";
import { instanceSettingsService } from "../services/instance-settings.js";
import { chatChannelService } from "../services/chat-channels.js";
import { voiceInboundService } from "../services/voice/voice-inbound-service.js";
import { voiceSessionService } from "../services/voice/voice-session-service.js";
import { captureVoiceDeliveryDiagnostics } from "../services/voice/voice-delivery-diagnostics.js";
import { parseSpekoDeliveryDiagnostics } from "../services/voice/speko-delivery-diagnostics.js";
@@ -79,6 +80,126 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication };
}
it("returns a committed denial while provider cleanup is still waiting", async () => {
const f = await inboundFixture();
let release!: () => void;
const pendingProvider = new Promise<void>(resolve => { release = resolve; });
f.transport.verifyAgent.mockImplementation(async () => {
await pendingProvider;
return { id: "agent_test", organizationId: `org_${f.companyId}`, name: "Test voice" };
});
const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
let denial: ReturnType<typeof f.service.inbound.decide> | undefined;
try {
await f.service.inbound.registerCleanupWork(coordinator, () => true).ready;
let returned = false;
denial = f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller,
{ approve: false, approvalCode: f.call.approvalCode }).then(result => { returned = true; return result; });
await vi.waitFor(() => expect(returned).toBe(true), { timeout: 2000 });
expect(await denial).toMatchObject({ state: "denied" });
await vi.waitFor(() => expect(f.transport.verifyAgent).toHaveBeenCalled(), { timeout: 5000 });
expect(f.transport.endSession).not.toHaveBeenCalled();
const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
expect(pending.cleanupConfirmedAt).toBeNull();
expect(pending.cleanupNextAttemptAt).not.toBeNull();
release();
await vi.waitFor(async () => {
const [settled] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
expect(settled.cleanupConfirmedAt).not.toBeNull();
}, { timeout: 5000 });
expect(f.transport.endSession).toHaveBeenCalledOnce(); expect(onError).not.toHaveBeenCalled();
} finally {
release(); await denial; await coordinator.stop();
}
});
it("wakes inbound cleanup from a real signed pre-call and retains an unconfirmed end", async () => {
const f = await inboundFixture();
const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" });
await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers);
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
const settle = () => vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0);
}, { timeout: 5000 });
const quiet = async (ms: number) => {
const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction");
try { await vi.advanceTimersByTimeAsync(ms); expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); }
finally { reads.mockRestore(); transactions.mockRestore(); }
};
try {
await f.service.inbound.registerCleanupWork(coordinator, () => true).ready;
await settle(); expect(coordinator.nextWakeAt()).toBeNull(); await quiet(10_000);
const providerSessionId = randomUUID(), incoming = f.sign({ ...f.event, call_id: providerSessionId, session_id: providerSessionId });
await f.service.inbound.lifecycle(f.endpoint.publicId, incoming.body, incoming.headers);
await settle();
const [call] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.providerSessionId, providerSessionId));
expect(coordinator.nextWakeAt()).toBe(call.expiresAt.getTime());
await quiet(60_000); expect(f.transport.inspectSession).not.toHaveBeenCalled();
// Disabling new voice work must not erase an existing physical cleanup.
await instanceSettingsService(db).updateExperimental({ enableChatConnectors: false });
await settle();
f.credentials.mockResolvedValue({ apiKey: "rotated-key-in-original-account", agentId: "agent_test" });
f.transport.endSession.mockResolvedValueOnce({ confirmed: false });
await vi.advanceTimersByTimeAsync(call.expiresAt.getTime() - Date.now()); await settle();
expect(f.transport.endSession).toHaveBeenCalledWith(providerSessionId);
const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id));
expect(pending.cleanupConfirmedAt).toBeNull();
expect(coordinator.nextWakeAt()).toBe(pending.cleanupNextAttemptAt!.getTime());
await quiet(10_000); expect(f.transport.endSession).toHaveBeenCalledOnce();
await vi.advanceTimersByTimeAsync(pending.cleanupNextAttemptAt!.getTime() - Date.now()); await settle();
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(coordinator.nextWakeAt()).toBeNull(); });
expect(f.transport.endSession).toHaveBeenCalledTimes(2); await quiet(120_000);
expect(onError).not.toHaveBeenCalled();
expect(f.transport.createPhoneSession).not.toHaveBeenCalled();
} finally { await coordinator.stop(); vi.useRealTimers(); await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true }); }
});
it("retains rejected inbound cleanup across provider failure and history retention", async () => {
const f = await inboundFixture(true);
f.transport.endSession.mockRejectedValueOnce(new Error("provider unavailable"));
const result = await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller,
{ approve: false, approvalCode: f.call.approvalCode });
expect(result.state).toBe("ended");
await f.service.inbound.reconcile(1000);
const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
expect(pending.cleanupConfirmedAt).toBeNull(); expect(pending.cleanupNextAttemptAt).not.toBeNull();
await db.update(chatVoiceInboundCalls).set({ expiresAt: new Date(0), updatedAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id));
await f.service.inbound.pruneHistory();
expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(1);
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
try {
await f.service.inbound.registerCleanupWork(coordinator, () => true).ready;
expect(coordinator.nextWakeAt()).toBe(pending.cleanupNextAttemptAt!.getTime());
await vi.advanceTimersByTimeAsync(pending.cleanupNextAttemptAt!.getTime() - Date.now());
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); });
expect(f.transport.endSession).toHaveBeenCalledTimes(2);
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)))[0].cleanupConfirmedAt).not.toBeNull();
expect(onError).not.toHaveBeenCalled();
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it("does not clean up an inbound call through a different provider account", async () => {
const f = await inboundFixture();
f.credentials.mockResolvedValue({ apiKey: "key-for-another-account", agentId: "other-agent" });
f.transport.verifyAgent.mockResolvedValue({ id: "other-agent", organizationId: "another-account", name: "Other voice" });
await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller,
{ approve: false, approvalCode: f.call.approvalCode });
await f.service.inbound.reconcile(1000);
expect(f.transport.inspectSession).not.toHaveBeenCalled(); expect(f.transport.endSession).not.toHaveBeenCalled();
const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
expect(pending.cleanupConfirmedAt).toBeNull(); expect(pending.cleanupNextAttemptAt).not.toBeNull();
// A usable key for the saved account restores cleanup without redialing.
f.credentials.mockResolvedValue({ apiKey: "restored-account-key", agentId: "agent_test" });
f.transport.verifyAgent.mockResolvedValue({ id: "agent_test", organizationId: `org_${f.companyId}`, name: "Test voice" });
await db.update(chatVoiceInboundCalls).set({ cleanupNextAttemptAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id));
await f.service.inbound.reconcile(1000);
expect(f.transport.endSession).toHaveBeenCalledWith(f.call.providerSessionId);
expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)))[0].cleanupConfirmedAt).not.toBeNull();
expect(f.transport.createPhoneSession).not.toHaveBeenCalled();
});
it("restores blocked replies through the production live-call recovery wiring", async () => {
const f = await pushFixture();
await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.principalId, f.callerId));
@@ -1187,6 +1308,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
await instanceSettingsService(db).updateExperimental({enableChatConnectors: true, enableIsolatedWorkspaces: guestIntake});
const signingSecret = `whsec_${Buffer.alloc(32, 7).toString("base64")}`;
f.credentials.mockResolvedValue({apiKey: "server-only-key", agentId: "agent_test", signingSecret} as never);
f.transport.verifyAgent.mockResolvedValue({ id: "agent_test", organizationId: `org_${f.companyId}`, name: "Test voice" });
await db.update(chatEndpoints).set({providerAccountId: `org_${f.companyId}`, botExternalId: "agent_test"}).where(eq(chatEndpoints.id, f.endpointId));
await db.insert(chatVoicePhoneLines).values({companyId: f.companyId, endpointId: f.endpointId, providerNumberId: randomUUID(), phoneNumber: "+12015550123", enabled: true, guestIntake, lowTrustEnvironmentId: sandboxId ?? null});
const [line] = await db.select().from(chatVoicePhoneLines).where(eq(chatVoicePhoneLines.endpointId, f.endpointId));
@@ -1359,6 +1481,78 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
expect(repeated?.latestRevisionId).toBe(first?.latestRevisionId);
expect(await docs.listIssueDocuments(f.call.intakeIssueId!)).toHaveLength(1);
});
it.each([false, true])("hands inbound cleanup ownership to the authorized session atomically (guest=%s)", async guest => {
const f = await inboundFixture(guest);
if (guest) await f.tool("submit_request", { text: "A public request before private approval" });
let release!: () => void, bound!: () => void;
const bindingReached = new Promise<void>(resolve => { bound = resolve; });
const releaseApproval = new Promise<void>(resolve => { release = resolve; });
const inbound = voiceInboundService(db, {
allowLocalBoard: false, credentials: f.credentials, provider: () => f.transport,
voicePrompt: "Test voice", lowTrustVoicePrompt: "Test guest voice",
start: async input => {
const result = await f.service.start(input);
bound(); await releaseApproval;
return result;
},
});
vi.useFakeTimers({ toFake: ["Date"] });
const approval = inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller,
{ approve: true, approvalCode: f.call.approvalCode, issueId: f.issueId });
try {
await Promise.race([bindingReached, approval.then(() => { throw new Error("Approval did not reach session binding"); })]);
// The session has committed but its caller has not received the result.
// Expiry or process death here must not leave two cleanup owners.
vi.setSystemTime(f.call.expiresAt.getTime() + 1);
await inbound.reconcile(1000);
expect(f.transport.endSession).not.toHaveBeenCalledWith(f.call.providerSessionId);
const [call] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
expect(call).toMatchObject({ state: "approved" });
expect(call.sessionId).not.toBeNull();
const [session] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, call.sessionId!));
expect(session).toMatchObject({ providerSessionId: f.call.providerSessionId, callerId: f.callerId, issueId: f.issueId });
release();
await expect(approval).resolves.toMatchObject({ state: "approved", sessionId: session.id });
} finally { release(); await approval.catch(() => undefined); vi.useRealTimers(); }
});
it.each([false, true])("finishes an interrupted inbound approval using its existing live session (guest=%s)", async guest => {
const f = await inboundFixture(guest);
if (guest) await f.tool("submit_request", { text: "A public request before private approval" });
const input = { approve: true, approvalCode: f.call.approvalCode, issueId: f.issueId };
const approved = await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, input);
// Older versions committed session creation/promotion before a separate
// admission binding. Reconstruct a crash in that gap, then retry the API.
await db.update(chatVoiceInboundCalls).set({ state: "approving", sessionId: guest ? approved.sessionId : null })
.where(eq(chatVoiceInboundCalls.id, f.call.id));
await expect(f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, input))
.resolves.toMatchObject({ state: "approved", sessionId: approved.sessionId });
await expect(f.service.start({ companyId: f.companyId, endpointId: f.endpointId, issueId: f.issueId,
caller: f.input.caller, idempotencyKey: `inbound:${f.call.id}`, maxDurationSeconds: 600,
mode: "inbound_phone", providerSessionId: f.event.session_id, toolToken: f.capability }))
.resolves.toMatchObject({ session: { id: approved.sessionId } });
const sessions = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.providerSessionId, f.event.session_id));
expect(sessions).toHaveLength(1);
expect(sessions[0]).toMatchObject({ id: approved.sessionId, callerId: f.callerId, issueId: f.issueId });
expect(f.transport.createBrowserSession).not.toHaveBeenCalled();
expect(f.transport.createPhoneSession).not.toHaveBeenCalled();
expect(f.transport.endSession).not.toHaveBeenCalled();
});
it.each(["ended", "expired", "token"] as const)("does not repair an interrupted approval after its session is %s", async change => {
const f = await inboundFixture();
const input = { approve: true, approvalCode: f.call.approvalCode, issueId: f.issueId };
const approved = await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, input);
await db.update(chatVoiceInboundCalls).set({ state: "approving", sessionId: null }).where(eq(chatVoiceInboundCalls.id, f.call.id));
await db.update(chatVoiceSessions).set(change === "ended" ? { state: "ended", endedAt: new Date() }
: change === "expired" ? { createdAt: new Date(Date.now() - 2000), expiresAt: new Date(Date.now() - 1000) } : { toolTokenHash: "a".repeat(64) })
.where(eq(chatVoiceSessions.id, approved.sessionId!));
await expect(f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, input))
.rejects.toMatchObject({ status: 403 });
const [call] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
expect(call).toMatchObject({ state: "approving", sessionId: null });
expect(await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.providerSessionId, f.event.session_id))).toHaveLength(1);
});
it("requires approval of the specific incoming call before delivering any private work", async () => {
const f = await inboundFixture();
expect(await f.service.inbound.lifecycle(f.endpoint.publicId, f.signed.body, f.signed.headers)).toEqual(f.response);
@@ -1418,6 +1612,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
it("denies expired and rejected calls and closes the existing provider leg without creating work", async () => {
const f = await inboundFixture();
expect(await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, {approve: false, approvalCode: f.call.approvalCode})).toMatchObject({state: "denied"});
await f.service.inbound.reconcile(1000);
expect(f.transport.endSession).toHaveBeenCalledWith(f.event.session_id);
expect(await f.tool("submit_request", {text: "Try again"})).toMatchObject({authorization: "denied", requestAccepted: false});
const expired = await inboundFixture();
+1
View File
@@ -1300,6 +1300,7 @@ export async function createApp(
registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
chatChannels.voice.registerRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.voice.registerReplyWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
emailChannels.start();
@@ -1,9 +1,14 @@
import { assertVoiceInboundCleanupSettled } from "../voice/voice-inbound-cleanup.js";
import { chatEndpoints } from "@paperclipai/db";
import { eq } from "drizzle-orm";
import type { CompanyDeletionParticipant } from "../../lib/company-deletion.js";
export const chatCompanyDeletion: CompanyDeletionParticipant = {
async deleteCompanyData(tx, companyId) {
// Ingress locks the endpoint before admitting a provider call. Hold that
// lock through the cleanup proof and delete so a new call cannot slip in.
await tx.select({ id: chatEndpoints.id }).from(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)).for("update");
await assertVoiceInboundCleanupSettled(tx, companyId);
await tx.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId));
},
};
@@ -45,6 +45,7 @@ export const DELIVERY_QUEUES = {
chatTelegramMaintenance: "chat-telegram-maintenance",
chatReceivers: "chat-receivers",
voiceSessionRecovery: "voice-session-recovery",
voiceInboundCleanup: "voice-inbound-cleanup",
voiceReplies: "voice-reply-delivery",
chatCommandRegistration: "chat-command-registration",
chatTelegramSubscriptions: "chat-telegram-subscriptions",
@@ -0,0 +1,155 @@
import { and, eq, gt, isNull } from "drizzle-orm";
import { chatEndpoints, chatVoiceInboundCalls, chatVoiceSessions, type Db } from "@paperclipai/db";
import { conflict } from "../../errors.js";
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
import { notifyVoiceInbound, voiceInboundScope, type VoiceInboundScope } from "./voice-inbound-notifications.js";
import { notifyVoiceRecovery } from "./voice-recovery-notifications.js";
type Admission = typeof chatVoiceInboundCalls.$inferSelect;
type Session = typeof chatVoiceSessions.$inferSelect;
export type VoiceInboundEntry = { id: string; sessionId: string | null; companyId: string; endpointId: string; connectionId: string; dueAt: number };
const terminal = (session: Session) => ["ended", "failed", "expired"].includes(session.state) && session.endedAt !== null;
const rejected = (call: Admission) => ["denied", "expired", "ended"].includes(call.state);
/** A bound call belongs to live-session recovery. Admission cleanup may only
* hand off a rejection or copy confirmed termination; it never ends that leg. */
export function inboundCleanupAt(call: Admission, session: Session | null): number | null {
if (call.cleanupConfirmedAt) return null;
if (session) {
if (terminal(session) || (call.state === "approving" && session.callerAuthority !== "guest_intake")) return 0;
return rejected(call) && session.state !== "ending" ? 0 : null;
}
if (call.cleanupNextAttemptAt) return call.cleanupNextAttemptAt.getTime();
return ["awaiting_approval", "guest_intake", "approving"].includes(call.state) ? call.expiresAt.getTime() : 0;
}
/** Caller locks company endpoints before this check, matching admission ingress. */
export async function assertVoiceInboundCleanupSettled(db: Pick<Db, "select">, companyId: string) {
const [pending] = await db.select({ id: chatVoiceInboundCalls.id }).from(chatVoiceInboundCalls)
.where(and(eq(chatVoiceInboundCalls.companyId, companyId), isNull(chatVoiceInboundCalls.cleanupConfirmedAt))).limit(1);
if (pending) throw conflict("Incoming phone calls must finish cleanup before deleting this company");
}
export function voiceInboundCleanup(db: Db, terminate: (call: Admission) => Promise<boolean>) {
const cursors = new Map<string, string>();
async function page(scope?: VoiceInboundScope, cursor?: string, signal?: AbortSignal) {
signal?.throwIfAborted();
const rows = await db.select({ call: chatVoiceInboundCalls, session: chatVoiceSessions, connectionId: chatEndpoints.connectionId })
.from(chatVoiceInboundCalls)
.innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceInboundCalls.endpointId), eq(chatEndpoints.companyId, chatVoiceInboundCalls.companyId)))
.leftJoin(chatVoiceSessions, and(eq(chatVoiceSessions.id, chatVoiceInboundCalls.sessionId), eq(chatVoiceSessions.companyId, chatVoiceInboundCalls.companyId)))
.where(and(isNull(chatVoiceInboundCalls.cleanupConfirmedAt),
scope?.kind === "admission" ? eq(chatVoiceInboundCalls.id, scope.id) : undefined,
scope?.kind === "session" ? eq(chatVoiceInboundCalls.sessionId, scope.id) : undefined,
scope?.kind === "endpoint" ? eq(chatVoiceInboundCalls.endpointId, scope.id) : undefined,
scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined,
scope?.kind === "company" ? eq(chatVoiceInboundCalls.companyId, scope.id) : undefined,
cursor ? gt(chatVoiceInboundCalls.id, cursor) : undefined))
.orderBy(chatVoiceInboundCalls.id).limit(100);
signal?.throwIfAborted();
return rows;
}
function entries(rows: Awaited<ReturnType<typeof page>>): VoiceInboundEntry[] {
const result: VoiceInboundEntry[] = [];
for (const { call, session, connectionId } of rows) {
const dueAt = inboundCleanupAt(call, session);
if (dueAt !== null) result.push({ id: call.id, sessionId: call.sessionId, companyId: call.companyId, endpointId: call.endpointId, connectionId, dueAt });
}
return result;
}
async function load(scope?: VoiceInboundScope, signal?: AbortSignal): Promise<VoiceInboundEntry[]> {
const result: VoiceInboundEntry[] = [];
let cursor: string | undefined;
for (;;) {
const rows = await page(scope, cursor, signal);
result.push(...entries(rows));
if (rows.length < 100) return result;
cursor = rows[rows.length - 1].call.id;
}
}
async function refresh(scope?: VoiceInboundScope, signal?: AbortSignal) {
const key = scope ? `${scope.kind}:${scope.id}` : "*";
const rows = await page(scope, cursors.get(key), signal), result = entries(rows), complete = rows.length < 100;
signal?.throwIfAborted();
if (complete) cursors.delete(key); else cursors.set(key, rows[rows.length - 1].call.id);
return { rows: result, complete };
}
async function check(id: string, signal?: AbortSignal): Promise<VoiceInboundEntry | null> {
signal?.throwIfAborted();
const claimed = await db.transaction(async tx => {
// Discovery grants no ownership. Approval and lifecycle use this same lock,
// always before the session lock; no provider request runs in this TX.
const [call] = await tx.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, id)).for("update");
if (!call || call.cleanupConfirmedAt) return null;
const [session] = await tx.select().from(chatVoiceSessions).where(and(
eq(chatVoiceSessions.companyId, call.companyId), eq(chatVoiceSessions.endpointId, call.endpointId),
eq(chatVoiceSessions.providerSessionId, call.providerSessionId))).for("update");
if (session) {
if (terminal(session)) {
if (call.sessionId && call.sessionId !== session.id) throw conflict("Incoming call cleanup has conflicting session ownership");
await notifyVoiceInbound(tx, { kind: "admission", id });
await tx.update(chatVoiceInboundCalls).set({ ...(!rejected(call) ? { state: "ended" as const } : {}), cleanupConfirmedAt: session.endedAt, cleanupNextAttemptAt: null, updatedAt: new Date() }).where(eq(chatVoiceInboundCalls.id, id));
return null;
}
if (rejected(call)) {
if (call.sessionId !== session.id) {
if (call.sessionId || session.mode !== "inbound_phone" || session.idempotencyKey !== `inbound:${call.id}`
|| session.generation !== call.generation || session.credentialFingerprint !== call.credentialFingerprint
|| session.toolTokenHash !== call.toolTokenHash) throw conflict("Incoming call cleanup has conflicting session ownership");
await notifyVoiceInbound(tx, { kind: "admission", id });
await tx.update(chatVoiceInboundCalls).set({ sessionId: session.id, updatedAt: new Date() }).where(eq(chatVoiceInboundCalls.id, id));
}
if (session.state !== "ending") {
await notifyVoiceRecovery(tx, { kind: "session", id: session.id });
await tx.update(chatVoiceSessions).set({ state: "ending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
}
return null;
}
if (call.sessionId !== session.id || (call.state === "approving" && session.callerAuthority !== "guest_intake")) {
// Repair older commits that inserted/promoted an authorized session
// before the separate admission binding write. Never infer approval
// from a matching provider ID alone.
if (!["approving", "approved"].includes(call.state) || !call.approvedByUserId
|| session.idempotencyKey !== `inbound:${call.id}` || session.callerId !== call.approvedByUserId
|| session.approvedByUserId !== call.approvedByUserId || session.callerAuthority !== call.callerAuthority
|| session.generation !== call.generation || session.credentialFingerprint !== call.credentialFingerprint
|| session.toolTokenHash !== call.toolTokenHash || (call.requestedIssueId !== null && session.issueId !== call.requestedIssueId))
throw conflict("Incoming call cleanup has conflicting session ownership");
await notifyVoiceInbound(tx, { kind: "admission", id });
await tx.update(chatVoiceInboundCalls).set({ state: "approved", sessionId: session.id, updatedAt: new Date() }).where(eq(chatVoiceInboundCalls.id, id));
call.state = "approved"; call.sessionId = session.id;
}
return null;
}
if (call.sessionId) throw conflict("Incoming call cleanup owner is missing");
const dueAt = inboundCleanupAt(call, null);
if (dueAt === null || dueAt > Date.now()) return null;
await notifyVoiceInbound(tx, { kind: "admission", id });
const [claim] = await tx.update(chatVoiceInboundCalls).set({
...(!rejected(call) ? { state: "expired" as const } : {}),
cleanupNextAttemptAt: new Date(Date.now() + 30_000), updatedAt: new Date(),
}).where(eq(chatVoiceInboundCalls.id, id)).returning();
return claim;
});
if (claimed) {
signal?.throwIfAborted();
if (await terminate(claimed)) await db.transaction(async tx => {
await notifyVoiceInbound(tx, { kind: "admission", id });
await tx.update(chatVoiceInboundCalls).set({ cleanupConfirmedAt: new Date(), cleanupNextAttemptAt: null, updatedAt: new Date() })
.where(and(eq(chatVoiceInboundCalls.id, id), isNull(chatVoiceInboundCalls.cleanupConfirmedAt), eq(chatVoiceInboundCalls.cleanupNextAttemptAt, claimed.cleanupNextAttemptAt!)));
});
}
return (await load({ kind: "admission", id }, signal))[0] ?? null;
}
function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
cursors.clear();
return registerScopedDeadlineWork(coordinator, {
queue: DELIVERY_QUEUES.voiceInboundCleanup, decodeScope: voiceInboundScope,
matches: (entry, scope) => !scope || (scope.kind === "admission" ? entry.id : scope.kind === "session" ? entry.sessionId : scope.kind === "endpoint" ? entry.endpointId : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id,
load: refresh, check: (entry, signal) => check(entry.id, signal),
}, canRun);
}
return { load, check, register };
}
@@ -0,0 +1,13 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
export type VoiceInboundScope = { kind: "admission" | "session" | "endpoint" | "connection" | "company"; id: string };
export function voiceInboundScope(hint: string): VoiceInboundScope | undefined {
if (hint === "*") return undefined;
const [kind, id, extra] = hint.split(":");
if (!id || extra !== undefined || !["admission", "session", "endpoint", "connection", "company"].includes(kind))
throw new Error("Invalid inbound voice cleanup scope");
return { kind: kind as VoiceInboundScope["kind"], id };
}
export function notifyVoiceInbound(tx: object, scope?: VoiceInboundScope) {
return notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceInboundCleanup, scope ? `${scope.kind}:${scope.id}` : "*");
}
@@ -1,7 +1,9 @@
import { voiceInboundCleanup } from "./voice-inbound-cleanup.js";
import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
import { notifyVoiceChange } 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";
import { and, asc, desc, eq, gte, inArray, lte, isNull, isNotNull, sql } from "drizzle-orm";
import { z } from "zod";
import { chatEndpointResources, chatConversations, chatActions, chatEndpoints, chatVoiceInboundCalls, chatVoicePhoneLines, chatVoiceReports, chatVoiceSessions, activityLog, type Db } from "@paperclipai/db";
import { LOW_TRUST_REVIEW_PRESET, LOW_TRUST_REVIEW_PRESET_VERSION, LOW_TRUST_REVIEW_RAW_OUTPUT_DISPOSITION } from "@paperclipai/shared";
@@ -120,7 +122,8 @@ export function voiceInboundService(db: Db, options: {
if (prior) return prior;
const recent = await tx.select({state: chatVoiceInboundCalls.state, expiresAt: chatVoiceInboundCalls.expiresAt}).from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, row.companyId), eq(chatVoiceInboundCalls.endpointId, row.id), gte(chatVoiceInboundCalls.createdAt, new Date(Date.now() - 3_600_000)))).limit(60);
if (recent.length >= 60 || recent.filter(call => ["guest_intake", "awaiting_approval", "approving"].includes(call.state) && call.expiresAt.getTime() > Date.now()).length >= 10) throw conflict("This phone line has reached its incoming call limit. Try again later.");
const [admission] = await tx.insert(chatVoiceInboundCalls).values({companyId: row.companyId, endpointId: row.id, providerSessionId, state: line.guestIntake ? "guest_intake" : "awaiting_approval", approvalCode: String(randomInt(100000, 1000000)), generation: context.generation, credentialFingerprint: context.fingerprint, toolTokenHash: voiceTokenHash(capability), requestFingerprint: fingerprint, callerAuthority: line.guestIntake ? "guest_intake" : null, expiresAt: new Date(Date.now() + (line.guestIntake ? 600_000 : 120_000))}).onConflictDoNothing().returning();
await notifyVoiceInbound(tx, { kind: "endpoint", id: row.id });
const [admission] = await tx.insert(chatVoiceInboundCalls).values({companyId: row.companyId, endpointId: row.id, providerSessionId, providerAccountId: event.organization_id, state: line.guestIntake ? "guest_intake" : "awaiting_approval", approvalCode: String(randomInt(100000, 1000000)), generation: context.generation, credentialFingerprint: context.fingerprint, toolTokenHash: voiceTokenHash(capability), requestFingerprint: fingerprint, callerAuthority: line.guestIntake ? "guest_intake" : null, expiresAt: new Date(Date.now() + (line.guestIntake ? 600_000 : 120_000))}).onConflictDoNothing().returning();
return admission;
});
if (!existing || existing.requestFingerprint !== fingerprint || existing.toolTokenHash !== voiceTokenHash(capability) || !["awaiting_approval", "guest_intake"].includes(existing.state) || existing.expiresAt.getTime() <= Date.now()) throw forbidden("Incoming call is no longer pending");
@@ -155,7 +158,10 @@ export function voiceInboundService(db: Db, options: {
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()}});
}
if (admission) await tx.update(chatVoiceInboundCalls).set({state: "ended", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, admission.id));
if (admission) {
await notifyVoiceInbound(tx, { kind: "admission", id: admission.id });
await tx.update(chatVoiceInboundCalls).set({state: "ended", cleanupConfirmedAt: new Date(), cleanupNextAttemptAt: null, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, admission.id));
}
await tx.insert(activityLog).values({companyId: row.companyId, actorType: "system", actorId: "speko-lifecycle", action: "voice.call.ended", entityType: session ? "voice_session" : "voice_inbound_call", entityId: session?.id ?? admission!.id, details: {endpointId: row.id}});
}
return true;
@@ -174,7 +180,7 @@ export function voiceInboundService(db: Db, options: {
const [call] = await tx.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, row.companyId), eq(chatVoiceInboundCalls.endpointId, row.id), eq(chatVoiceInboundCalls.providerSessionId, envelope.session_id))).for("update");
const hash = voiceTokenHash(capability);
if (!call || !timingSafeEqual(Buffer.from(hash), Buffer.from(call.toolTokenHash)) || context.generation !== call.generation || context.fingerprint !== call.credentialFingerprint) throw forbidden("Invalid incoming call credential");
if (call.expiresAt.getTime() <= Date.now() && ["guest_intake", "awaiting_approval", "approving"].includes(call.state)) { await tx.update(chatVoiceInboundCalls).set({state: "expired", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); return {authorization: "expired", status: "unavailable", updates: [], cursor: 0}; }
if (call.expiresAt.getTime() <= Date.now() && ["guest_intake", "awaiting_approval", "approving"].includes(call.state)) { await notifyVoiceInbound(tx, { kind: "admission", id: call.id }); await tx.update(chatVoiceInboundCalls).set({state: "expired", cleanupNextAttemptAt: null, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); return {authorization: "expired", status: "unavailable", updates: [], cursor: 0}; }
if (call.state === "guest_intake") {
const [line] = await tx.select().from(chatVoicePhoneLines).where(and(eq(chatVoicePhoneLines.companyId, row.companyId), eq(chatVoicePhoneLines.endpointId, row.id), eq(chatVoicePhoneLines.enabled, true), eq(chatVoicePhoneLines.guestIntake, true))).for("share");
if (!line) throw forbidden("Guest intake is no longer enabled");
@@ -230,6 +236,7 @@ export function voiceInboundService(db: Db, options: {
if (!["guest_intake", "awaiting_approval", "approving"].includes(row.state)) throw conflict("This incoming call is no longer waiting for approval");
if (row.state === "approving" && (!input.approve || row.requestedIssueId !== (input.issueId ?? row.intakeIssueId ?? null))) throw conflict("This call is already being approved");
if (input.issueId) await store.authorizeTask(tx, companyId, input.issueId, caller);
await notifyVoiceInbound(tx, { kind: "admission", id: row.id });
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) {
@@ -239,12 +246,23 @@ export function voiceInboundService(db: Db, options: {
return changed;
});
if (admission.state === "approved") return projection(admission);
if (!input.approve) {
// The committed admission signal owns cleanup. Return the saved denial
// without waiting for provider account verification or physical hangup.
// Bound calls have already been handed to live-session recovery.
return projection(admission);
}
const row = await endpoint(companyId, endpointId), credentials = await options.credentials(row, admission.credentialFingerprint);
const client = provider(credentials.apiKey);
if (!input.approve) { await client.endSession(admission.providerSessionId).catch(() => undefined); return projection(admission); }
if ((await client.inspectSession(admission.providerSessionId)).endedAt) { await db.update(chatVoiceInboundCalls).set({state: "ended", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, callId)); throw conflict("The caller already hung up"); }
if ((await client.inspectSession(admission.providerSessionId)).endedAt) {
await db.transaction(async tx => {
await notifyVoiceInbound(tx, { kind: "admission", id: callId });
await tx.update(chatVoiceInboundCalls).set({state: "ended", cleanupConfirmedAt: new Date(), cleanupNextAttemptAt: null, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, callId));
});
throw conflict("The caller already hung up");
}
const result = await options.start({companyId, endpointId, issueId: admission.requestedIssueId ?? undefined, caller, idempotencyKey: `inbound:${admission.id}`, maxDurationSeconds: 600, mode: "inbound_phone", providerSessionId: admission.providerSessionId, toolToken: token(row, admission.providerSessionId, admission.generation, credentials.signingSecret)});
[admission] = await db.update(chatVoiceInboundCalls).set({state: "approved", sessionId: result.session.id, updatedAt: new Date()}).where(and(eq(chatVoiceInboundCalls.id, callId), eq(chatVoiceInboundCalls.state, "approving"))).returning();
[admission] = await db.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.id, callId), eq(chatVoiceInboundCalls.state, "approved"), eq(chatVoiceInboundCalls.sessionId, result.session.id)));
if (!admission) throw conflict("This call ended during approval");
return projection(admission);
}
@@ -254,17 +272,26 @@ export function voiceInboundService(db: Db, options: {
try { await db.transaction(tx => store.authorizeSession(tx, session.companyId, sessionId)); return true; }
catch (error) { if (error instanceof Error && "status" in error && [403, 404].includes(Number(error.status))) return false; throw error; }
}
async function reconcile(limit: number) {
await db.delete(chatVoiceInboundCalls).where(and(isNull(chatVoiceInboundCalls.sessionId), isNull(chatVoiceInboundCalls.intakeIssueId), inArray(chatVoiceInboundCalls.state, ["denied", "expired", "ended"]), lte(chatVoiceInboundCalls.expiresAt, new Date(Date.now() - 7 * 86_400_000)), lte(chatVoiceInboundCalls.updatedAt, new Date(Date.now() - 7 * 86_400_000))));
const rows = await db.select().from(chatVoiceInboundCalls).where(and(inArray(chatVoiceInboundCalls.state, ["guest_intake", "awaiting_approval", "approving", "denied", "expired"]), lte(chatVoiceInboundCalls.expiresAt, new Date()))).orderBy(asc(chatVoiceInboundCalls.updatedAt)).limit(limit);
for (const call of rows) {
try {
const row = await endpoint(call.companyId, call.endpointId), credentials = await options.credentials(row);
const client = provider(credentials.apiKey), state = await client.inspectSession(call.providerSessionId);
if (state.endedAt) await db.update(chatVoiceInboundCalls).set({state: "ended", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id));
else { await db.update(chatVoiceInboundCalls).set({state: "expired", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); await client.endSession(call.providerSessionId); }
} catch { await db.update(chatVoiceInboundCalls).set({updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); }
}
const cleanup = voiceInboundCleanup(db, async call => {
const row = await endpoint(call.companyId, call.endpointId);
// New admissions bind the signed provider account, allowing key rotation
// within that account. Older rows require the original credential version.
const credentials = await options.credentials(row, call.providerAccountId ? undefined : call.credentialFingerprint);
const client = provider(credentials.apiKey);
if (call.providerAccountId && (await client.verifyAgent(credentials.agentId)).organizationId !== call.providerAccountId)
throw conflict("Incoming call belongs to a different voice provider account");
if ((await client.inspectSession(call.providerSessionId)).endedAt) return true;
return (await client.endSession(call.providerSessionId)).confirmed;
});
async function pruneHistory() {
// Retention cannot erase an unconfirmed physical cleanup obligation.
await db.delete(chatVoiceInboundCalls).where(and(isNull(chatVoiceInboundCalls.sessionId), isNull(chatVoiceInboundCalls.intakeIssueId), isNotNull(chatVoiceInboundCalls.cleanupConfirmedAt), inArray(chatVoiceInboundCalls.state, ["denied", "expired", "ended"]), lte(chatVoiceInboundCalls.expiresAt, new Date(Date.now() - 7 * 86_400_000)), lte(chatVoiceInboundCalls.updatedAt, new Date(Date.now() - 7 * 86_400_000))));
}
return {configuration, saveConfiguration, lifecycle, pendingTool, list, decide, reconcile, authorizeGuest, history};
/** Explicit one-shot compatibility path. Recurring maintenance only owns retention. */
async function reconcile(limit: number) {
const rows = (await cleanup.load()).filter(row => row.dueAt <= Date.now()).sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)).slice(0, limit);
for (const row of rows) await cleanup.check(row.id).catch(error => logger.warn({ err: error, callId: row.id }, "Incoming call cleanup remains pending"));
await pruneHistory();
}
return {configuration, saveConfiguration, lifecycle, pendingTool, list, decide, reconcile, pruneHistory, registerCleanupWork: cleanup.register, authorizeGuest, history};
}
@@ -1,3 +1,4 @@
import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
import { notifyVoiceReplies } from "./voice-reply-notifications.js";
import type { Db } from "@paperclipai/db";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
@@ -18,6 +19,7 @@ export function notifyVoiceRecovery(tx: object, scope?: VoiceRecoveryScope) {
export async function notifyVoiceChange(tx: object, scope?: VoiceRecoveryScope) {
await notifyVoiceRecovery(tx, scope);
await notifyVoiceReplies(tx, scope);
await notifyVoiceInbound(tx, scope);
}
export function withVoiceSessionChange<T>(db: Db, id: string, write: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => PromiseLike<T>, options: { replyDelivery?: boolean } = {}): Promise<T> {
return db.transaction(async tx => {
@@ -1,3 +1,4 @@
import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
import { notifyVoiceReplies } from "./voice-reply-notifications.js";
import { recoverVoiceReplyDispatches } from "./voice-reply-dispatch-recovery.js";
import { notifyVoiceRecovery } from "./voice-recovery-notifications.js";
@@ -83,6 +84,7 @@ export async function pushVoiceReplies(db: Db, options: {
await notifyVoiceReplies(tx, { kind: "session", id: candidate.id });
await tx.update(chatActions).set({status, result: {errorCode: providerError?.code ?? "voice_authority_unavailable", ...(status === "retry" ? {retryAt: Date.now() + 10_000} : {})}, updatedAt: new Date()}).where(eq(chatActions.id, claim.action.id));
if (status === "call_ended" || status === "unknown") await notifyVoiceRecovery(tx, { kind: "session", id: candidate.id });
if (status === "call_ended") await notifyVoiceInbound(tx, { kind: "session", id: candidate.id });
if (status === "call_ended") await 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));
});
@@ -230,7 +230,7 @@ export function voiceSessionService(db: Db, options: {
return registerVoiceReplyWork(coordinator, replies, canRun);
}
async function reconcileMaintenance(limit = 25) {
await inbound.reconcile(limit);
await inbound.pruneHistory();
await reconcileReports(limit);
}
/** Explicit one-shot recovery for diagnostics/tests; never register on a recurring lane. */
@@ -238,7 +238,8 @@ export function voiceSessionService(db: Db, options: {
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 pushReplies(limit);
await reconcileMaintenance(limit);
await inbound.reconcile(limit);
await reconcileReports(limit);
return due.length;
}
async function pushReplies(limit = 25, companyId?: string, endpointId?: string) {
@@ -1,3 +1,4 @@
import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
import { notifyVoiceReplies } from "./voice-reply-notifications.js";
import { notifyVoiceChange } from "./voice-recovery-notifications.js";
import { notifyChatDeliveryWork } from "../chat-work-notifications.js";
@@ -166,19 +167,42 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`voice:${input.companyId}:${input.caller.id}:${input.idempotencyKey}`}, 0))`);
await authorizeCaller(tx, input.companyId, input.caller);
const previous = await tx.select().from(chatVoiceSessions).where(and(eq(chatVoiceSessions.companyId, input.companyId), eq(chatVoiceSessions.callerId, input.caller.id), eq(chatVoiceSessions.idempotencyKey, input.idempotencyKey))).then((rows) => rows[0]);
if (previous && (previous.requestFingerprint !== input.requestFingerprint || previous.endpointId !== input.endpointId)) throw conflict("Voice request key was already used for different work");
const context = await endpointContext(tx, input.companyId, input.endpointId);
await authorizeConnection(tx, context, input.caller);
let inboundAdmission: typeof chatVoiceInboundCalls.$inferSelect | undefined;
async function bindInboundSession(sessionId: string) {
if (!inboundAdmission || inboundAdmission.state === "approved" && inboundAdmission.sessionId === sessionId) return;
// Admission expiry must relinquish cleanup in the same commit that
// establishes an authorized owner for the existing provider call.
await notifyVoiceInbound(tx, { kind: "admission", id: inboundAdmission.id });
await tx.update(chatVoiceInboundCalls).set({ state: "approved", sessionId, updatedAt: new Date() })
.where(eq(chatVoiceInboundCalls.id, inboundAdmission.id));
}
if (input.mode === "inbound_phone") {
const [admission] = await tx.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, input.companyId), eq(chatVoiceInboundCalls.endpointId, input.endpointId), eq(chatVoiceInboundCalls.providerSessionId, input.providerSessionId ?? ""))).for("update");
const approvedRetry = previous && admission?.state === "approved" && admission.sessionId === previous.id;
if (!admission || admission.state !== "approving" && !approvedRetry || admission.approvedByUserId !== input.caller.id || admission.callerAuthority !== input.caller.authority || admission.expiresAt.getTime() <= Date.now() || admission.generation !== context.generation || admission.credentialFingerprint !== context.fingerprint || admission.toolTokenHash !== voiceTokenHash(input.toolToken ?? "") || input.idempotencyKey !== `inbound:${admission.id}`) throw forbidden("This live call has not been approved for private work");
inboundAdmission = admission;
}
if (previous) {
if (previous.requestFingerprint !== input.requestFingerprint || previous.endpointId !== input.endpointId) throw conflict("Voice request key was already used for different work");
await authorizeConnection(tx, await endpointContext(tx, input.companyId, input.endpointId), input.caller);
if (inboundAdmission) {
// A legacy approval may have committed the session before binding the
// admission. Lock admission before session, as inbound cleanup does.
const { session } = await authorizeSession(tx, input.companyId, previous.id);
if (session.mode !== "inbound_phone" || session.providerSessionId !== input.providerSessionId
|| session.callerId !== input.caller.id || session.callerAuthority !== input.caller.authority
|| session.approvedByUserId !== input.caller.id || session.toolTokenHash !== inboundAdmission.toolTokenHash
|| inboundAdmission.sessionId !== null && inboundAdmission.sessionId !== session.id
|| inboundAdmission.requestedIssueId !== null && inboundAdmission.requestedIssueId !== session.issueId)
throw forbidden("This live call has not been approved for private work");
await bindInboundSession(session.id);
return { session, created: false as const };
}
await authorizeTask(tx, input.companyId, previous.issueId, input.caller);
return { session: previous, created: false as const };
}
const context = await endpointContext(tx, input.companyId, input.endpointId);
await authorizeConnection(tx, context, input.caller);
if (context.generation < 1 || input.maxDurationSeconds < 30 || input.maxDurationSeconds > 1800 || !Number.isInteger(input.maxDurationSeconds)) throw conflict("Voice setup or duration is invalid");
if (input.mode === "inbound_phone") {
const [admission] = await tx.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, input.companyId), eq(chatVoiceInboundCalls.endpointId, input.endpointId), eq(chatVoiceInboundCalls.providerSessionId, input.providerSessionId ?? ""))).for("update");
if (!admission || admission.state !== "approving" || admission.approvedByUserId !== input.caller.id || admission.callerAuthority !== input.caller.authority || admission.expiresAt.getTime() <= Date.now() || admission.generation !== context.generation || admission.credentialFingerprint !== context.fingerprint || admission.toolTokenHash !== voiceTokenHash(input.toolToken ?? "") || input.idempotencyKey !== `inbound:${admission.id}`) throw forbidden("This live call has not been approved for private work");
}
const binding = await input.prepareBinding(tx);
const task = await authorizeTask(tx, input.companyId, binding.issueId, input.caller);
const conversation = await tx.select().from(chatConversations).where(and(eq(chatConversations.id, binding.conversationId), eq(chatConversations.companyId, input.companyId), eq(chatConversations.endpointId, input.endpointId), eq(chatConversations.issueId, binding.issueId))).then((rows) => rows[0]);
@@ -197,6 +221,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
await notifyVoiceChange(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}});
await bindInboundSession(promoted.id);
return {session: promoted, created: false as const};
}
}
@@ -207,6 +232,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
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" } });
await bindInboundSession(session.id);
return { session, created: true as const, token };
});
}