fix(server): schedule inbound voice history retention by deadline (#17583)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Incoming voice calls leave history after provider cleanup is
confirmed.
> - The chat maintenance tick repeatedly deletes old admission history.
> - Each eligible row already contains the dates needed for its
retention deadline.
> - This pull request schedules that deadline and wakes on committed
changes.
> - Idle retention stops querying while the existing seven-day policy
remains enforced.

## Linked Issues or Issue Description

**What existing behavior does this improve?**

Inbound voice history retention. This PR follows merged #17580, which
owns physical call cleanup, and now targets master. Call reports remain
a separate migration.

**Current behavior**

Voice maintenance scans admission history repeatedly. It deletes
confirmed, terminal, unbound admissions when both expiry and last update
are at least seven days old.

**Proposed behavior**

Load existing retention deadlines at startup. Wake from the existing
admission transaction notifications. Delete at the later timestamp plus
seven days. Lock and recheck current eligibility before deletion.

**Reason and benefit**

Configured idle endpoints and future history do not cause recurring
retention SQL. Failed deletion retains a retry. Restart reconstructs the
same saved deadlines.

## What Changed

- Add a voice-owned retention worker using the existing scoped deadline
helper and coordinator. Yield after 100 history rows per scope refresh
and reset reconstruction on registration. Keep complete one-shot
inventory reads. No table, field, or migration is added.
- Extend `notifyVoiceInbound` to notify retention before the same
committed writes that already notify cleanup. Callers need no new
recovery wrapper.
- Preserve confirmation, terminal-state, session, task, and seven-day
eligibility conditions. Round PostgreSQL submillisecond deadlines up.
- Lock and recheck the current row before deletion. New bindings or
moved dates invalidate stale candidates.
- Remove history pruning from recurring voice maintenance. Keep the
explicit one-shot diagnostic path. Leave call reports for their own
slice.
- Document producer ownership, retry/drain behavior, tests, and external
wake limitations.

## Verification

Published head: `53f2fafcb64d104887eb4455c292b48e6dd10253`. It includes
the normal merge of actual landed master
`71b448b9cb841e6fa18fe883585430c5081245da` after cleanup #17580. That
ancestry integration at `768c650365` preserved the complete binary diff
of all 13 reviewed retention paths. This follow-up changes only the
existing cleanup fairness test and its failure ledger, bringing the
final scope to 14 paths; all production source is unchanged from that
integration. All unrelated master changes, cleanup repair, migration
`0396_tan_paladin.sql`, and gateway clock repair are preserved. Database
files, shared worker, workflows and lockfile match actual master. The
earlier 282-case/type/build evidence below belongs to unchanged
retention source at its stated head; current repair acceptance is 37
distinct complete cleanup and retention lifecycle cases. All 58
current-head checks are now clean, including the Canary Dry Run;
Greptile is exact-head 5/5 with no new actionable issues. The current
Ponytail turn stopped without a verdict, which is an authorized
nonblocking state. The earlier failing CI remains preserved below.

- **282 distinct targeted cases pass**: all 127 voice integration cases
and all 155 cases in ten complete retention, cleanup, recovery,
company-removal, shared-worker, startup and hostname suites. These
include 13 retention and 14 cleanup lifecycle cases. Voice integration,
parent recovery and company removal start on separate empty databases
and migrate themselves. Retention uses its dedicated database after the
setup repair; cleanup uses its already-migrated dedicated database.
- Ponytail's prior exact-head 4/5 review found configuration-driven
history rescans. Five real connection saves selected 1,025 rows over 205
future admissions before repair. Retention now receives only admission
hints and startup reconstruction; the same saves select zero admission
rows and retain the original deadline. Cleanup still receives its full
configuration and authority scopes. Restrictive binding foreign keys
prevent implicit unbinding outside the admission-write boundary.
- Successful deletion now returns its confirmed result without reloading
the deleted row. A real SQL-count case and a guarded negative probe
cover that path. Changed or not-yet-due history still reloads current
eligibility and its deadline.
- Server typecheck, build and module-boundary checks pass. No schema,
driver, shared scheduler contract or provider protocol changes are added
by this slice.
- A real 105-row future-history case failed before the reader repair
because startup selected 105 rows rather than yielding at 100. The
existing paged shared-worker contract now bounds each scope refresh. The
same-owner restart case reconstructs every deadline and deletes all 105
eligible rows at the original deadline. It verifies SQL quiet before
that deadline and after settlement. Removing only cursor reset fails at
five rows instead of 100 after restart; guarded source was restored
before final acceptance.
- Existing cases retain the later-of-two-timestamps seven-day policy,
submillisecond rounding, current-state confirmation and binding checks,
concurrent binding lock, rollback/outer commit, first write during
drain, failure retry and shutdown. Real signed ingress, cleanup
confirmation and maintenance-lane removal remain covered by the complete
voice suite.
- The fresh retention fixture first failed on missing schema and now
applies the application migrator. The initial parent merge duplicated a
test import and failed to parse before any voice case ran. Both failures
are retained in the audit; the corrected complete suites pass.
- Prior head `daca270a99` had a failed server (4/12) check, run
40557790710/job 122386307340. Its actual log reports a runner shutdown
and cancellation before completion. This is incomplete acceptance, not a
pass or a proven product flake. New-head CI must pass independently.
- [Current-parent acceptance and failure
ledger](https://github.com/paperclipai/paperclip/blob/68b62b6d13f28ac5f17f35b98adf5488752f9b97/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md#current-parent-paging-and-restart-acceptance-october-28)
records the exact scope, commands, counts and preserved failures.
Earlier 218/133/134-case and full-workspace results belong to their
stated heads. No new full local repository suite, live provider, Cloud
wake, host/database suspension or savings claim is made.
- [Notification review repair and current 282-case
evidence](https://github.com/paperclipai/paperclip/blob/68b62b6d13f28ac5f17f35b98adf5488752f9b97/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md#configuration-rescan-review-repair-october-28)
records the new source. The prior head had green CI and a 4/5 Ponytail
review; neither is accepted for this new commit. Fresh exact-head
Greptile 5/5 and all CI remain required. Completed actual failing
Ponytail findings must be repaired; a stopped or paused turn without a
verdict is not a merge blocker under the user’s authorization. An active
Ponytail review waits at most 30 minutes from its original request
(25-minute monitor cutoff).

- **Preserved current-master CI failures:** run `40697141057` failed
server shard 7/12 in cleanup’s fairness test and Canary Dry Run on an
npm `rollup-4.66.0.tgz` 404. A fixed final admission UUID reproduces the
same fairness assertion: the second company was sometimes beyond the
first 100-row page, so expecting it among the first 25 calls was
invalid. The corrected test controls time and proves service within the
next 25-call batch after discovery, all 105 successful calls before the
failed provider’s saved retry, exactly one retry at its deadline, and
idle SQL after settlement. Bypassing company rotation fails the
corrected assertion; the shared helper was restored byte-for-byte. Both
complete cleanup (24) and retention (13) suites pass after restoration.
An intermediate 23/24 run incorrectly demanded idle before
reconstruction finished; the audit preserves that failure and its
corrected pending-intent wait. No production behavior, timeouts,
dependencies or workflow gates changed. The exact npm metadata and
tarball now return 200; new-head canary CI must pass independently.
Prior Greptile/Ponytail 5/5 on `768c650365` are stale for this new head.
[Full CI repair
ledger](https://github.com/paperclipai/paperclip/blob/53f2fafcb64d104887eb4455c292b48e6dd10253/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md#current-master-ci-repair-october-28).

## Risks

- Supported admission writers must use the existing transaction
notification boundary. Independent SQL writers and additional processes
need a separate wake contract.
- Retention never deletes unconfirmed cleanup or session/task-bound
history. Those records remain with their owners.
- The in-process deadline does not wake stopped compute. This PR does
not extend Cloud alarms or relax sleep eligibility, and does not claim
deployed host/database suspension.
- Call-report reconciliation and other chat maintenance still need
separate migrations.

## Model Used

OpenAI Codex with reasoning, shell, and code-editing tools. The exact
model version and context-window size are not exposed in this session.

## 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
- [x] All Paperclip CI gates are green for this new head
- [x] Greptile is 5/5 on this new head with no unresolved actionable
findings
- [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:57:58 -05:00
1 parent 71b448b9cb
commit 980775e0e9
14 files changed
+928 -23

No files matched your search

+47
View File
@@ -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.
@@ -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.
@@ -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.
@@ -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).
@@ -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()");
});
@@ -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<typeof chatVoiceInboundCalls.$inferInsert> = {}, 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();
});
});
@@ -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<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 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<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: "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<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 = 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<void>(resolve => { release = resolve; });
const ready = new Promise<void>(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();
});
});
@@ -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;
+1
View File
@@ -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();
@@ -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",
@@ -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);
}
@@ -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<string, string>();
async function page(scope?: VoiceInboundScope, cursor?: string, signal?: AbortSignal): Promise<RetentionEntry[]> {
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<string>`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<RetentionEntry[]> {
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<RetentionEntry | null> {
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<typeof createDeliveryWorkCoordinator>, 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 };
}
@@ -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};
}
@@ -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. */