diff --git a/doc/idle-voice-inbound-retention.md b/doc/idle-voice-inbound-retention.md new file mode 100644 index 00000000000..c5abfe696a5 --- /dev/null +++ b/doc/idle-voice-inbound-retention.md @@ -0,0 +1,47 @@ +# Inbound voice history retention + +Inbound history retention uses the shared work scheduler. It does not scan on +the chat maintenance tick. Call-report reconciliation remains a separate duty. + +An admission is eligible only when it is denied, expired or ended, provider +cleanup is confirmed, and neither a session nor an intake task owns it. The +existing policy retains it until both expiry and last update are seven days old: +`max(expiresAt, updatedAt) + 7 days`. No new table or persisted state is added. + +The voice owner reconstructs eligible deadlines on startup, yielding after at +most 100 selected rows per scope refresh. Re-registering the same service resets +its page cursor so a partial scan cannot skip old deadlines. The existing +`notifyVoiceInbound` boundary now notifies cleanup and retention before writes. +Outer transaction settlement wakes the workers. Rolled-back changes do not +create retention work. Signed callbacks, rejection cleanup, approval/session +handoff and cleanup recovery retain their existing transaction ownership. +Configuration, authority and session-only changes do not change admission +retention eligibility, so they do not wake retention. Startup reconstructs +existing history, and admission writes carry the exact retention hint. + +At the deadline, retention locks the admission and rechecks current eligibility +and both timestamps before deleting. PostgreSQL submillisecond timestamps round +up for scheduling. Cached candidates cannot authorize deleting a new binding or +history whose retention deadline moved. Batches page through history and use the +shared worker's company fairness, coalescing, bounded backoff and drain behavior. +Deletion failure retains a retry; a paused first write remains pending for resume. +Shutdown cancels or joins work, and restart reloads remaining durable deadlines. + +Configured idle endpoints, ineligible history, and future retained history do not +cause recurring SQL between changes or actual deadlines. Unconfirmed cleanup +remains owned by the cleanup worker. Session/task history is not deleted here. +Explicit one-shot `pruneHistory` and diagnostic reconciliation remain available; +startup does not register those bulk operations as periodic work. + +Supported writers must participate in the shared transaction notification +boundary. Independent SQL writers or additional processes require their own +notification/recovery contract. This migration does not extend Cloud alarms or +sleep eligibility: an in-process deadline cannot wake stopped compute. The +external wake owner must cover retention before that deadline can permit sleep. + + +The [current-parent acceptance](plans/2026-10-25-voice-inbound-retention-sleep-audit.md#current-parent-paging-and-restart-acceptance-october-28) +records bounded paging and restart acceptance. The subsequent +[notification review repair](plans/2026-10-25-voice-inbound-retention-sleep-audit.md#configuration-rescan-review-repair-october-28) +records 282 current passing cases and repeated configuration saves selecting zero +history rows, with the original failures and verification limits preserved. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 27ad9bb5341..a70e2892888 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1589,6 +1589,34 @@ 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. + +Inbound cleanup is [#17580](https://github.com/paperclipai/paperclip/pull/17580), +head 4b4bf2d8af6e08d3859ada4b720e7fbd3ea27e24, stacked on #17578. It is +mergeable; CI 40181994295 and review 40181994134 are live with no initial failure. +The final migration applies/replays and its backfill is tested twice. Slice 24 +starts the separate inbound-history retention audit. Its deadline preserves both +seven-day age conditions and requires confirmed cleanup plus no task/session +binding. Call reports follow separately. No original remaining scope is removed. + +## Slice 24: inbound voice history retention + +Branch `codex/sleep-24-voice-inbound-retention` follows inbound cleanup #17580 +including its migration runner repair `4f0364d269`. It removes only periodic +history pruning. The existing admission row owns the seven-day deadline and +confirmation/binding conditions. The existing notification helper wakes retention +on outer transaction settlement; no new table or caller wrapper is added. + +218 selected tests pass: 131 full voice/app wiring, 24 retention/cleanup lifecycle, +52 deletion/startup, and 11 parent recovery. Workspace typecheck/build and final +server typecheck pass. Tests prove real signed-callback and provider-cleanup +reactivation, pre-deadline quiet, exact deadlines, concurrent binding protection, +restart, drain, rollback, failure and shutdown. Initial fixture failures remain +recorded in the slice audit. Remote checks and review are pending publication. + +Next separate owner: voice call-report reconciliation. Preserve the existing +24-hour post-call refresh obligation, transcript ordering, task projection and +provider diagnostics while removing scans with no eligible calls. Inspect +completion semantics before treating a report marked available as fully settled. ## Acceptance follow-up after the plugin parent merged (October 27) Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at @@ -1703,3 +1731,26 @@ entries, and commits historical backfill in batches of at most 1,000. The 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. + + +## Inbound history retention current-parent acceptance, October 28 + +Retention #17583 includes cleanup parent `b8b12199b9` and preserves its generated +0395 migration. The existing shared page contract now yields after 100 history +rows per scope refresh and resets on same-owner restart. All 280 distinct cases, +server types/build and module boundaries pass. The [retention audit](2026-10-25-voice-inbound-retention-sleep-audit.md#current-parent-paging-and-restart-acceptance-october-28) +records the 13-path scope, real 105-row reconstruction, negative probe, setup and +merge failures, and limits. Latest-head review, CI and parent landing remain +required. Call-report reconciliation follows as a separate duty. + + +## Retention notification review follow-up, October 28 + +The current retention review found repeated configuration-driven history scans. +Five real connection saves selected 1,025 rows across 205 retained admissions. +Retention now wakes only for admission writes and startup; cleanup retains all +of its broader scopes. The same writes select zero history rows and preserve the +saved deadline. Confirmed deletes also avoid a redundant reload. The +[retention review ledger](2026-10-25-voice-inbound-retention-sleep-audit.md#configuration-rescan-review-repair-october-28) +records 282 passing cases, types/build, failures, writer coverage and limits. +Current-head review/CI and the earlier stack remain required. diff --git a/doc/plans/2026-10-25-voice-drain-repair.md b/doc/plans/2026-10-25-voice-drain-repair.md index e7f761ae35b..da0ae09914d 100644 --- a/doc/plans/2026-10-25-voice-drain-repair.md +++ b/doc/plans/2026-10-25-voice-drain-repair.md @@ -50,3 +50,13 @@ then verifies the saved reply sends on resume without another producer. All later shared-worker consumers inherit this repair through the stack. Their combined-tip checks and PR checks still need fresh verification after propagation. + +## Retention resume assertion + +The combined-tip consumer run exposed a retention test that called flush after +resume and assumed a zero activity counter meant the retained retry had run. Its +helper checks active work only; zero is now also the correct paused state. The +owner test now advances the actual retained nextWakeAt, waits for row removal, +and checks disarm. Before resume it also proves a stable idle snapshot through a +minute of quiet. All 11 retention cases pass in /tmp/sleep24-drain-tests.log. +No production scheduling deadline or dispatch behavior changed for this repair. diff --git a/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md b/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md new file mode 100644 index 00000000000..c92ea90b097 --- /dev/null +++ b/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md @@ -0,0 +1,339 @@ +# Inbound voice history retention sleep audit + +Original baseline: inbound cleanup PR #17580, head +4b4bf2d8af6e08d3859ada4b720e7fbd3ea27e24. +Branch: codex/sleep-24-voice-inbound-retention. +Current integrated parent: b8b12199b9185e6318319cceb9511f5285734e41. +Status: implementation and local acceptance complete; current-head CI/review and parent landing remain required. + +## One reason + +Remove the remaining recurring inbound history delete. Keep call reports in the +next separate slice. `voiceInboundService.pruneHistory` currently runs from +`voiceSessionService.reconcileMaintenance` in chat's periodic maintenance lane. + +The existing retention policy deletes only denied, expired or ended admissions +with no session or intake task, confirmed provider cleanup, and both expiresAt +and updatedAt at least seven days old. Preserve that policy. The actual deletion +deadline is max(expiresAt, updatedAt) + seven days. Never treat unconfirmed cleanup +or a terminal UI label as deletable history. + +## Owners and producer boundary + +The existing admission rows and cleanup confirmation remain truth. No new table +or row fields are needed. Reuse the scoped deadline worker and register a separate +retention queue. Keep state selection, deadline calculation and deletion in the +voice owner. Startup/app wiring only registers the owner. + +All supported admission mutations already use notifyVoiceInbound directly or the +voice-change composition. Extend that boundary to signal the separate retention +queue before writes. This includes signed terminal callbacks, cleanup confirmation, +legacy binding repair, approval/session handoff, expiry and rejection. Bound rows +remain ineligible; their history belongs to their other owners. Company deletion +can remove confirmed unbound history within its existing transaction. + +At deletion time, lock and recheck the row and deadline. A cached candidate must +not delete a row that acquired a binding, lost confirmation, or had its updatedAt +moved by a committed write. Preserve fair paging, coalescing, backoff, drain and +shutdown through the existing shared mechanism. Own deletes should retire their +cached entries without starting another periodic scan. + +## Required verification + +- Configured idle and no eligible history become SQL-quiet. +- Confirmed retained history schedules the exact seven-day deadline and does not + query before it; restart reconstructs that same deadline. +- Real cleanup confirmation and signed callbacks register retention through their + transaction boundary; rollback/outer commit behave correctly. +- Unconfirmed cleanup and bound history are never deleted. +- A changed deadline/binding between discovery and deletion is rechecked. +- Paged backlog, first write during drain, failures, and shutdown retain work. +- Existing voice, cleanup, deletion and startup consumers pass. +- Remove only history pruning from recurring maintenance; call reports remain. + +No Cloud wake or sleep eligibility expansion is included. The external scheduler +must cover a deadline before that deadline can justify stopping the host. + +## Stack state + +#17580 is published and attached; 205 selected tests, workspace types/build and +migration checks passed locally. Initial CI run 40181994295 and review run +40181994134 are live on its exact published head. It is mergeable and had no +failures at inspection. Fresh checks/reviews remain required; no merge authority +is implied. #17578 at 90caaf3fa9 has passing checks and exact-head Greptile 5/5; +its latest Paperclip reviewer stopped before completion. + +The full original recurring-server ledger, older #17490 E2E failure and final +stack/master reconciliation remain required. + +## Implementation and initial evidence + +The dedicated voice retention owner now loads eligible deadlines, rounds SQL +submillisecond dates up, and locks/rechecks before deletion. The existing producer +helper signals both voice admission cleanup and retention. App wiring registers +the owner; voice maintenance no longer calls `pruneHistory`. Reports remain in +that lane, and manual diagnostic reconciliation remains compatible. + +The initial ten PostgreSQL lifecycle cases had one invalid assertion: an open +transaction was expected to have no coordinator deadline. Intent reconciliation +correctly retains a deadline until settlement. The test now checks invisible +uncommitted state and the row still present within its owning transaction; it +requires deletion only after commit and quiet after rollback. All ten then passed. +An additional real concurrent binding-lock regression and production service +confirmation tests are being verified. Full voice acceptance uses a fresh database +created through the repaired application migrator, not direct SQL replay. + +Parent #17580 required repair 4f0364d269: register migration 0388 as top-level +because of its concurrent index. A new migration-runner regression reproduced the +failure, then proved fresh setup, invalid-index repair and concurrent/repeat +replay after the fix. This parent repair was merged into this branch; its fresh +CI is running. See the parent's audit for preserved failure evidence. + +## Acceptance evidence + +- 11 retention lifecycle tests and 13 parent cleanup tests pass (24 total). + The added concurrent-binding test holds the real row lock, commits a task + binding while deletion waits, and proves retention rechecks and keeps the row. +- Full voice integration and app wiring pass 131 cases on a fresh database. + Two new service tests start retention while the admission is unconfirmed, then + use a signed terminal callback or real rejection/provider cleanup to establish + confirmation. Each observes its saved seven-day deadline, SQL quiet before it, + deletion at it, and quiet afterwards. +- The first acceptance run waited 120 seconds before rejection, exceeding the + original call's approval window. That correctly rejected the test operation and + left an unconfirmed fixture affecting two later global cleanup assertions. The + corrected test checks pre-confirmation quiet for ten seconds and the full suite + passes on a newly migrated database. Production expiry policy did not change. +- Workspace typecheck and build pass. A final server typecheck covers the last + added concurrent-binding test. Shared deletion/startup and parent recovery + checks are tracked below when their current processes finish. +- No repository-wide `pnpm test:run` success is claimed: the local embedded + PostgreSQL initialization limitation from earlier slices remains. The selected + database suites use isolated real PostgreSQL; fresh remote CI is required. + +Logs: `/tmp/sleep24-retention-cleanup-final.log`, +`/tmp/sleep24-voice-acceptance.log` (preserved initial failure), +`/tmp/sleep24-voice-acceptance-fixed.log`, `/tmp/sleep24-workspace-types.log`, +`/tmp/sleep24-workspace-build.log`, `/tmp/sleep24-final-server-types.log`. + +Final local consumer checks pass: 52 company-deletion/startup tests, 11 parent +voice recovery tests, and final server typecheck. Selected nonoverlapping test +coverage totals 218 passing cases. The implementation and local verification are +complete; remote PR CI/review and the rest of the original stack remain pending. + +## Review repairs + +The first published head 9c38ddfe53 passed all CI and Greptile 5/5. Paperclip +Review found unnecessary endpoint-wide retention refresh on every new call and +missing behavioral coverage of removal from the maintenance lane. + +Pre-call admission now allocates its ID before insertion and registers the +existing intent for that admission only. Retention lookup uses the primary key; +new calls no longer scan the endpoint's retained history. Cleanup receives the +same precise hint. Existing endpoint/company configuration changes retain their +scope, and startup still reconstructs eligible history. No schema change. + +Two service regressions verify the actual signed pre-call scope and drive the +real chat coordinator/voice maintenance path with due, confirmed history. The row +survives maintenance and is deleted by the retention worker. Temporarily restoring +pruning to voice maintenance makes this latter test fail with the row missing; +restoring the implementation passes. All 133 full voice/app wiring tests and +server typecheck pass after the repair. Logs are /tmp/sleep24-review-acceptance.log, +/tmp/sleep24-review-types.log, /tmp/sleep24-maintenance-mutation.log and +/tmp/sleep24-maintenance-restored.log. Typecheck covers production sources; the +server tsconfig excludes the test directory, whose behavior is verified by Vitest. +Fresh remote checks/review are required on the repaired head. + +Parent #17580 has all checks terminal and Greptile 5/5 at 4f0364d269. Its final +Paperclip review added two findings: unresolved cleanup can block deletion, and +rejection awaits provider calls. These require explicit disposition/repair at the +parent; green checks do not resolve them. An arbitrary retry cap or credential +failure is not proof that a provider call ended and must not silently discard the +cleanup obligation. Preserve that requirement while assessing the deletion path. + +## Denial worker propagation + +The parent denial-latency fix is incorporated. Retention's real cleanup-receipt +producer case now registers the cleanup worker and waits for its confirmed result, +instead of assuming the decision request itself performs provider I/O. Both +retention and denial regression cases are retained. The initial merge verification +failed to parse unresolved conflict markers; no voice cases ran. After preserving +both sides and completing the merge, all 134 voice/app cases passed on that still +unused voice database. Logs: `/tmp/sleep24-denial-acceptance.log` (parse failure), +`/tmp/sleep24-denial-acceptance-fixed.log` (134 pass). + + +## Current-parent paging and restart acceptance, October 28 + +This revision normally merges cleanup parent `b8b12199b9185e6318319cceb9511f5285734e41`. +It retains that parent's generated migration 0395, cleanup paging and approval +repairs. The 13 original owned paths remain the complete slice. No database, +shared worker, workflow, lockfile or provider protocol change belongs to retention. +The plan merge preserved both dated histories. The automatic source merge added +a duplicate DELIVERY_QUEUES test import; the first combined voice suite failed +to parse before running any case. Removing only the duplicate import lets all +126 full voice cases run and pass on the still-empty dedicated test database. +That parse failure remains `/tmp/sleep24-final-voice-consumers.log`. + +### Bound reconstruction without changing retention policy + +The old retention reader performed all 100-row queries inside one load. A new +105-row future-history fixture showed it selected 105 rows in startup's first +pass, failing the 100-row budget. The worker now uses the parent's existing +`{ rows, complete }` contract. It returns at most 100 rows per scope refresh and +preserves the full explicit `load` API. It changes no seven-day calculation, +submillisecond rounding, confirmation requirement or binding exclusion. + +Per-scope cursors advance only after successful reads and reset on registration. +The new case stops after the first 100 of 105 future records, starts the same +service owner, checks another initial 100 rows, reconstructs the original deadline, +and deletes all 105 rows at that deadline. It observes SQL quiet before the +actual deadline and after settlement. Removing just the reset makes restart read +five rows and fails the budget assertion. The guarded mutation was restored +before the complete lifecycle acceptance. Row measurements are selected rows, +not PostgreSQL tuple visits or deployed costs. + +The external retention fixture also assumed a pre-existing schema. Its first +empty-database run failed on missing companies/admission tables. It now applies +the application migrations with the standard setup timeout. The first-pass +budget then failed at the intended 105-versus-100 assertion before the reader +repair. Both failures are retained in `/tmp/sleep24-empty-fixture-red.log` and +`/tmp/sleep24-page-budget-red.log`. + +### Final evidence and limits + +280 distinct cases pass on the combined source: + +- All 126 voice-session integration cases, including actual signed ingress, + cleanup confirmation, scoped retention wake and maintenance-lane behavior. +- All 154 cases in ten full retention, cleanup, parent recovery, company removal, + shared deadline/coordinator/batching, startup and hostname suites. This includes + 12 retention and 14 cleanup lifecycle cases. Voice integration, parent recovery + and company removal use separate initially empty databases and migrate themselves. + Retention uses its dedicated database after the fresh-setup repair. Cleanup + uses its already-migrated dedicated database. These suites do not share fixture + state with a live application. + +Server typecheck, build and the module-boundary check pass. Logs: +`/tmp/sleep24-final-voice-consumers-fixed.log`, +`/tmp/sleep24-final-lifecycle-consumers.log`, +`/tmp/sleep24-final-current-types.log`, `/tmp/sleep24-final-current-build.log`, and +`/tmp/sleep24-final-module-boundaries.log`. The passing initial 12-case retention +run is `/tmp/sleep24-bounded-retention-first-green.log`; it is included in the +154, not counted twice. The guarded restart failure is +`/tmp/sleep24-cursor-reset-mutant-red.log`. + +Older 218/133/134-case and workspace-wide results apply to their stated heads. +No new complete local repository suite, live provider, Cloud wake, host/database +suspension or savings claim is made. Independent writers still need an explicit +wake integration. Call reports remain their own duty. Current-head reviews, +all CI and earlier stack landing remain required; an older passing score or +inherited local acceptance cannot replace these gates. + + +The prior published head `daca270a99` had a failed server (4/12) check in run +40557790710, job 122386307340. Its actual log records a runner shutdown signal +and operation cancellation before the suite completed. This is incomplete +acceptance, not a passing run or an established product flake. The trusted runner +selection also had a cancelled check. The source failure log is preserved at +`/tmp/sleep24-daca-server4-failed.log`. New-head CI must complete independently. + + +## Configuration rescan review repair, October 28 + +Ponytail's current-head 4/5 review of `ffe9d47b7f` identified that completed +connection/company/endpoint scans restarted after every configuration save. +Paging bounded each burst but did not remove total repeated work. A real +producer regression seeds 205 future confirmed admissions and performs five +separate `toolAccessService.updateConnection` saves, allowing each scan to settle. +It selected 1,025 admission rows before the repair and selects zero afterward. +All original retention deadlines remain armed and unchanged. + +`notifyVoiceInbound` still notifies cleanup for every supported scope. Retention +now receives only admission-scoped writes. Startup retains its complete paged +reconstruction. Configuration, authority and voice-session changes do not modify +admission eligibility or its timestamps. Admission creation, confirmation, +expiry, approval/binding and signed lifecycle writes retain their exact hints. +The schema uses restrictive session/task foreign keys, not implicit SET NULL +transitions that could create eligible history behind this boundary. Cascading +endpoint/company removal creates no retained work. A previously cached deadline +may expire once after a row has been removed; it cannot recreate history or start +recurring reconciliation. Arbitrary SQL and other processes still require their +own explicit wake contract. + +Successful deletion now uses its returned row identity to settle the cached +entry directly. A changed, bound or not-yet-due row still reloads current state +so its deadline is preserved. The new real-delete case observes no post-delete +SELECT. Guardedly removing that fast path reproduces the extra SELECT; the +original source was restored. There is no new helper, queue, schema or scheduler +contract. + +All 282 distinct cases pass after this repair: 127 complete voice integration +cases and 155 cases in ten full retention/cleanup/recovery/company-removal/shared/ +startup suites. These include 13 retention cases. The complete voice suite uses +a fresh dedicated database; the lifecycle suites use their existing separate +isolated databases. Server typecheck and build pass. The earlier 280 count belongs +to `ffe9d47b7f`; these cases include that coverage and are not added to it. + +Logs: `/tmp/sleep24-config-amplification-red.log` (1,025 rows), +`/tmp/sleep24-review-final-voice-consumers.log` (127 pass and zero-row budget), +`/tmp/sleep24-review-final-lifecycle-consumers.log` (155 pass), +`/tmp/sleep24-review-final-types.log`, `/tmp/sleep24-review-final-build.log`, and +`/tmp/sleep24-deleted-reload-mutant-red.log`. Fresh exact-head reviews and CI remain +required after publication; the prior green CI and 4/5 review are not reused for +the new commit. No complete local repository suite, live provider, Cloud wake or +physical suspension claim is added. + +## Current-master CI repair — October 28 + +After normal integration of landed cleanup, head `768c650365362b12b75ca6481f44468132cf599a` +received exact-head Greptile 5/5 and Ponytail 5/5, but CI run `40697141057` +failed. This is not accepted as green evidence. Server shard 7/12 job +`122846941793` reported 1 failed test, 1,750 passed and 6 skipped: cleanup's +backlog fairness case expected the other company in the first 25 provider +calls even when its random admission ID placed it after the first 100-row +inventory page. The secondary afterEach error retained the deliberately +injected provider failure because the assertion interrupted settlement. + +Giving the second company a fixed final UUID reproduced the same first-batch +assertion on unchanged production scheduling. The fixture now uses its supplied +ID for both the inserted row and transaction notification. The corrected case +controls fake time, places that company on the second inventory page, checks it +is absent before discovery and served within the next 25-call batch, completes +all 105 successful calls before the failed provider's saved retry, and checks +that failed provider runs exactly once again when its deadline arrives. It also +checks idle SQL after settlement. No production worker, policy, deadline, +assertion timeout, or workflow changes are made. + +The first controlled-time repair run passed 23 cases but failed the fairness +case by demanding idle active=0 between inventory passes. An unresolved +transaction intent can legitimately retain that hold while the next pass is +scheduled. The final test waits for the next scheduled pass or actual idle +settlement while advancing one second at a time, then requires actual idle +settlement after the backlog. This failure is retained rather than counted as +a pass. A guarded mutation bypassing company rotation fails the corrected +fairness assertion with missing second-company delivery; the original shared +helper was restored byte-for-byte before final acceptance. + +The same CI run's Canary Dry Run job `122846941357` independently failed +installing the standalone Kubernetes plugin: npm returned 404 for the exact +`rollup-4.66.0.tgz` requested by Vite through Vitest. At 13:39:46 UTC both that +version metadata and exact tarball returned HTTP 200. No dependency pin, +workflow exception, or blind CI retry was introduced. New-head CI, including +the canary, must pass independently; the original 404 remains recorded. + +Final local acceptance after restoring the guarded shared helper: both complete +cleanup (24) and retention (13) lifecycle suites pass, 37 distinct cases total. +Retention starts from a new empty dedicated PostgreSQL database and applies the +application migrator; cleanup uses its isolated already-migrated database. +The earlier 282-case evidence belongs to unchanged retention source at its +stated head and is not added to 37. No new full local repository suite, live +provider, Cloud wake, suspension, or cost-saving result is claimed. + +Logs: `/tmp/sleep24-768-server7-job.log`, `/tmp/sleep24-768-canary-job.log`, +`/tmp/sleep24-cleanup-fairness-red.log`, +`/tmp/sleep24-cleanup-fairness-green.log` (the 23/24 failed intermediate run), +`/tmp/sleep24-cleanup-fairness-final.log` (24 pass), +`/tmp/sleep24-cleanup-fairness-mutant-red.log`, +`/tmp/sleep24-ci-fairness-final-consumers.log` (37 pass), and +`/tmp/sleep24-rollup-registry-probe.json` (current HTTP availability only). diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index f1fdef98fa8..8e7fbd20d38 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -87,6 +87,7 @@ describe("createChatReconciliationCoordinator", () => { 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.inbound.registerRetentionWork(deliveryWork,"); expect(source).toContain("chatChannels.voice.reconcileMaintenance()"); expect(source).not.toContain("chatChannels.voice.reconcile()"); }); diff --git a/server/src/__tests__/voice-inbound-cleanup.test.ts b/server/src/__tests__/voice-inbound-cleanup.test.ts index 6a31911a3e6..be34c5a6ea2 100644 --- a/server/src/__tests__/voice-inbound-cleanup.test.ts +++ b/server/src/__tests__/voice-inbound-cleanup.test.ts @@ -59,7 +59,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu }); afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); async function admission(patch: Partial = {}, writer = db) { - const id = randomUUID(); + const id = patch.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]; @@ -260,19 +260,33 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu 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(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + const broken = await admission({ id: "00000000-0000-4000-8000-000000000001" }); for (let i = 0; i < 104; i++) await admission(); - const other = await fixture(), last = await admission({ companyId: other.companyId, endpointId: other.endpointId }); + const other = await fixture(), last = await admission({ id: "ffffffff-ffff-4fff-bfff-ffffffffffff", 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(terminate.mock.calls).toHaveLength(25); + expect(terminate.mock.calls.some(([call]) => call.id === last.id)).toBe(false); + const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, broken.id)); + const retryAt = pending.cleanupNextAttemptAt!.getTime(); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(terminate).toHaveBeenCalledTimes(50), { interval: 0 }); + const otherPosition = terminate.mock.calls.findIndex(([call]) => call.id === last.id); + expect(otherPosition).toBeGreaterThanOrEqual(25); + expect(otherPosition).toBeLessThan(50); + for (let pass = 0; pass < 20; pass++) { + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt() !== null || idleWorkSnapshot().active === 0).toBe(true), { interval: 0 }); + } + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { interval: 0 }); + expect(terminate.mock.calls.filter(([call]) => call.id !== broken.id)).toHaveLength(105); + expect(terminate.mock.calls.filter(([call]) => call.id === broken.id)).toHaveLength(1); + expect(Date.now()).toBeLessThan(retryAt); expect(errors).toHaveLength(1); failed = false; - await settled(); errors.length = 0; await quiet(); + await vi.advanceTimersByTimeAsync(retryAt - Date.now()); await settled(); + expect(terminate.mock.calls.filter(([call]) => call.id === broken.id)).toHaveLength(2); + errors.length = 0; await quiet(); }); }); diff --git a/server/src/__tests__/voice-inbound-retention.test.ts b/server/src/__tests__/voice-inbound-retention.test.ts new file mode 100644 index 00000000000..306e43e9408 --- /dev/null +++ b/server/src/__tests__/voice-inbound-retention.test.ts @@ -0,0 +1,236 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, agents, chatConversations, chatEndpoints, 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 { voiceInboundRetention } from "../services/voice/voice-inbound-retention.js"; +import { notifyVoiceInbound } from "../services/voice/voice-inbound-notifications.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_RETENTION_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("inbound voice retention lifecycle", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string, endpointId: string, agentId: string, issueId: string, conversationId: string; + let coordinator: ReturnType | undefined; + let enabled = true; + const errors: unknown[] = [], fixtureCompanies: string[] = []; + const week = 7 * 86_400_000; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("voice-inbound-retention-"); + 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; + ({ 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 = {}, 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: "ended", cleanupConfirmedAt: new Date(), updatedAt: new Date(Date.now() - week - 1000), approvalCode: "123456", generation: 1, credentialFingerprint: "test", toolTokenHash: "a".repeat(64), requestFingerprint: id, expiresAt: new Date(Date.now() - week - 1000), ...patch }).returning())[0]; + }); + } + async function owner(call: typeof chatVoiceInboundCalls.$inferSelect, patch: Partial = {}) { + 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 = voiceInboundRetention(db)) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + await service.register(coordinator, () => enabled).ready; + return service; + } + async function flush() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 }); + } + async function quiet(ms = 120_000) { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(ms); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + } + async function row(id: string) { return (await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, id)))[0]; } + it("bounds history inventory and restores all deadlines when the same owner restarts mid-scan", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + const deadline = Date.now() + 60_000; + for (let index = 0; index < 105; index++) await admission({ expiresAt: new Date(deadline - week), updatedAt: new Date(deadline - week) }); + 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(deadline), { interval: 0 }); + expect(Math.max(...counts)).toBeLessThanOrEqual(100); reads.mockRestore(); + await quiet(1000); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, companyId))).toHaveLength(105); + await vi.advanceTimersByTimeAsync(deadline - Date.now()); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, companyId))).toHaveLength(0); + await quiet(); + } finally { reads.mockRestore(); } + }); + it("settles a confirmed deletion without reloading the deleted history", async () => { + const call = await admission(), service = voiceInboundRetention(db); + const reads = vi.spyOn(db, "select"); + try { + await expect(service.check(call.id)).resolves.toBeNull(); + expect(reads).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); } + expect(await row(call.id)).toBeUndefined(); + }); + it("leaves configured idle quiet and reacts to the first committed history", async () => { + await start(); await flush(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + const call = await admission(); await flush(); expect(await row(call.id)).toBeUndefined(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("uses the later timestamp, stays quiet before its deadline, and reconstructs it on restart", async () => { + const now = Date.now(), later = now + 400_000; + const call = await admission({ expiresAt: new Date(now - week + 200_000), updatedAt: new Date(later - week) }); + await start(); await flush(); expect(coordinator!.nextWakeAt()).toBe(later); await quiet(); + await coordinator!.stop(); await start(); await flush(); expect(coordinator!.nextWakeAt()).toBe(later); + expect(await row(call.id)).toBeDefined(); + await vi.advanceTimersByTimeAsync(later - Date.now() + 1000); await flush(); + expect(await row(call.id)).toBeUndefined(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("rounds PostgreSQL submillisecond deadlines up and never deletes early", async () => { + const call = await admission(); const dueAt = Math.trunc(Date.now()) + 60_000; + await db.execute(sql`UPDATE chat_voice_inbound_calls SET updated_at = to_timestamp(${dueAt - week}::double precision / 1000) + interval '0.0005 seconds' WHERE id = ${call.id}`); + const service = voiceInboundRetention(db), [entry] = await service.load({ kind: "admission", id: call.id }); + expect(entry.dueAt).toBe(dueAt + 1); + vi.setSystemTime(dueAt); await service.check(call.id); expect(await row(call.id)).toBeDefined(); + vi.setSystemTime(dueAt + 1); await service.check(call.id); expect(await row(call.id)).toBeUndefined(); + }); + it("never prunes unconfirmed, live, task-bound or session-bound history", async () => { + const unconfirmed = await admission({ cleanupConfirmedAt: null }); + const live = await admission({ state: "awaiting_approval" }); + const task = await admission({ intakeIssueId: issueId }); + const bound = await admission(), session = await owner(bound); + await db.update(chatVoiceInboundCalls).set({ sessionId: session.id }).where(eq(chatVoiceInboundCalls.id, bound.id)); + const service = await start(); await flush(); await service.prune(); + for (const call of [unconfirmed, live, task, bound]) expect(await row(call.id)).toBeDefined(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("rechecks current dates and bindings after stale discovery", async () => { + const service = voiceInboundRetention(db), changed = await admission(), bound = await admission(); + expect(await service.load()).toHaveLength(2); + await db.update(chatVoiceInboundCalls).set({ updatedAt: new Date() }).where(eq(chatVoiceInboundCalls.id, changed.id)); + await db.update(chatVoiceInboundCalls).set({ intakeIssueId: issueId }).where(eq(chatVoiceInboundCalls.id, bound.id)); + expect((await service.check(changed.id))!.dueAt).toBeGreaterThan(Date.now()); + expect(await service.check(bound.id)).toBeNull(); + expect(await row(changed.id)).toBeDefined(); expect(await row(bound.id)).toBeDefined(); + }); + it("waits for a concurrent binding transaction and rechecks after its commit", async () => { + const call = await admission(), service = voiceInboundRetention(db); + let release!: () => void, acquired!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { acquired = resolve; }); + const binding = db.transaction(async tx => { + await tx.update(chatVoiceInboundCalls).set({ intakeIssueId: issueId }).where(eq(chatVoiceInboundCalls.id, call.id)); + acquired(); await gate; + }); + await ready; + let finished = false; + const deleting = service.check(call.id).then(result => { finished = true; return result; }); + try { + await vi.advanceTimersByTimeAsync(100); expect(finished).toBe(false); + } finally { release(); await binding; } + expect(await deleting).toBeNull(); expect((await row(call.id)).intakeIssueId).toBe(issueId); + }); + it("defers nested writes until outer commit and discards rollback", async () => { + await start(); await flush(); let committed!: string; + await db.transaction(async tx => { + committed = (await admission({}, tx as typeof db)).id; + await vi.advanceTimersByTimeAsync(5000); + expect(await row(committed)).toBeUndefined(); + expect(await tx.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, committed))).toHaveLength(1); + }); + await flush(); expect(await row(committed)).toBeUndefined(); await quiet(); + await expect(db.transaction(async tx => { await admission({}, tx as typeof db); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await flush(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("keeps the first write during drain and runs it on resume", async () => { + await start(); await flush(); enabled = false; + const call = await admission(); await flush(); + const drained = idleWorkSnapshot(); + await quiet(60_000); expect(idleWorkSnapshot()).toEqual(drained); expect(await row(call.id)).toBeDefined(); + const resumeAt = coordinator!.nextWakeAt(); expect(resumeAt).not.toBeNull(); + enabled = true; await vi.advanceTimersByTimeAsync(Math.max(0, resumeAt! - Date.now())); + await vi.waitFor(async () => { await flush(); expect(await row(call.id)).toBeUndefined(); }); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("pages through more than one batch of history across companies", async () => { + for (let i = 0; i < 104; i++) await admission(); + const other = await fixture(); const last = await admission({ companyId: other.companyId, endpointId: other.endpointId }); + const service = await start(); + expect(await row(last.id)).toBeUndefined(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(await service.load()).toHaveLength(0); }, { timeout: 5000 }); + await flush(); await quiet(); + }); + it("retains failed deletion for a bounded retry and then becomes quiet", async () => { + const call = await admission(); + const transaction = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("database unavailable")); + await start(); transaction.mockRestore(); + expect(errors).toHaveLength(1); errors.length = 0; + expect(await row(call.id)).toBeDefined(); + await quiet(10_000); + await vi.advanceTimersByTimeAsync(31_000); await flush(); expect(await row(call.id)).toBeUndefined(); await quiet(); + }); + it("stops future deletion at shutdown and resumes its durable deadline after restart", async () => { + const call = await admission({ updatedAt: new Date(Date.now() - week + 60_000) }); + await start(); await flush(); await coordinator!.stop(); + await quiet(120_000); expect(await row(call.id)).toBeDefined(); + await start(); await flush(); expect(await row(call.id)).toBeUndefined(); await quiet(); + }); +}); diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 12b65b0873c..fc36fad8f49 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -1,3 +1,5 @@ +import { createChatReconciliationCoordinator } from "../app.js"; +import { subscribeDatabaseWork } from "@paperclipai/db"; import * as voiceReplyNotifications from "../services/voice/voice-reply-notifications.js"; import { voiceReplySchedule } from "../services/voice/voice-reply-schedule.js"; import { registerVoiceReplyWork } from "../services/voice/voice-reply-work.js"; @@ -80,6 +82,127 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication }; } + it("keeps retained history quiet when a voice connection configuration is saved repeatedly", async () => { + const f = await inboundFixture(); + const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" }); + await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers); + await db.insert(chatVoiceInboundCalls).values(Array.from({ length: 204 }, () => ({ + ...f.call, id: randomUUID(), providerSessionId: randomUUID(), state: "ended" as const, + cleanupConfirmedAt: new Date(), updatedAt: new Date(), + }))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const expected = f.call.expiresAt.getTime() + 7 * 86_400_000; + const settle = () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBe(expected); + }, { timeout: 5000 }); + let selected = 0; + const select = db.select.bind(db); + try { + await f.service.inbound.registerRetentionWork(coordinator, () => true).ready; await settle(); + const reads = vi.spyOn(db, "select").mockImplementation((...args) => { + const selection = select(...args), from = selection.from.bind(selection); + vi.spyOn(selection, "from").mockImplementation(table => { + const query = from(table); + if (table === chatVoiceInboundCalls) { + const execute = query.execute.bind(query); + vi.spyOn(query, "execute").mockImplementation(async () => { + const rows = await execute(); selected += rows.length; return rows; + }); + } + return query; + }); + return selection; + }); + try { + for (let index = 0; index < 5; index++) { + await toolAccessService(db).updateConnection(f.connectionId, { name: `Saved voice ${index}` }); + await settle(); + } + expect(selected).toBe(0); expect(onError).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); } + } finally { + await coordinator.stop(); vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, f.companyId)); + } + }); + + it("limits a new pre-call wake to its admission instead of scanning endpoint history", async () => { + const f = await inboundFixture(); const hints: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.voiceInboundRetention, + (event, scopes) => { if (event === "settled") hints.push(...scopes ?? ["*"]); }); + try { + const id = randomUUID(), incoming = f.sign({ ...f.event, session_id: id, call_id: id }); + await f.service.inbound.lifecycle(f.endpoint.publicId, incoming.body, incoming.headers); + const [saved] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.providerSessionId, id)); + expect(hints).toEqual([`admission:${saved.id}`]); + } finally { + unsubscribe(); await db.delete(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, f.companyId)); + } + }); + + it("leaves due history intact during chat maintenance and deletes it through retention", 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); + await db.update(chatVoiceInboundCalls).set({ expiresAt: new Date(0), updatedAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id)); + const onError = vi.fn(); + const maintenance = createChatReconciliationCoordinator({ + reconcileVoiceSessions: () => f.service.reconcileMaintenance(), + processPendingDeliveries: async () => undefined, projectRunMilestones: async () => 0, + flushPublications: async () => undefined, onError, + }); + try { + maintenance.reconcile(); await maintenance.drain(); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(1); + expect(onError).not.toHaveBeenCalled(); + } finally { maintenance.stop(); await maintenance.drain(); } + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await f.service.inbound.registerRetentionWork(coordinator, () => true).ready; + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(0); + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); } + }); + + it.each(["signed callback", "cleanup receipt"])("schedules retained history from a real %s", async confirmation => { + const f = await inboundFixture(); + 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 = 120_000) => { + 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.registerRetentionWork(coordinator, () => true).ready; + await settle(); expect(coordinator.nextWakeAt()).toBeNull(); await quiet(10_000); + if (confirmation === "signed callback") { + const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" }); + await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers); + } else { + await f.service.inbound.registerCleanupWork(coordinator, () => true).ready; + await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, + { approve: false, approvalCode: f.call.approvalCode }); + await settle(); + expect(f.transport.endSession).toHaveBeenCalledWith(f.call.providerSessionId); + } + await settle(); + const [saved] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)); + expect(saved.cleanupConfirmedAt).not.toBeNull(); + const dueAt = Math.max(saved.updatedAt.getTime(), saved.expiresAt.getTime()) + 7 * 86_400_000; + expect(coordinator.nextWakeAt()).toBe(dueAt); await quiet(); + await vi.advanceTimersByTimeAsync(dueAt - Date.now() + 1000); await settle(); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(0); + expect(coordinator.nextWakeAt()).toBeNull(); await quiet(); expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + it("returns a committed denial while provider cleanup is still waiting", async () => { const f = await inboundFixture(); let release!: () => void; diff --git a/server/src/app.ts b/server/src/app.ts index 4c582dc0faa..50c1596b1d2 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1301,6 +1301,7 @@ export async function createApp( chatChannels.voice.registerRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.voice.registerReplyWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.voice.inbound.registerRetentionWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 2b99724be8a..af91b88a76e 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -45,6 +45,7 @@ export const DELIVERY_QUEUES = { chatTelegramMaintenance: "chat-telegram-maintenance", chatReceivers: "chat-receivers", voiceSessionRecovery: "voice-session-recovery", + voiceInboundRetention: "voice-inbound-retention", voiceInboundCleanup: "voice-inbound-cleanup", voiceReplies: "voice-reply-delivery", chatCommandRegistration: "chat-command-registration", diff --git a/server/src/services/voice/voice-inbound-notifications.ts b/server/src/services/voice/voice-inbound-notifications.ts index 0cfe67d3d8f..2c4197dd6d4 100644 --- a/server/src/services/voice/voice-inbound-notifications.ts +++ b/server/src/services/voice/voice-inbound-notifications.ts @@ -5,9 +5,11 @@ 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"); + throw new Error("Invalid inbound voice 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}` : "*"); +export async function notifyVoiceInbound(tx: object, scope?: VoiceInboundScope) { + const hint = scope ? `${scope.kind}:${scope.id}` : "*"; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceInboundCleanup, hint); + if (scope?.kind === "admission") await notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceInboundRetention, hint); } diff --git a/server/src/services/voice/voice-inbound-retention.ts b/server/src/services/voice/voice-inbound-retention.ts new file mode 100644 index 00000000000..3176e91b1be --- /dev/null +++ b/server/src/services/voice/voice-inbound-retention.ts @@ -0,0 +1,81 @@ +import { and, eq, gt, inArray, isNotNull, isNull, lte, sql } from "drizzle-orm"; +import { chatEndpoints, chatVoiceInboundCalls, type Db } from "@paperclipai/db"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { voiceInboundScope, type VoiceInboundScope } from "./voice-inbound-notifications.js"; + +type RetentionEntry = { id: string; companyId: string; endpointId: string; connectionId: string; dueAt: number }; +const RETENTION_MS = 7 * 86_400_000; +const eligible = () => and(isNull(chatVoiceInboundCalls.sessionId), isNull(chatVoiceInboundCalls.intakeIssueId), + isNotNull(chatVoiceInboundCalls.cleanupConfirmedAt), inArray(chatVoiceInboundCalls.state, ["denied", "expired", "ended"])); +const due = (now: number) => and(eligible(), lte(chatVoiceInboundCalls.expiresAt, new Date(now - RETENTION_MS)), + lte(chatVoiceInboundCalls.updatedAt, new Date(now - RETENTION_MS))); + +/** Retain the existing seven-day policy for both timestamps. SQL rounds the wake + * up to a millisecond so PostgreSQL submillisecond timestamps cannot fire early. */ +export function voiceInboundRetention(db: Db) { + const cursors = new Map(); + async function page(scope?: VoiceInboundScope, cursor?: string, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + if (scope?.kind === "session") return []; + const rows = await db.select({ id: chatVoiceInboundCalls.id, companyId: chatVoiceInboundCalls.companyId, + endpointId: chatVoiceInboundCalls.endpointId, connectionId: chatEndpoints.connectionId, + dueAt: sql`ceil(extract(epoch from greatest(${chatVoiceInboundCalls.expiresAt}, ${chatVoiceInboundCalls.updatedAt})) * 1000) + ${RETENTION_MS}`, + }).from(chatVoiceInboundCalls) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceInboundCalls.endpointId), eq(chatEndpoints.companyId, chatVoiceInboundCalls.companyId))) + .where(and(eligible(), + scope?.kind === "admission" ? eq(chatVoiceInboundCalls.id, 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.map(row => ({ ...row, dueAt: Number(row.dueAt) })); + } + async function load(scope?: VoiceInboundScope, signal?: AbortSignal): Promise { + const entries: RetentionEntry[] = []; + let cursor: string | undefined; + for (;;) { + const rows = await page(scope, cursor, signal); + entries.push(...rows); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + } + async function refresh(scope?: VoiceInboundScope, signal?: AbortSignal) { + const key = scope ? `${scope.kind}:${scope.id}` : "*"; + const rows = await page(scope, cursors.get(key), signal), complete = rows.length < 100; + signal?.throwIfAborted(); + if (complete) cursors.delete(key); else cursors.set(key, rows[rows.length - 1].id); + return { rows, complete }; + } + async function check(id: string, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + const deleted = await db.transaction(async tx => { + // Eligibility and both age conditions are current-state predicates under + // the admission lock. Cached discovery cannot delete a changed binding. + const [row] = await tx.select({ id: chatVoiceInboundCalls.id }).from(chatVoiceInboundCalls) + .where(and(eq(chatVoiceInboundCalls.id, id), due(Date.now()))).for("update"); + if (!row) return false; + signal?.throwIfAborted(); + return (await tx.delete(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.id, id), due(Date.now()))) + .returning({ id: chatVoiceInboundCalls.id })).length > 0; + }); + if (deleted) return null; + return (await load({ kind: "admission", id }, signal))[0] ?? null; + } + /** Explicit compatibility path; never register this bulk operation on a tick. */ + async function prune() { + await db.delete(chatVoiceInboundCalls).where(due(Date.now())); + } + function register(coordinator: ReturnType, canRun: () => boolean) { + cursors.clear(); + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.voiceInboundRetention, decodeScope: voiceInboundScope, + matches: (entry, scope) => !scope || scope.kind !== "session" && (scope.kind === "admission" ? entry.id : scope.kind === "endpoint" ? entry.endpointId : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id, + load: refresh, check: (entry, signal) => check(entry.id, signal), + }, canRun); + } + return { load, check, prune, register }; +} diff --git a/server/src/services/voice/voice-inbound-service.ts b/server/src/services/voice/voice-inbound-service.ts index c43ec1f3782..5c781d5a21b 100644 --- a/server/src/services/voice/voice-inbound-service.ts +++ b/server/src/services/voice/voice-inbound-service.ts @@ -1,9 +1,10 @@ +import { voiceInboundRetention } from "./voice-inbound-retention.js"; 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, isNotNull, sql } from "drizzle-orm"; +import { and, desc, eq, gte, inArray, isNull, 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"; @@ -122,8 +123,9 @@ 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."); - 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(); + const admissionId = randomUUID(); + await notifyVoiceInbound(tx, { kind: "admission", id: admissionId }); + const [admission] = await tx.insert(chatVoiceInboundCalls).values({id: admissionId, 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"); @@ -283,15 +285,13 @@ export function voiceInboundService(db: Db, options: { 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)))); - } - /** Explicit one-shot compatibility path. Recurring maintenance only owns retention. */ + const retention = voiceInboundRetention(db); + const pruneHistory = retention.prune; + /** Explicit one-shot compatibility path; never used by periodic maintenance. */ 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}; + return {configuration, saveConfiguration, lifecycle, pendingTool, list, decide, reconcile, pruneHistory, registerCleanupWork: cleanup.register, registerRetentionWork: retention.register, authorizeGuest, history}; } diff --git a/server/src/services/voice/voice-session-service.ts b/server/src/services/voice/voice-session-service.ts index e262900772e..49a2cc796d5 100644 --- a/server/src/services/voice/voice-session-service.ts +++ b/server/src/services/voice/voice-session-service.ts @@ -230,7 +230,6 @@ export function voiceSessionService(db: Db, options: { return registerVoiceReplyWork(coordinator, replies, canRun); } async function reconcileMaintenance(limit = 25) { - await inbound.pruneHistory(); await reconcileReports(limit); } /** Explicit one-shot recovery for diagnostics/tests; never register on a recurring lane. */