Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-10-28 20:21:50 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(server): schedule voice call reports from saved deadlines (#17586)

Browse Source 
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Voice calls retain reports and project guest transcripts into intake
tasks.
> - Chat maintenance repeatedly scans recent calls even when no report
is due.
> - Empty scans keep the tenant database active.
> - Existing session and report rows already contain the work and its
deadlines.
> - This pull request moves report refresh to the shared coordinator and
removes the final voice maintenance lane.
> - Idle voice configuration and old call history no longer cause report
scans.

## Linked Issues or Issue Description

Refs #17583, #17580, #17578, #17576. This is slice 25 of the
recurring-server sleep stack. It now follows merged #17583 and targets
master.

**What existing behavior does this improve?**

Voice call-report refresh and guest transcript projection.

**Current behavior**

The chat interval scans terminal calls for reports. It repeats the scan
even when there are no eligible calls or all saved deadlines are in the
future.

**Proposed behavior**

Committed session transitions and signed callbacks wake a scoped worker.
Startup reconstructs existing rows. Later refreshes use saved deadlines.
The worker becomes quiet after all eligible windows expire.

**Reason and benefit**

Stop empty database scans without losing late transcript and billing
updates. Keep the existing 24-hour refresh policy, provider response
ordering, and guest-task trust restrictions.

## What Changed

- Add a voice-owned report worker using the existing scoped deadline
helper. No queue table or field is added; the existing session table
receives a report-window index.
- Save each attempt and its retry before provider I/O. Compare that
ownership on settlement so an older response cannot overwrite a later
callback or claim.
- Preserve available reports when a response is incomplete or stale.
Preserve their finite refresh window.
- Route voice-change and reply call-ended writes through
transaction-aware report signals.
- Preserve guest transcript locks and use the observed document revision
when updating an existing transcript.
- Register the worker in app startup. Remove voice reconciliation from
the chat interval. Keep explicit one-shot diagnostics.
- Document producer scope, lifecycle behavior, tests, and external wake
limits.

## Verification

Published head: `758cfe15c3e6283d38ba512c8416ccce4d200862`. The
follow-up after `c28de398` changes only two documentation files; the
verified production, migration and test source is unchanged. It normally
merges landed retention/master
`980775e0e956eecdc28c73c9b4aca0407ad45e1a`. The 20 owned paths remain
the complete slice, including officially regenerated migration 0397, its
snapshot and required pruning of snapshot 0392. The non-database
reviewed patch was preserved during ancestry integration; the only
subsequent source edit makes one existing report test’s provider
responses call-specific. All production report behavior and unrelated
master changes are preserved. No shared worker, workflow or lockfile
change belongs to reports. All 58 current-head checks are now clean,
including the previously failing approval shard; exact-head Greptile is
5/5 with no new actionable issues. Ponytail’s current turn stopped
without a verdict, an authorized nonblocking state. The prior
conditional documentation requirement was fixed; old CI failures remain
recorded below.

- **Prior-head evidence at `9bf20467df`: 303 distinct targeted cases
passed**, not rerun wholesale for this integration: all 132 voice
integration cases; 170 cases across 11 complete report, retention,
cleanup, recovery, company-removal, shared-worker and startup/hostname
suites; and one complete snapshot/schema drift case. This includes 15
report, 13 retention and 14 cleanup lifecycle cases. Voice integration
and the four other database owners use separate initially empty
databases and migrate themselves. Report cases use their dedicated
database after the setup repair.
- Five real connection saves over 205 sessions with saved future report
deadlines selected 1,025 session rows before the notification repair.
They now select zero and preserve the original deadline. The report
notification helper accepts only exact session scopes. Configuration,
authority and connection hints retain their existing live
recovery/reply/cleanup behavior but do not rebuild report history. Real
session terminal writes, callback hints, report claims and receipts
still reactivate reports.
- Database/server typechecks and builds, migration numbering/safety, and
module-boundary checks pass. Final server types/build also pass after
integrating the retention review repairs. The parent 205-admission,
five-configuration-save test still observes zero retained-history rows
and preserves its deadline.
- The old eager reader selected all 105 saved-report headers before
yielding and failed the 100-row budget. Registered reconstruction now
uses the existing paged contract and yields at 100 rows per scope
refresh. A partial scan can restart on the same owner and reconstruct
every saved deadline. At due time all 105 report IDs are fetched exactly
once. Future deadlines remain quiet between actual retries.
- The company-fairness case places a small company deterministically on
page two and requires it in the next 25-work batch after discovery, with
all 105 reports processed before the failed provider's retry. Guarded
mutations catch a missing cursor reset and bypassed company rotation.
Original source was restored before final acceptance.
- At prior head `9bf20467df`, migration 0396 was generated with `pnpm
db:generate` on the parent 0395 lineage. It creates the partial
`(ended_at, id)` report-window index concurrently and replayably through
the existing migration runner. All 394 historical SQL files and the
journal prefix are unchanged; snapshots 0392 through 0396 remain. Fresh
application migration and two replay invocations succeed with a valid
index. The real plan case retains 20,001 old sessions, 205 recent
sessions, precision-safe keysets, and fewer than 1,000 rows removed by
filtering on each page.
- Existing real hangup/callback/reply-error producers, callback
ownership fencing, monotonic report results, saved retry before provider
contact, guest transcript projection after a document lock, transaction
rollback/nesting, drain and shutdown remain covered. No table, field or
provider protocol is added. Available reports retain the existing finite
24-hour refresh policy.
- [Current-parent acceptance and failure
ledger](https://github.com/paperclipai/paperclip/blob/9bf20467dfd50b81848e1c8dd7ddc905d206418f/doc/plans/2026-10-25-voice-call-report-sleep-audit.md#current-parent-reconstruction-and-migration-acceptance-october-28)
records the failed fresh fixture, eager read, fairness-clock attempts,
guards, commands, counts and limits. Older 174/138-case and
workspace-wide results belong to their stated heads. No new complete
local repository suite, live provider, Cloud wake, host/database
suspension or savings claim is made.
- [Report configuration-scope
repair](https://github.com/paperclipai/paperclip/blob/9bf20467dfd50b81848e1c8dd7ddc905d206418f/doc/plans/2026-10-25-voice-call-report-sleep-audit.md#configuration-scope-review-repair-october-28)
records the current 303 cases and measured rescan repair. The prior
`80207ba6bb` head had Ponytail 5/5; that review is stale after this
change. Fresh exact-head Greptile 5/5 and all CI remain required. Actual
failing Ponytail findings must be fixed; a stopped or paused turn
without a verdict is an authorized nonblocker. Active no-verdict review
waits are capped at 30 minutes from the original request/start
(25-minute monitor cutoff). Older scores and parent checks are not
inferred as current passes.

- **Current-master verification: 185 distinct cases pass**: final
complete voice integration 132, complete report/cleanup/retention
15+24+13, and snapshot/schema drift 1. Each database owner used its own
empty fixture database. The first combined run had 183 passing cases and
one report-test failure; it is preserved in the audit. A deliberately
earlier-ended competing call reproduces the old one-shot mock consuming
the intended response. The corrected test selects responses by provider
session ID, verifies independent costs for both calls, then rejects
stale target data. The isolated repair and full 132-case suite pass;
production behavior and timeouts are unchanged. These 185 cases are not
added to the older 303 count.
- **Current migration lineage:** official `pnpm db:generate` after
actual master creates `0397_known_madrox.sql`; the reviewed
concurrent/idempotent report-index SQL is retained byte-for-byte. All
395 existing SQL files and journal entries are unchanged; snapshots
0393–0397 remain and the new snapshot links parent 0396. The application
migrator boots a fresh database, repairs a deliberately invalid index
during two concurrent replay invocations, records one history row, and
replays again successfully. Database build, server build/typecheck,
migration checks and module boundaries pass. [Full current-master
acceptance and failure
ledger](https://github.com/paperclipai/paperclip/blob/c28de398fef074f650eff532952ab6f5563739a8/doc/plans/2026-10-25-voice-call-report-sleep-audit.md#landed-master-migration-and-acceptance-october-28).

- **Review follow-up and preserved CI timeout:** Ponytail’s prior
`c28de398` review was 4/5 advisory, with no must-fix production defect
and a request to fix the stale design-doc migration reference before
shipping. That document now names migration 0397 and links current
185-case acceptance; older 302/303-case evidence is explicitly
historical. The optional suggestions about one-shot diagnostic inventory
cost and inherited test-clock helpers remain disclosed. CI run
`40703490340` failed the unchanged approval-queue backoff test on its
existing 15-second timeout and then an activity-log FK in teardown
(2,097 pass/one fail/one skip in that shard). The complete unchanged
28-case approval suite passes on a new isolated database; the exact CI
timing cause remains unproven. No timeout, workflow or production logic
was changed, and no old-head rerun or failure waiver is claimed. This
documentation head requires fresh CI and exact-head Greptile. [Review
and timeout
ledger](https://github.com/paperclipai/paperclip/blob/758cfe15c3e6283d38ba512c8416ccce4d200862/doc/plans/2026-10-25-voice-call-report-sleep-audit.md#review-documentation-and-preserved-remote-timeout-october-28).

## Risks

- Process-local commit signals do not observe independent SQL writers or
another server process.
- Available reports still refresh for 24 hours. The provider's complete
flag does not promise immutable final data.
- Transcript projection retries remain within the existing report
window. This change does not add indefinite projection repair.
- One-shot diagnostic `reconcile(limit)` inventories the eligible window
before slicing; it remains a manual diagnostic cost, not recurring idle
work. Some inherited test helpers advance fake time during assertion
retries, an optional reliability follow-up noted by review.
- A local deadline cannot wake a stopped host. Cloud alarms and sleep
eligibility remain unchanged. No deployed sleep or cost saving is
claimed.
- This PR depends on the preceding stack. Final master reconciliation
and remote review gates remain required.

## Model Used

OpenAI GPT-6 through Codex, with repository inspection, code editing,
and local test execution. The exact serving model ID and context window
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
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-28 09:41:15 -05:00
1 parent 8fe5d5defc
commit 8049fbf51d
19 files changed
+1449 -90
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/idle-voice-call-reports.md
+108
View File
@@ -0,0 +1,108 @@
# Voice call reports and idle work
Voice report reconciliation uses the shared delivery coordinator and scheduler.
It no longer runs from the chat maintenance interval. Existing
`chat_voice_sessions` and `chat_voice_reports` rows own its durable state; this
change adds no queue table, provider endpoint, or independent timer. A concurrent
partial index supports bounded reconstruction of the existing report window.
## Trigger and deadline
A terminal session with a provider call ID remains eligible for report refresh
for 24 hours after `ended_at`, inclusive. A missing report is due immediately.
An existing report uses its saved `next_check_at`. The retry starts at 60 seconds
and doubles to one hour, as before. Available reports remain refreshable during
this finite window: provider availability is not a promise that transcript or
billing data can no longer change.
Session transitions and signed callbacks notify the report worker before their
writes, inside the owning database transaction. The reply sender's `call_ended`
result does the same. Healthy active-session observations do not activate report
work. Nested transactions wait for the outer commit; rollback does not publish
new work. The existing work-signals protocol owns uncertain outcomes.
Startup reconstructs eligible sessions in pages ordered by `(ended_at, id)`,
yielding after at most 100 rows per scope refresh. Re-registering the same owner
resets partial scan cursors. The explicit one-shot inventory still loads every
page when requested.
A partial index selects terminal sessions with provider IDs. The query uses
its ended-at range to restrict reads to the current 24-hour window. The cursor retains PostgreSQL timestamp precision. Committed report hints reload the exact changed session. Configuration,
authority and connection metadata changes do not alter report eligibility or
saved retries, so they do not wake report reconstruction. Each cached entry carries the
saved report deadline or the first millisecond after the refresh window ends,
whichever comes first. Submillisecond PostgreSQL timestamps round in the safe
direction. Expiring the window removes the cached obligation without a provider
request. Configured idle endpoints, active calls, old terminal history, and
sessions without provider IDs do not cause recurring report SQL or requests.
## Claims, responses, and projection
The worker locks and rechecks the session and report, then saves its attempt and
next retry before contacting the provider. Network I/O happens outside that
transaction. Settlement compares the saved attempt and retry deadline. A newer
claim or callback that moves the deadline invalidates the older response.
Incomplete or older responses cannot overwrite a retained available report.
Eight failures can mark a pending report unavailable; failure cannot downgrade
an already available report.
Guest transcripts remain confined to the call's quarantined intake task. The
worker reads the retained report, keeps the task's source-trust metadata, and
uses the existing document revision contract. A document lock or conflicting
edit leaves the saved report and retry intact. After the lock clears, the next
successful refresh retries projection. The existing 24-hour eligibility policy
still applies; this PR does not add indefinite transcript repair or extend it.
Provider credentials, call-report parsing, and delivery diagnostics remain in
the voice integration. The shared scheduler receives only normalized work and
deadlines. No provider-specific policy is added to app startup.
## Lifecycle and limitations
The shared scoped worker coalesces notifications, processes fair company batches,
and retains dirty scopes across drain or failed reads. Shutdown joins in-flight
work. Restart reconstructs the persisted retry instead of immediately repeating
the provider GET. The explicit one-shot voice reconciliation method remains for
diagnostics; production maintenance does not call it periodically.
The supported deployment has one active server process writing voice session
and report state through its owning database client. Active-active servers and
out-of-band SQL writers are not supported by this worker. Process-local signals
do not reach another process; such a deployment needs a durable cross-process
wake contract before removing its reconciliation backstop. This PR neither adds
that transport nor asserts that startup reconstruction covers concurrent writers. This is worker quiet, not a claim of physical host sleep. Cloud alarm
coverage and conservative sleep eligibility are unchanged; a local deadline
cannot wake stopped compute.
## Verification
`voice-call-reports.test.ts` exercises real PostgreSQL claims, configured-idle
quiet, finite expiry, timestamp precision, retries and restart, committed writes,
rollback and nesting, drain/resume, callback races, concurrent claims, backlog
fairness, result ordering, and shutdown. The full voice integration suite checks
real hangup/callback/reply-error producers and transcript projection after a lock.
The app wiring suite protects removal of the voice maintenance lane.
The audit and evidence ledger is
[voice call-report scheduling](plans/2026-10-25-voice-call-report-sleep-audit.md).
The [historical parent acceptance](plans/2026-10-25-voice-call-report-sleep-audit.md#current-parent-reconstruction-and-migration-acceptance-october-28)
records the earlier 302 distinct passing cases, the bounded startup and same-owner restart
regressions, deterministic company fairness, generated migration 0396 and the
preserved failures. The report owner keeps its finite provider refresh obligation;
this is not an unconditional idle scan or proof of deployed suspension.
The subsequent [configuration-scope repair](plans/2026-10-25-voice-call-report-sleep-audit.md#configuration-scope-review-repair-october-28)
records the earlier 303-case acceptance at `9bf20467df`. Five real connection saves over 205 saved
reports select zero session rows, after reproducing 1,025 selected rows before
the repair. The notification helper accepts only an exact session scope.
The [landed-master acceptance](plans/2026-10-25-voice-call-report-sleep-audit.md#landed-master-migration-and-acceptance-october-28)
records the current 185-case acceptance and generated migration
`0397_known_madrox.sql`, after landed cleanup migration 0396. It preserves all
existing SQL and journal entries, retains snapshots 0393 through 0397, and checks
fresh application migration plus concurrent replay and invalid-index recovery.
The older 302/303-case sections above are historical evidence, not the current
migration filename or a claim that those full groups were rerun on this head.
doc/plans/2026-10-24-recurring-server-sleep-stack.md
+55
View File
@@ -1617,6 +1617,38 @@ 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.
Slice 24 is published as [#17583](https://github.com/paperclipai/paperclip/pull/17583)
at `9c38ddfe53`, based on #17580 `4f0364d269`. It is attached to this chat.
CI/review are running; no merge. The next branch is
`codex/sleep-25-voice-call-reports`, with its owner audit recorded separately.
## Slice 25: voice call reports
Report refresh now owns a scoped deadline worker. Terminal transitions, signed
callbacks, and reply call-ended responses activate existing report/session rows.
Attempts save the next retry before provider I/O, and callback/new-attempt fences
prevent stale settlement. The existing 24-hour refresh obligation is preserved,
including available reports. Empty/configured-idle/live/old history is quiet.
This removes the last voice duty from recurring chat maintenance.
Acceptance: 174 selected real-database/wiring cases, workspace typecheck and build
pass. A new integration case exposed missing baseRevisionId on an existing guest
transcript; the fix uses the document revision contract and preserves lock
behavior. Full failure history and external-wake limits are in
`2026-10-25-voice-call-report-sleep-audit.md`. Publication and remote checks follow.
Parent #17583 has a Discord rollback/drain test failure to repair at its owner;
parent #17580 has two new cleanup review findings. Neither is marked complete.
The remaining chat duties, original ledger, #17490 E2E failure, and final
master/stack reconciliation remain required. No stack merge is authorized.
Slice 25 published as [#17586](https://github.com/paperclipai/paperclip/pull/17586).
Parent Discord test settlement and denial-latency repairs are propagated. At the
report tip, 41 Discord tests and 138 full voice/app tests pass; unchanged report,
cleanup and retention workers had 37 passing cases. CI/reviews remain required.
Next owner audit splits Teams unused-consent expiry from abandoned send/upload
attempt recovery, preserving one reason per PR. Company deletion cleanup policy
and the earlier E2E failure remain unresolved completion duties.
## Acceptance follow-up after the plugin parent merged (October 27)
Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at
@@ -1754,3 +1786,26 @@ 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.
## Report current-parent acceptance, October 28
Report scheduling #17586 integrates retention `68b62b6d13` and uses the existing
shared paged result contract. It yields after 100 headers per scope refresh,
resets same-owner partial scans and retains the finite 24-hour report obligation.
Officially generated migration 0396 preserves the preceding 394 SQL files and
retains the five newest snapshots. The [report acceptance ledger](2026-10-25-voice-call-report-sleep-audit.md#current-parent-reconstruction-and-migration-acceptance-october-28)
records 302 passing cases, types/build, migration/plan checks, guarded restart and
fairness probes, failures and limits. Current-head review, all CI and parent
landing still gate this slice. The full remaining stack stays in scope.
## Report notification review follow-up, October 28
Report reconstruction also received configuration hints that could not alter
its eligibility or saved retry. Five real connection saves over 205 saved
reports selected 1,025 session rows. The report owner now accepts exact session
notifications and startup; those saves select zero rows and preserve the deadline.
The [report notification ledger](2026-10-25-voice-call-report-sleep-audit.md#configuration-scope-review-repair-october-28)
records 303 passing cases, types/build, the narrower contract and limits. Fresh
exact-head review and CI are required after the repair.
doc/plans/2026-10-25-voice-call-report-sleep-audit.md
+446
View File
@@ -0,0 +1,446 @@
# Voice call-report scheduling sleep audit
Original baseline: voice history retention PR #17583 at
9c38ddfe5349ef1872306a91c7e4c5c1bc083259.
Current integrated parent: 68b62b6d13f28ac5f17f35b98adf5488752f9b97.
Branch: codex/sleep-25-voice-call-reports.
Status: implementation and local acceptance complete; publication and remote gates pending.
## The obligation
`voiceSessionService.reconcileReports` is the final voice duty called from the
chat maintenance cadence. It scans terminal sessions with a provider call ID and
endedAt within the last 24 hours. A missing report is immediately due; an existing
report uses nextCheckAt. The existing exponential retry is 30 * 2^attempts seconds,
capped at one hour. Its first post-fetch retry is 60 seconds.
The current policy deliberately revisits even available reports during that
24-hour window. Do not assume available means no future obligation: later provider
reports can update transcript and billing, and delivery diagnostics may still
need their final snapshot. The provider parser's complete flag only means the
response has a report object. It is not an explicit final-and-immutable promise.
Preserve this finite refresh window while replacing the repeated global scan.
No recording URL or provider metadata is stored; the provider parser allowlists
spoken turns and safe billing/duration values.
## Ownership and implementation direction
Existing chat_voice_sessions and chat_voice_reports remain durable truth. Keep
selection, refresh policy, fetch, result ordering and projection within the voice
owner. Use the scoped deadline helper for coalescing, fair batches, backoff, drain
and shutdown. Register a report topic on the existing voice-change composition;
do not add caller recovery wrappers or a queue table.
- Startup loads paged terminal-session candidates and actual nextCheckAt values.
- Missing reports are due at terminal transition. Future nextCheckAt remains
unchanged across inspection/restart.
- A report entry also has a finite window end. Expire its cached obligation when
the 24-hour window ends, even if its next attempt would be later. Define SQL
timestamp precision and inclusive boundary consistently; do not spin at cutoff.
- Lock/recheck current session eligibility and report deadline before reserving
an attempt. Save retry intent before network I/O. Provider GET remains outside
database transactions. Inspect whether a concurrent manual reconciliation or
callback can invalidate a result; use existing persisted fields for ownership
where sufficient. Do not add speculative leases or schema without evidence.
- Preserve providerUpdatedAt ordering. An incomplete or older response must not
replace a newer available report. Stale responses must still back off.
- Preserve the guest-task transcript projection, its advisory lock, low-trust
quarantine check, owner access, and activity log. A saved report is not proof
that projection completed. Preserve retry after a projection failure.
- Preserve bounded diagnostic attempts and never replay speech. Diagnostic
failure must not prevent transcript/history reconciliation.
- Keep explicit one-shot diagnostic reconciliation compatible. Remove the
recurring voice lane only after all report producers are covered.
## Producer audit
- `voice-inbound-service.ts` signed terminal and report callbacks call
notifyVoiceChange before the session update and report upsert. Existing reports
get nextCheckAt = now, which may advance a scheduled refresh.
- `voice-session-service.ts` close and inspect paths use withVoiceSessionChange.
- `voice-session-recovery.ts` uses full voice-change notification for terminal
inspection results. Healthy inspection and error-only updates intentionally
notify only recovery; do not make those trigger empty report scans.
- `voice-reply-push.ts` handles HTTP 409 by writing state=ended and endedAt. It
currently notifies reply, recovery and inbound explicitly; report notification
must be included before this terminal write.
- Configuration/authority/company changes use notifyVoiceChange composition.
- The existing report loop inserts and updates report rows without transaction
notifications. Move that ownership into the new report service. Account for
callbacks during an in-flight provider response and for any remaining direct
writes discovered in the full source audit.
## Verification required
- Configured idle, live calls, old terminal history and calls without provider IDs
are SQL/network quiet after startup.
- Real session end, signed callbacks, and reply call-ended results wake reports.
- Saved refresh deadlines survive restart; no early reads, bounded failures, and
final window expiry disarm. Available reports retain the existing window.
- Nested commit/rollback, moved-earlier callback deadlines, concurrent writes,
first work during drain, failed work entering drain, and shutdown keep work.
- Multi-page backlog and per-company fairness do not starve later calls.
- Provider failure, incomplete responses, older provider timestamps, guest
transcript projection failures and diagnostic behavior retain correct data.
- Existing full voice integration/history/access tests pass on a fresh database.
No Cloud alarm coverage or sleep-eligibility expansion is authorized by this
migration. A future in-process refresh cannot wake stopped compute. Preserve
current external-wake limitations explicitly.
## Parent state
#17583 is published and attached. Its 218 selected tests, workspace types/build,
and final server typecheck pass locally; fresh CI/reviews are running.
#17580 migration-runner repair 4f0364d269 has fresh Greptile 5/5. At the last
inspection it had no failing checks and three pending checks. Its prior database
setup failure and regression/repair evidence remain in its audit.
The complete original recurring-server ledger, older #17490 E2E failure,
remaining PR reviews and final master/stack reconciliation remain required.
## Implementation and acceptance, 2026-10-25
The parent retention repair at `360e1253a0` is incorporated. The report owner now
loads scoped, paged obligations, claims each due attempt before network I/O, and
uses attempt/deadline ownership for settlement. Report writes notify the same
shared coordinator. Existing voice-change producers and the explicit reply
`call_ended` path activate it. App startup registers it once and the final voice
lane is removed from the chat interval. No schema or Cloud eligibility change.
The saved report worker suite passed 13 tests. Initial full integration acceptance
had 136 passes and one failure: an unlocked existing transcript still could not
update because the copied projection omitted the document's base revision ID.
The repair supplies the observed `latestRevisionId`, preserving document locks
and conflict checks. A focused rerun against the already-used full-suite database
verified the document update but failed its global provider-call count because
133 retained fixture calls were eligible; this is not accepted as a clean run.
Final full acceptance uses a new database migrated through the application
migrator. Results are recorded below when terminal.
Earlier worker fixture failures (assignedAgentId vs agentId) and the production
missing `or` import were repaired before the 13-case worker pass. Source wiring
assertions still expecting reconcileMaintenance were updated to require its
removal. Full-suite tests remain Vitest coverage; server tsconfig excludes tests,
so production typecheck is not described as test typechecking.
Parent #17583 at `360e1253a0` has a failed server shard: the Discord rollback/drain
quiet test observes one scoped select after its nextWakeAt becomes null. The job
log is retained at `/tmp/sleep24-review-ci-failure.log`. This needs investigation
at its owning earlier slice and propagation; it is not silently retried or
claimed green. Parent #17580 also has two review findings about cleanup latency
and deletion with unconfirmed provider cleanup. These remain open. The older
#17490 E2E failure and final master reconciliation remain part of completion.
Final local acceptance after the revision fix: 137 full voice/app wiring cases
passed on `paperclip_sleep25_projection_fixed_20261025_test`; 37 report/cleanup/
retention worker cases passed serially on the isolated worker database. Workspace
`pnpm -r typecheck` and `pnpm build` passed. Logs: `/tmp/sleep25-voice-fixed-acceptance.log`,
`/tmp/sleep25-voice-workers-acceptance.log`, `/tmp/sleep25-workspace-types.log`,
and `/tmp/sleep25-workspace-build.log`. These 174 selected cases do not claim the
entire `pnpm test:run` suite passed locally. This host's embedded PostgreSQL test
initialization limitation remains; CI must run the complete relevant gates.
## Publication and parent repairs
Published as #17586. The Discord quiet-test settlement repair is propagated from
#17557 and passes all 41 command ownership tests at the report branch. The inbound
denial latency fix is also propagated; its held-provider regression is included.
After both merges, all 138 full voice/app cases pass on a fresh migrated database
(`/tmp/sleep25-denial-acceptance.log`). The earlier 37 worker cases remain valid;
those worker implementations are unchanged by the parent repairs. Remote checks
and reviews are required for the final pushed head. Parent #17580's company
deletion finding remains unresolved; denial latency finding is resolved.
## Paperclip review: report window index and writer scope
Review findings 4370594150 and 4370594357 identified deployment-contract clarity
and expensive report reconstruction. The supported writer model is now explicit:
one active server process owns voice writes through its DB client. Active-active
servers and independent SQL mutation are unsupported without a separate durable
wake contract. This is a documented scope restriction, not enforcement added by
this PR, and not a claim that startup recovers concurrent foreign writers.
The query finding reproduces on fresh PostgreSQL with 20,001 old sessions and
205 eligible sessions: an actual reconstruction query filters 20,001 historical
rows. The repair adds partial `(ended_at, id)` index 0389 and uses that order and
keyset cursor. The cursor retains PostgreSQL text timestamp precision, so sessions
100 microseconds apart do not repeat or disappear across a page. All 14 report
lifecycle cases pass after the migration; the query-plan regression checks all
three pages and each stays below 1,000 rows removed by filtering. Report scope,
24-hour policy, and provider behavior are unchanged.
The generated migration is concurrent and idempotent and registered with the
application's top-level migration runner. It applies and a second application is
a no-op. Migration numbering/safety checks pass. The generator retained exactly
five snapshots, 0385 through 0389, pruning 0384 per repository policy.
Retained failures: the first probe reused a database with two unrelated eligible
fixture sessions and failed its exact fixture count. It is not performance proof.
A fresh database reproduced the 20,001-row filtering problem. The first migration
invocation used unsupported top-level await in tsx eval and did not apply; its
subsequent test still failed without the index. The corrected chained invocation
applied/replayed successfully and all 14 cases passed. No assertion was relaxed.
Logs: /tmp/sleep25-report-index-before.log (fixture contamination),
/tmp/sleep25-report-index-clean-before.log (cost reproduction),
/tmp/sleep25-report-index-apply.log (invalid invocation),
/tmp/sleep25-report-index-after.log (unapplied index),
/tmp/sleep25-report-index-apply-fixed.log (application/replay),
/tmp/sleep25-report-index-fixed-tests.log (14 pass).
Workspace typecheck and build pass. Fresh full voice integration and app wiring
pass all 138 cases after the index repair. Logs:
/tmp/sleep25-report-review-{types,build,acceptance}.log.
Publish and propagate through Teams #17590/#17592 after acceptance. Fresh remote
review/checks and all prior full-stack requirements remain open.
## Current-parent reconstruction and migration acceptance, October 28
The report branch normally integrates retention `68b62b6d13` and its cleanup,
recovery and reply foundations. The retention repair prevents configuration saves
from rescanning history. Its real 205-admission, five-save regression is retained
and passes in the complete voice suite here. Both report and retention tests were
preserved when their additions conflicted at the same point in that suite.
The original 20 owned paths remain the complete report slice. Shared scheduler,
workflow and lockfile changes are inherited context, not additions by reports.
### Bounded discovery, restart and fair dispatch
The old report `load` executed every page before yielding. A new 105-session
fixture with saved future report deadlines selected 105 rows in its first pass,
failing the 100-row startup budget. The registered worker now returns one page
per scope refresh through the existing shared `{ rows, complete }` contract.
The domain owner retains full explicit `load` and one-shot reconciliation APIs.
The keyset cursor still stores PostgreSQL timestamp text, not a truncated Date.
Only successful, non-aborted reads advance a cursor, and registration resets it.
The real same-owner restart case stops after 100 of 105 saved deadlines,
re-registers the service, reconstructs the original deadline, and refreshes each
of the 105 reports once at that time. It observes no SQL/provider activity before
the deadline or between saved retries afterward. Removing the cursor reset fails
at five rows instead of 100 on restart. This measures selected rows, not physical
tuple visits or instance cost.
Fairness applies to discovered work. The prior test demanded that a company
beyond the first page run in the initial batch; that assumes an eager inventory.
The revised test puts that company deterministically on page two and requires
it in the next 25-work batch as soon as it is discovered. It also verifies all
105 calls run and the failing provider retains one saved attempt without
starving the backlog. Bypassing company rotation makes the next-batch assertion
fail. Both guarded mutations restore the exact original source afterward.
Preserved test failures distinguish this contract change from production defects.
The first bounded full run failed the old first-batch assertion. The first
controlled-clock revision allowed random UUID order to put the other company
on page one; a fixed final key now makes page placement deterministic. An
intermediate revision expected zero active intent holds before all dirty scopes
settled. The test now uses 20 controlled one-second scheduler turns, retaining
those holds until reconciliation completes, and finishes before the provider's
60-second retry. It still checks fair dispatch, all 105 calls, one failed attempt,
and quiet after settlement. It does not advance time in every assertion retry.
The external fixture first failed on missing tables. It now applies the actual
application migrations with the standard setup timeout before creating data.
The failed setup and 105-versus-100 budget remain recorded. No test is skipped
and no eligibility or provider assertion was removed to repair setup.
### Generated migration lineage
The old unmerged report migration 0393 conflicts with the landed checkpoint and
cleanup lineage. The complete retention parent migration tree and client were
restored, then the unchanged report-index schema edit was regenerated with
`pnpm db:generate`. The result is `0396_careful_blindfold.sql`. It creates the
existing proposed partial `(ended_at, id)` index concurrently and replayably,
through the application's top-level migration runner. No table or field is added.
All 394 parent SQL files and the journal prefix are preserved byte-for-byte.
Snapshots 0392 through 0396 are the five retained snapshots. The generator's
required deletion of 0391 is included; no retained snapshot or ID was hand-edited.
Generation used parent `ffe9d47b`; retention review head `68b62b6d13` has the exact
same database tree. The snapshot/schema drift regression passes. The report
fixture boots through the application migrator, and two later complete migration
invocations succeed with the report index valid. The existing query-plan case
still tests 20,001 old sessions, 205 eligible sessions and all three precision-safe
pages; each query removes fewer than 1,000 rows by filtering.
### Final local evidence
302 distinct cases pass on this combined source:
- 131 complete voice integration cases, including real hangup, callback and reply
terminal producers, transcript projection after a document lock, and the parent
retention amplification regression. This suite uses an initially empty DB.
- 170 cases across 11 complete report, retention, cleanup, recovery, company
removal, shared deadline/coordinator/batching, startup and hostname suites.
This includes 15 report, 13 retention and 14 cleanup lifecycle cases. Recovery,
company removal, retention and cleanup use separate initially empty DBs and
migrate themselves. Reports use their dedicated DB after the setup repair.
- One complete snapshot/schema drift case. The schema and migration tree are
identical to the checked generation after the final retention merge.
Database and server typechecks/builds, migration numbering/safety and the module
boundary check pass. Earlier 174/138-case and workspace-wide claims apply only to
their stated heads. No new full local repository suite, live provider, Cloud wake,
physical suspension or savings claim is made. The 24-hour mutable-report window,
retry, confirmation, callback ownership and quarantined transcript policy remain.
Evidence logs are `/tmp/sleep25-final-voice-consumers.log`,
`/tmp/sleep25-final-lifecycle-consumers.log`,
`/tmp/sleep25-final-snapshot-drift.log`,
`/tmp/sleep25-final-integrated-{types,build}.log`,
`/tmp/sleep25-current-migration-{check,replay}.log`, and
`/tmp/sleep25-current-migration-lineage-proof.json`. Failed setup, eager paging,
and the three fairness-clock attempts are in `/tmp/sleep25-empty-fixture-red.log`,
`/tmp/sleep25-page-budget-red.log`, `/tmp/sleep25-bounded-report-first-full.log`,
`/tmp/sleep25-bounded-report-green.log` and
`/tmp/sleep25-bounded-report-fairness-green.log`. The passing initial full report
run is `/tmp/sleep25-bounded-report-final-green.log`; its 15 cases are included
in the final 170 and not counted twice. Negative probes are
`/tmp/sleep25-{cursor-reset,company-fairness}-mutant-red.log`.
Fresh exact-head review, CI and earlier stack landing remain required. Cleanup
parent `b8b12199b9` completed green CI after its single recorded E2E retry, but
lacks current Ponytail acceptance. Retention's earlier 4/5 review was repaired
in `68b62b6d13`; that new parent still requires its own fresh reviews and CI.
Neither parent state is a waiver for reports or for remaining stack duties.
## Configuration scope review repair, October 28
The earlier security thread correctly identified another source of repeated
report inventory. A voice connection save reached `notifyVoiceChange`, which
notified reports for connection, endpoint, company or global scopes. Bounded
paging controlled each burst but still reloaded the whole eligible set after
separate saves. A real producer regression seeds 205 terminal sessions with saved
future report deadlines, then performs five separate connection saves. It selects
1,025 session rows before repair and zero afterward. The original next deadline
is unchanged, and no provider GET occurs during the saves.
The composed notification now signals reports only for actual session changes.
The report-specific helper accepts only an exact session scope, preventing a
future configuration caller from reintroducing broad hints through that boundary.
Startup still uses the shared worker's initial paged reconstruction. Report claims,
receipts and signed callbacks keep exact session notifications. Existing terminal
hangup and reply-ended paths remain covered through their real producer APIs.
Live recovery, replies and inbound cleanup retain their own broader authority
and configuration signals; their eligibility can depend on those changes.
The report predicate depends on session terminal state, provider identity and
endedAt, plus saved report state. A configuration save changes none of them.
Previously such a save only reloaded nextCheckAt and did not advance a future
provider attempt; the repair retains that same saved retry. There is no new
credential policy, refresh cadence, schema or scheduler contract. Manual inventory
APIs remain complete. Unsupported external SQL still needs its own wake contract.
303 distinct cases pass after the repair: 132 complete voice integration cases,
170 complete lifecycle/shared/startup cases, and the unchanged schema/snapshot
case. The voice suite uses a fresh separate DB; the lifecycle suites use their
existing dedicated isolated DBs. Server typecheck/build and module boundaries
pass. The earlier 302 count applies to `80207ba6bb` and is not added to this count.
The current Ponytail 5/5 on that earlier head cannot be reused after this edit;
fresh exact-head review and all CI remain required.
Logs: `/tmp/sleep25-config-amplification-red.log`,
`/tmp/sleep25-scope-final-voice-consumers.log`,
`/tmp/sleep25-scope-final-lifecycle-consumers.log`,
`/tmp/sleep25-scope-final-types.log`, `/tmp/sleep25-scope-final-build.log`, and
`/tmp/sleep25-scope-final-module-boundaries.log`. No full local repository suite,
live provider, Cloud wake or physical suspension claim is added. The original
20-path report scope and the existing finite 24-hour obligation remain intact.
## Landed-master migration and acceptance — October 28
The normal merge uses actual master `980775e0e956eecdc28c73c9b4aca0407ad45e1a`
after cleanup and retention landed. The reviewed report head was
`9bf20467dfd50b81848e1c8dd7ddc905d206418f`, based on retention `68b62b6d13`.
An explicit-base composition preserved the complete non-database report patch
before the acceptance-test repair below. Only the shared client, journal, and
396 snapshot conflicted. Those metadata files were restored from actual master
before invoking the official `pnpm db:generate` workflow.
The generated report migration is now `0397_known_madrox.sql`. It retains the
reviewed partial `(ended_at, id)` concurrent, idempotent index SQL byte-for-byte.
The existing top-level migration registry adds only the new filename and keeps
landed cleanup migration 396 registered. All 395 existing SQL files and the
complete journal prefix match actual master byte-for-byte. Snapshots 393 through
397 remain, with the generated 397 snapshot pointing to actual parent 396.
No landed migration, snapshot contents, or journal entry was renumbered.
A new empty PostgreSQL database boots through the application migrator with a
valid report index. A deliberately failed unique concurrent index build under
the report index name leaves an invalid index; two concurrent invocations of
the application migrator repair it, restore the intended report index, and
record one migration history row. Another replay succeeds. This exercises the
new 397 registration and invalid-index recovery rather than assuming that a
successful Drizzle generation proves executable migration acceptance.
The first complete four-suite run failed: 183 passed and one report integration
case failed, 184 total. `retains one safe report per call and rejects history
after access revocation` expected cost 240000 but retained 120000. The test used
a global one-shot transport response while reconciliation correctly visits
other due calls. An unrelated call could consume the target's newer response.
Adding an explicit earlier-ended competing call reproduced the same assertion
in isolation on unchanged production. The test now selects responses by the
target provider session ID, asserts the target accepts 240000, asserts the
other call independently keeps 120000, and then proves stale cost 1 cannot
overwrite the target. Existing access-revocation and one-report-per-call
assertions remain. No production report or provider behavior changed.
The isolated repaired case passes, followed by the complete 132-case voice
integration suite on a new empty database. The unchanged complete report (15),
cleanup (24), and retention (13) suites passed in the first combined run, each
on its separate empty database. These 184 distinct consumer cases plus one
complete schema/snapshot drift case yield 185 distinct passing cases for this
integration. The isolated case is not added again. Database build, server
build, final server typecheck, migration safety/numbering and module-boundary
checks pass. The earlier 303-case evidence belongs to the stated previous head;
it is not added to this acceptance count or claimed rerun here.
Logs: `/tmp/sleep25-landed-parent-db-generate.log`,
`/tmp/sleep25-landed-parent-db-build.log`,
`/tmp/sleep25-landed-parent-real-migration.log`,
`/tmp/sleep25-landed-parent-real-consumers.log` (the retained 183-pass/one-fail run),
`/tmp/sleep25-report-mock-isolation-red.log`,
`/tmp/sleep25-report-mock-isolation-green.log`,
`/tmp/sleep25-landed-parent-final-voice.log`,
`/tmp/sleep25-landed-parent-snapshot.log`,
`/tmp/sleep25-landed-parent-final-types.log`,
`/tmp/sleep25-landed-parent-server-build.log`, and
`/tmp/sleep25-landed-parent-module-boundaries.log`.
Fresh current-head CI and Greptile 5/5 remain required. Prior-head review scores
are stale after this normal merge. A stopped or paused Ponytail turn without a
verdict does not block under the user's authorization; an actual failing
verdict must be fixed. No new full local repository suite, live-provider test,
Cloud wake, host/database suspension, or cost-saving result is claimed.
## Review documentation and preserved remote timeout — October 28
Ponytail reviewed `c28de398` at 4/5 advisory, found no must-fix production defect,
and said to ship after correcting the design document's stale migration
reference. The design document now links the current 185-case acceptance and
migration `0397_known_madrox.sql`; the older 302/303-case paragraphs are marked
historical. Its two remaining suggestions are explicitly optional: one-shot
diagnostic `reconcile(limit)` still inventories the whole eligible window before
slicing, and some existing test helpers advance fake time during assertion
retries. They are not silently claimed fixed or treated as current idle polling.
CI run `40703490340`, server shard 8/12 job `122869422044`, failed one unchanged
approval-queue test: `resets consecutive failure backoff after a successful batch
with remaining debt` exceeded its existing 15-second timeout. Teardown then
reported a company foreign-key violation from activity-log work. The shard had
2,097 passed tests, one failed and one skipped. No report assertion failed in
that job, and the exact CI timing cause is not proven. The complete unchanged
28-case approval-queue suite passes on a new isolated PostgreSQL database. Its
test, worker, coordinator and transaction-signals source were checked byte-for-
byte against `c28de398` before that probe, which ran from the local child checkout.
No timeout, workflow, approval logic or production source was changed. The old
CI failure remains recorded and the new documentation head requires fresh CI;
no blind rerun of the failed old job or waiver is claimed.
Evidence: `/tmp/sleep25-c28-server8-job.log`,
`/tmp/sleep25-approval-timeout-local-probe.log` and its exit status 0. The 28 cases
are a separate investigation result, not added to the 185-case report acceptance.
Prior-head review scores are stale after this documentation commit.
doc/plans/2026-10-25-voice-drain-repair.md
+10
View File
@@ -60,3 +60,13 @@ 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.
## Report resume assertion
The report test had the same active-counter-only flush assumption. It now checks
idle snapshot stability while paused, advances the retained nextWakeAt on resume,
and verifies one fetch and the available durable report before testing rollback.
All 14 report cases pass in /tmp/sleep25-drain-tests.log. An intermediate new
assertion incorrectly expected state=completed; the schema uses status=available.
That test-authoring failure is retained in /tmp/sleep25-drain-invalid-assertion.log.
The scheduler repair and business deadlines were not changed to satisfy tests.
packages/db/src/client.ts
+1
View File
@@ -548,6 +548,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
"0393_open_grandmaster.sql",
"0395_tranquil_firestar.sql",
"0396_tan_paladin.sql",
"0397_known_madrox.sql",
"0384_funny_adam_warlock.sql",
packages/db/src/migrations/0397_known_madrox.sql
+1
View File
@@ -0,0 +1 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_voice_sessions_report_window_idx" ON "chat_voice_sessions" USING btree ("ended_at","id") WHERE "chat_voice_sessions"."state" in ('ended', 'failed', 'expired') and "chat_voice_sessions"."provider_session_id" is not null;
packages/db/src/migrations/meta/0392_snapshot.json → packages/db/src/migrations/meta/0397_snapshot.json
Generated
+201 -2
View File
@@ -1,6 +1,6 @@
{
"id": "6974dd25-b8e9-4143-b4b7-ee5a5c3ff54a",
"prevId": "e080fc1b-af8e-4012-8ce5-3b5065b65143",
"id": "c7947622-f079-44c9-a57a-5a11d1bec5e8",
"prevId": "be87a5d8-4164-4d95-89ae-b895701fa21b",
"version": "7",
"dialect": "postgresql",
"tables": {
@@ -1350,6 +1350,34 @@
"concurrently": false,
"method": "btree",
"with": {}
},
"activity_log_company_user_issue_actor_idx": {
"name": "activity_log_company_user_issue_actor_idx",
"columns": [
{
"expression": "company_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "actor_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "entity_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"activity_log\".\"actor_type\" = 'user' AND \"activity_log\".\"entity_type\" = 'issue'",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
@@ -14160,6 +14188,24 @@
"primaryKey": false,
"notNull": false
},
"provider_account_id": {
"name": "provider_account_id",
"type": "text",
"primaryKey": false,
"notNull": false
},
"cleanup_confirmed_at": {
"name": "cleanup_confirmed_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"cleanup_next_attempt_at": {
"name": "cleanup_next_attempt_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"expires_at": {
"name": "expires_at",
"type": "timestamp with time zone",
@@ -14182,6 +14228,22 @@
}
},
"indexes": {
"chat_voice_inbound_cleanup_idx": {
"name": "chat_voice_inbound_cleanup_idx",
"columns": [
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_voice_inbound_expiry_idx": {
"name": "chat_voice_inbound_expiry_idx",
"columns": [
@@ -14934,6 +14996,28 @@
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_voice_sessions_report_window_idx": {
"name": "chat_voice_sessions_report_window_idx",
"columns": [
{
"expression": "ended_at",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
@@ -41945,6 +42029,28 @@
}
},
"indexes": {
"join_requests_pending_retention_idx": {
"name": "join_requests_pending_retention_idx",
"columns": [
{
"expression": "updated_at",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"join_requests\".\"status\" = 'pending_approval'",
"concurrently": false,
"method": "btree",
"with": {}
},
"join_requests_invite_unique_idx": {
"name": "join_requests_invite_unique_idx",
"columns": [
@@ -42965,6 +43071,99 @@
"checkConstraints": {},
"isRLSEnabled": false
},
"public.native_runner_checkpoints": {
"name": "native_runner_checkpoints",
"schema": "",
"columns": {
"company_id": {
"name": "company_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"run_id": {
"name": "run_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"issue_id": {
"name": "issue_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"part": {
"name": "part",
"type": "text",
"primaryKey": false,
"notNull": true
},
"material": {
"name": "material",
"type": "jsonb",
"primaryKey": false,
"notNull": true
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"native_runner_checkpoints_headers_idx": {
"name": "native_runner_checkpoints_headers_idx",
"columns": [
{
"expression": "run_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"native_runner_checkpoints\".\"part\" = 'header'",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk": {
"name": "native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk",
"tableFrom": "native_runner_checkpoints",
"tableTo": "heartbeat_runs",
"columnsFrom": [
"company_id",
"issue_id",
"run_id"
],
"columnsTo": [
"company_id",
"native_issue_id",
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {
"native_runner_checkpoints_run_id_part_pk": {
"name": "native_runner_checkpoints_run_id_part_pk",
"columns": [
"run_id",
"part"
]
}
},
"uniqueConstraints": {},
"policies": {},
"checkConstraints": {},
"isRLSEnabled": false
},
"public.pipeline_automation_executions": {
"name": "pipeline_automation_executions",
"schema": "",
packages/db/src/migrations/meta/_journal.json
Generated
+7
View File
@@ -2766,6 +2766,13 @@
"when": 1793191988458,
"tag": "0396_tan_paladin",
"breakpoints": true
},
{
"idx": 397,
"version": "7",
"when": 1793195976732,
"tag": "0397_known_madrox",
"breakpoints": true
}
]
}
packages/db/src/schema/chat_voice_sessions.ts
+2
View File
@@ -36,6 +36,8 @@ export const chatVoiceSessions = pgTable("chat_voice_sessions", {
uniqueIndex("chat_voice_sessions_request_uq").on(t.companyId, t.callerId, t.idempotencyKey),
uniqueIndex("chat_voice_sessions_live_caller_task_uq").on(t.companyId, t.callerId, t.issueId).where(sql`${t.state} not in ('ended', 'failed', 'expired')`),
index("chat_voice_sessions_expiry_idx").on(t.state, t.expiresAt),
index("chat_voice_sessions_report_window_idx").on(t.endedAt, t.id)
.where(sql`${t.state} in ('ended', 'failed', 'expired') and ${t.providerSessionId} is not null`),
foreignKey({ columns: [t.companyId, t.endpointId], foreignColumns: [chatEndpoints.companyId, chatEndpoints.id], name: "chat_voice_sessions_endpoint_fk" }),
foreignKey({ columns: [t.companyId, t.conversationId], foreignColumns: [chatConversations.companyId, chatConversations.id], name: "chat_voice_sessions_conversation_fk" }),
foreignKey({ columns: [t.companyId, t.issueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_sessions_issue_fk" }),
server/src/__tests__/app-private-hostname-gate.test.ts
+4 -13
View File
@@ -25,7 +25,6 @@ describe("createChatReconciliationCoordinator", () => {
processFailedGitHubWebhookDeliveries,
processPendingDeliveries,
flushPublications,
reconcileVoiceSessions: async () => undefined,
projectRunMilestones: async () => 0,
processPendingSlackFileUploadReceipts: async () => undefined,
processPendingSlackSessionSyncs: async () => undefined,
@@ -74,11 +73,11 @@ describe("createChatReconciliationCoordinator", () => {
},
);
it("keeps receiver inventory out of periodic chat maintenance while preserving voice", () => {
it("keeps receiver and voice inventory out of periodic chat maintenance", () => {
const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals"));
expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()");
expect(maintenance).toContain("chatChannels.voice.reconcileMaintenance()");
expect(maintenance).not.toContain("chatChannels.voice.");
expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,");
});
@@ -88,7 +87,8 @@ describe("createChatReconciliationCoordinator", () => {
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).toContain("chatChannels.voice.registerReportWork(deliveryWork,");
expect(source).not.toContain("reconcileVoiceSessions");
expect(source).not.toContain("chatChannels.voice.reconcile()");
});
@@ -153,7 +153,6 @@ describe("createChatReconciliationCoordinator", () => {
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones,
reconcileVoiceSessions: async () => undefined,
processPendingDeliveries: async () => undefined,
flushPublications,
processPendingSlackFileUploadReceipts: async () => undefined,
@@ -216,7 +215,6 @@ describe("createChatReconciliationCoordinator", () => {
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones,
flushPublications,
reconcileVoiceSessions: async () => undefined,
processPendingDeliveries: async () => undefined,
processPendingSlackFileUploadReceipts: async () => undefined,
processPendingSlackSessionSyncs: async () => undefined,
@@ -256,7 +254,6 @@ describe("createChatReconciliationCoordinator", () => {
const slackStatusReleased = new Promise<void>((resolve) => {
releaseSlackStatus = resolve;
});
const reconcileVoiceSessions = vi.fn(async () => undefined);
const processPendingDeliveries = vi.fn(async () => deliveryReleased);
const flushPublications = vi.fn(async () => undefined);
const processPendingSlackFileUploadReceipts = vi.fn(async () => undefined);
@@ -266,7 +263,6 @@ describe("createChatReconciliationCoordinator", () => {
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones: async () => 0,
reconcileVoiceSessions,
processPendingDeliveries,
flushPublications,
processPendingSlackFileUploadReceipts,
@@ -276,7 +272,6 @@ describe("createChatReconciliationCoordinator", () => {
coordinator.reconcile();
await vi.waitFor(() => {
expect(reconcileVoiceSessions).toHaveBeenCalledTimes(1);
expect(processPendingDeliveries).toHaveBeenCalledTimes(1);
expect(flushPublications).toHaveBeenCalledTimes(1);
expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(1);
@@ -284,7 +279,6 @@ describe("createChatReconciliationCoordinator", () => {
});
coordinator.reconcile();
await vi.waitFor(() => {
expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2);
expect(flushPublications).toHaveBeenCalledTimes(2);
expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2);
});
@@ -311,14 +305,12 @@ describe("createChatReconciliationCoordinator", () => {
await receiptReleased;
if (rejects) throw lookupError;
});
const reconcileVoiceSessions = vi.fn(async () => undefined);
const processPendingDeliveries = vi.fn(async () => undefined);
const flushPublications = vi.fn(async () => undefined);
const processPendingSlackSessionSyncs = vi.fn(async () => undefined);
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones: async () => 0,
reconcileVoiceSessions,
processPendingDeliveries,
flushPublications,
processPendingSlackFileUploadReceipts,
@@ -336,7 +328,6 @@ describe("createChatReconciliationCoordinator", () => {
});
coordinator.reconcile();
await vi.waitFor(() => {
expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2);
expect(processPendingDeliveries).toHaveBeenCalledTimes(2);
expect(flushPublications).toHaveBeenCalledTimes(2);
expect(processPendingSlackSessionSyncs).toHaveBeenCalledTimes(2);
server/src/__tests__/voice-call-reports.test.ts
+309
View File
@@ -0,0 +1,309 @@
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, chatVoiceSessions, chatVoiceReports, 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 { voiceCallReports } from "../services/voice/voice-call-reports.js";
import type { SpekoCallReport } from "../services/voice/speko-call-report.js";
import { notifyVoiceReports } from "../services/voice/voice-report-notifications.js";
import { withVoiceSessionChange } from "../services/voice/voice-recovery-notifications.js";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_REPORT_TEST_DATABASE_URL");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("voice call-report lifecycle", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>, companyId: string, endpointId: string, agentId: string, issueId: string, conversationId: string;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let enabled = true;
const errors: unknown[] = [];
const fixtureCompanies: string[] = [];
const day = 86_400_000;
const result = (): SpekoCallReport => ({ complete: true, transcript: [], costMicroUsd: "100", durationSeconds: 10, providerUpdatedAt: new Date() });
const fetchReport = vi.fn(async (_session: typeof chatVoiceSessions.$inferSelect) => result());
beforeAll(async () => {
if (!external) temporary = await startEmbeddedPostgresTestDatabase("voice-call-reports-");
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: "Voice recovery", issuePrefix: `V${companyId.slice(0, 7)}` });
await db.insert(agents).values({ id: agentId, companyId, name: "Voice", adapterType: "paperclip_runner" });
await db.insert(issues).values({ id: issueId, companyId, title: "Voice", assigneeAgentId: agentId, status: "in_progress" });
await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Voice", type: "chat" });
await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Voice", uid: connectionId, transport: "voice", connectionPurpose: "channel", enabled: true, status: "active" });
await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, assignedAgentId: agentId, publicId: endpointId, provider: "speko", status: "active", setup: { step: "complete", runtimeGeneration: 1 } });
await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId, externalConversationId: conversationId, externalThreadId: conversationId, externalLabel: "Voice", isDirectMessage: true });
return { companyId, endpointId, assignedAgentId: agentId, issueId, conversationId };
}
beforeEach(async () => {
enabled = true; errors.length = 0; fetchReport.mockReset().mockImplementation(async () => result());
({ companyId, endpointId, assignedAgentId: agentId, issueId, conversationId } = await fixture());
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
});
afterEach(async () => {
await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks();
for (const companyId of fixtureCompanies.splice(0)) {
await db.delete(chatVoiceSessions).where(eq(chatVoiceSessions.companyId, companyId));
await db.delete(chatConversations).where(eq(chatConversations.companyId, companyId));
await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId));
await db.delete(issues).where(eq(issues.companyId, companyId));
await db.delete(agents).where(eq(agents.companyId, companyId));
await db.delete(companies).where(eq(companies.id, companyId));
}
expect(errors).toEqual([]);
});
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
async function session(patch: Partial<typeof chatVoiceSessions.$inferInsert> = {}, writer = db) {
const id = patch.id ?? randomUUID();
return withVoiceSessionChange(writer, id, tx => tx.insert(chatVoiceSessions).values({ id, companyId, endpointId, conversationId, issueId, assignedAgentId: agentId,
callerId: randomUUID(), callerAuthority: "member", mode: "browser", state: "ended", endedAt: new Date(), generation: 1, credentialFingerprint: "test", providerSessionId: randomUUID(),
toolTokenHash: "a".repeat(64), idempotencyKey: id, requestFingerprint: id, createdAt: new Date(Date.now() - 60_000), updatedAt: new Date(Date.now() - 60_000), expiresAt: new Date(Date.now() + 600_000), ...patch }).returning()).then(rows => rows[0]);
}
async function start(service = voiceCallReports(db, fetchReport)) {
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 = 10_000) {
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
const requests = fetchReport.mock.calls.length;
try { await vi.advanceTimersByTimeAsync(ms); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(fetchReport).toHaveBeenCalledTimes(requests); }
finally { reads.mockRestore(); writes.mockRestore(); }
}
async function report(id: string) { return (await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.sessionId, id)))[0]; }
async function due(id: string) {
await db.transaction(async tx => {
await notifyVoiceReports(tx, { kind: "session", id });
await tx.update(chatVoiceReports).set({ nextCheckAt: new Date(0) }).where(eq(chatVoiceReports.sessionId, id));
});
}
it("bounds report reconstruction and resets a partial scan when the same owner restarts", async () => {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false });
const deadline = Date.now() + 120_000;
const ids: string[] = [];
for (let index = 0; index < 105; index++) {
const call = await session({ endedAt: new Date(Date.now() - 60_000) }); ids.push(call.id);
await db.insert(chatVoiceReports).values({ sessionId: call.id, companyId, nextCheckAt: new Date(deadline) });
}
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 === chatVoiceSessions) {
const execute = query.execute.bind(query);
vi.spyOn(query, "execute").mockImplementation(async () => {
const rows = await execute(); counts.push(rows.length); return rows;
});
}
return query;
});
return selection;
});
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(); expect(fetchReport).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(deadline - Date.now());
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(1000);
expect(idleWorkSnapshot().active).toBe(0);
expect(fetchReport).toHaveBeenCalledTimes(105);
}, { timeout: 5000 });
expect(fetchReport.mock.calls.map(([call]) => call.id).sort()).toEqual(ids.sort());
expect((await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.companyId, companyId)))
.every(row => row.attempts === 1)).toBe(true);
await quiet();
} finally { reads.mockRestore(); }
});
it("leaves configured idle, live calls, old history and unidentified calls quiet", async () => {
await session({ state: "active", endedAt: null });
await session({ endedAt: new Date(Date.now() - day - 1000) });
await session({ providerSessionId: null });
await start(); await flush(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(120_000);
expect(fetchReport).not.toHaveBeenCalled();
});
it("wakes on committed termination and preserves refresh of available reports", async () => {
await start(); await flush(); await quiet();
const call = await session(); await flush();
expect(fetchReport).toHaveBeenCalledOnce(); expect((await report(call.id)).status).toBe("available");
const next = (await report(call.id)).nextCheckAt.getTime();
expect(coordinator!.nextWakeAt()).toBe(next); await quiet();
await vi.advanceTimersByTimeAsync(next - Date.now() + 1000); await flush();
expect(fetchReport).toHaveBeenCalledTimes(2); expect((await report(call.id)).attempts).toBe(2);
});
it("reconstructs persisted retries after restart without early provider reads", async () => {
const call = await session(); fetchReport.mockRejectedValueOnce(new Error("provider unavailable"));
await start(); await flush(); const saved = await report(call.id);
expect(saved.attempts).toBe(1); expect(saved.status).toBe("pending");
await coordinator!.stop(); await start(); await flush();
expect(coordinator!.nextWakeAt()).toBe(saved.nextCheckAt.getTime()); await quiet();
await vi.advanceTimersByTimeAsync(saved.nextCheckAt.getTime() - Date.now() + 1000); await flush();
expect(fetchReport).toHaveBeenCalledTimes(2); expect((await report(call.id)).status).toBe("available");
});
it("disarms at the end of the finite window even when the next check is later", async () => {
const endedAt = new Date(Date.now() - day + 25_000), call = await session({ endedAt });
await start(); await flush();
expect(coordinator!.nextWakeAt()).toBe(endedAt.getTime() + day + 1);
expect((await report(call.id)).nextCheckAt.getTime()).toBeGreaterThan(coordinator!.nextWakeAt()!);
await vi.advanceTimersByTimeAsync(30_000); await flush();
expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(120_000); expect(fetchReport).toHaveBeenCalledOnce();
await coordinator!.stop(); await start(); await flush(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet();
});
it("rounds the inclusive PostgreSQL window end up without another provider fetch", async () => {
const call = await session(), end = Math.trunc(Date.now()) + 60_000;
await db.execute(sql`UPDATE chat_voice_sessions SET ended_at = to_timestamp(${end - day}::double precision / 1000) + interval '0.0005 seconds' WHERE id = ${call.id}`);
await db.insert(chatVoiceReports).values({ companyId, sessionId: call.id, status: "available", nextCheckAt: new Date(end + 60_000) });
const service = voiceCallReports(db, fetchReport);
expect((await service.load())[0].dueAt).toBe(end + 1);
vi.setSystemTime(end + 1);
expect(await service.check(call.id)).toBeNull(); expect(fetchReport).not.toHaveBeenCalled();
});
it("marks repeated missing reports unavailable but never downgrades retained available data", async () => {
const available = await session(), missing = await session(), service = voiceCallReports(db, fetchReport);
await service.check(available.id);
await db.update(chatVoiceReports).set({ attempts: 7, nextCheckAt: new Date(0) }).where(eq(chatVoiceReports.sessionId, available.id));
await db.insert(chatVoiceReports).values({ companyId, sessionId: missing.id, attempts: 7, nextCheckAt: new Date(0) });
fetchReport.mockRejectedValue(new Error("provider unavailable"));
await service.check(available.id); await service.check(missing.id);
expect(await report(available.id)).toMatchObject({ attempts: 8, status: "available", costMicroUsd: "100" });
expect(await report(missing.id)).toMatchObject({ attempts: 8, status: "unavailable" });
expect((await report(missing.id)).nextCheckAt.getTime()).toBeGreaterThan(Date.now());
});
it("retains committed work during drain and ignores rolled-back nested writes", async () => {
await start(); await flush(); enabled = false;
const call = await session(); await flush();
const drained = idleWorkSnapshot();
await quiet(60_000); expect(idleWorkSnapshot()).toEqual(drained); expect(await report(call.id)).toBeUndefined();
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(fetchReport).toHaveBeenCalledOnce(); });
expect(await report(call.id)).toMatchObject({ status: "available" });
await expect(db.transaction(async tx => { await session({}, tx as unknown as typeof db); throw new Error("rollback"); })).rejects.toThrow("rollback");
await flush(); expect(fetchReport).toHaveBeenCalledOnce(); await quiet();
});
it("does not fetch an uncommitted nested termination", async () => {
await start(); await flush();
await db.transaction(async tx => {
await session({}, tx as unknown as typeof db);
await vi.advanceTimersByTimeAsync(5000); expect(fetchReport).not.toHaveBeenCalled();
});
await flush(); expect(fetchReport).toHaveBeenCalledOnce();
});
it("keeps a callback's earlier deadline when an older provider response finishes", async () => {
const call = await session(), service = voiceCallReports(db, fetchReport);
let release!: (value: SpekoCallReport) => void;
fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
const checking = service.check(call.id);
await vi.waitFor(() => expect(fetchReport).toHaveBeenCalledOnce());
await due(call.id);
release({ ...result(), costMicroUsd: "1" }); await checking;
expect((await report(call.id)).nextCheckAt.getTime()).toBe(0);
expect((await report(call.id)).costMicroUsd).toBeNull();
await service.check(call.id); expect((await report(call.id)).costMicroUsd).toBe("100");
});
it("does not regress an available report from incomplete or older responses", async () => {
const call = await session(), service = voiceCallReports(db, fetchReport);
const latest = new Date(Date.now() + 10000);
fetchReport.mockResolvedValueOnce({ ...result(), providerUpdatedAt: latest, costMicroUsd: "200" });
await service.check(call.id);
for (const response of [{ ...result(), complete: false, providerUpdatedAt: new Date(latest.getTime() + 1), costMicroUsd: "1" },
{ ...result(), providerUpdatedAt: new Date(0), costMicroUsd: "2" }]) {
await due(call.id); fetchReport.mockResolvedValueOnce(response); await service.check(call.id);
expect(await report(call.id)).toMatchObject({ status: "available", costMicroUsd: "200" });
}
});
it("allows only one concurrent due claim and saves its retry before provider contact", async () => {
const call = await session(), first = voiceCallReports(db, fetchReport), second = voiceCallReports(db, fetchReport);
let release!: (value: SpekoCallReport) => void;
fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
const checking = first.check(call.id);
await vi.waitFor(() => expect(fetchReport).toHaveBeenCalledOnce());
const saved = await report(call.id); expect(saved.attempts).toBe(1); expect(saved.nextCheckAt.getTime()).toBeGreaterThan(Date.now());
await second.check(call.id); expect(fetchReport).toHaveBeenCalledOnce();
release(result()); await checking;
});
it("pages and fairly schedules a backlog with one failing provider", async () => {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false });
const broken = await session(); for (let i = 0; i < 103; i++) await session();
const other = await fixture(); const last = await session({ ...other, id: "ffffffff-ffff-ffff-ffff-ffffffffffff" });
fetchReport.mockImplementation(async call => { if (call.id === broken.id) throw new Error("unavailable"); return result(); });
await start(); expect(fetchReport).toHaveBeenCalledTimes(25);
await vi.advanceTimersByTimeAsync(1000);
await vi.waitFor(() => expect(fetchReport).toHaveBeenCalledTimes(50), { interval: 0 });
expect(fetchReport.mock.calls.slice(25, 50).some(([call]) => call.id === last.id)).toBe(true);
for (let pass = 0; pass < 20; pass++) {
await vi.advanceTimersByTimeAsync(1000);
await vi.waitFor(() => expect(coordinator!.nextWakeAt() !== null || idleWorkSnapshot().active === 0).toBe(true), { interval: 0 });
}
expect(fetchReport).toHaveBeenCalledTimes(105);
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { interval: 0 });
expect((await report(broken.id)).attempts).toBe(1); await quiet();
});
it("reconstructs recent reports without scanning old history and preserves submillisecond page boundaries", async () => {
const seed = await session({ endedAt: new Date(Date.now() - 2 * day) });
await db.execute(sql`insert into chat_voice_sessions
(company_id, endpoint_id, conversation_id, issue_id, assigned_agent_id, caller_id,
caller_authority, mode, state, generation, credential_fingerprint, provider_session_id,
tool_token_hash, idempotency_key, request_fingerprint, expires_at, ended_at, created_at)
select company_id, endpoint_id, conversation_id, issue_id, assigned_agent_id, gen_random_uuid()::text,
caller_authority, mode, state, generation, credential_fingerprint, gen_random_uuid()::text,
tool_token_hash, gen_random_uuid()::text, request_fingerprint, expires_at, ended_at, created_at
from chat_voice_sessions cross join generate_series(1, 20000) where id = ${seed.id}`);
const stamp = new Date(Date.now() - 60_000).toISOString();
const expected: string[] = [];
for (let i = 0; i < 205; i++) {
const row = await session(); expected.push(row.id);
await db.execute(sql`update chat_voice_sessions set ended_at = ${stamp}::timestamptz + (${i % 3} * interval '0.0001 second') where id = ${row.id}`);
}
await db.execute(sql`analyze chat_voice_sessions`);
const queries: Array<{ text: string; parameters: unknown[] }> = [];
const original = db.$client.unsafe.bind(db.$client);
const capture = vi.spyOn(db.$client, "unsafe").mockImplementation(((text: string, parameters: unknown[] = [], ...rest: unknown[]) => {
if (text.includes('from "chat_voice_sessions"') && text.includes('left join "chat_voice_reports"')) queries.push({ text, parameters });
return original(text, parameters as never, ...rest as []);
}) as typeof db.$client.unsafe);
let loaded: Awaited<ReturnType<ReturnType<typeof voiceCallReports>["load"]>>;
try { loaded = await voiceCallReports(db, fetchReport).load(); }
finally { capture.mockRestore(); }
expect(loaded!.map(row => row.id).sort()).toEqual(expected.sort());
expect(queries).toHaveLength(3);
function removed(node: Record<string, unknown>): number {
return Math.max(Number(node["Rows Removed by Filter"] ?? 0),
...(Array.isArray(node.Plans) ? node.Plans.map(child => removed(child)) : []));
}
for (const query of queries) {
const [row] = await db.$client.unsafe(`explain (analyze, buffers, format json) ${query.text}`, query.parameters as never);
expect(removed(row["QUERY PLAN"][0].Plan)).toBeLessThan(1000);
}
expect(fetchReport).not.toHaveBeenCalled();
}, 30_000);
it("joins an in-flight report at shutdown and resumes its saved retry", async () => {
const call = await session(); let release!: (value: SpekoCallReport) => void;
fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
const worker = voiceCallReports(db, fetchReport).register(coordinator, () => true);
await vi.waitFor(() => expect(fetchReport).toHaveBeenCalledOnce());
let stopped = false; const stopping = coordinator.stop().then(() => { stopped = true; });
await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false);
release(result()); await stopping; await worker.ready;
expect((await report(call.id)).status).toBe("available");
await start(); await flush(); await quiet(); expect(fetchReport).toHaveBeenCalledOnce();
});
});
server/src/__tests__/voice-session-store.integration.test.ts
+130 -4
View File
@@ -82,6 +82,126 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication };
}
it("keeps saved report deadlines quiet across repeated voice connection saves", async () => {
const f = await nativeServiceFixture();
const [seed] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, f.sessionId));
const endedAt = new Date(), deadline = Date.now() + 600_000;
await db.update(chatVoiceSessions).set({ state: "ended", endedAt }).where(eq(chatVoiceSessions.id, seed.id));
const ids = [seed.id, ...Array.from({ length: 204 }, () => randomUUID())];
await db.insert(chatVoiceSessions).values(ids.slice(1).map(id => ({
...seed, id, callerId: randomUUID(), idempotencyKey: id, providerSessionId: randomUUID(),
state: "ended" as const, endedAt,
})));
await db.insert(chatVoiceReports).values(ids.map(sessionId => ({ companyId: f.companyId, sessionId, nextCheckAt: new Date(deadline) })));
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);
expect(coordinator.nextWakeAt()).toBe(deadline);
}, { timeout: 5000 });
let selected = 0; const select = db.select.bind(db);
try {
await f.service.registerReportWork(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 === chatVoiceSessions) {
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 report ${index}` });
await settle();
}
expect(selected).toBe(0); expect(f.transport.callReport).not.toHaveBeenCalled();
expect(onError).not.toHaveBeenCalled();
} finally { reads.mockRestore(); }
} finally {
await coordinator.stop(); vi.useRealTimers(); vi.restoreAllMocks();
await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.companyId, f.companyId));
}
});
it.each(["hangup", "signed callback", "reply ended"])("wakes report refresh from real %s and advances it on a later callback", async trigger => {
const f = await inboundFixture();
await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.id, f.sessionId));
await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller,
{ approve: true, approvalCode: f.call.approvalCode, issueId: f.issueId });
const [bound] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
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 () => {
const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction");
try { await vi.advanceTimersByTimeAsync(10_000); expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); }
finally { reads.mockRestore(); transactions.mockRestore(); }
};
try {
await f.service.registerReportWork(coordinator, () => true).ready;
await settle(); expect(coordinator.nextWakeAt()).toBeNull(); await quiet();
if (trigger === "hangup") await f.service.end(f.companyId, bound.sessionId!, f.input.caller);
else if (trigger === "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 {
const publication = await f.publication(); await f.store.enqueuePublication(f.companyId, publication.id);
f.transport.sendCallMessage.mockRejectedValueOnce(new SpekoProviderError("provider_unavailable", false, 409));
await f.service.pushReplies(25, f.companyId, f.endpointId);
expect(f.transport.sendCallMessage).toHaveBeenCalledOnce();
}
await settle(); expect(f.transport.callReport).toHaveBeenCalledOnce();
const [saved] = await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.sessionId, bound.sessionId!));
expect(saved.status).toBe("available"); expect(coordinator.nextWakeAt()).toBe(saved.nextCheckAt.getTime());
await quiet(); expect(f.transport.callReport).toHaveBeenCalledOnce();
const report = f.sign({ type: "call.report", session_id: f.call.providerSessionId, organization_id: f.event.organization_id });
await f.service.inbound.lifecycle(f.endpoint.publicId, report.body, report.headers);
await settle(); expect(f.transport.callReport).toHaveBeenCalledTimes(2); expect(onError).not.toHaveBeenCalled();
} finally {
await coordinator.stop(); vi.useRealTimers();
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(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.companyId, f.companyId));
}
});
it("retains a saved report and retries guest transcript projection after a document lock", async () => {
const f = await inboundFixture(true);
await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.id, f.sessionId));
await f.tool("submit_request", { text: "Draft a public inquiry" });
const docs = documentService(db), key = `voice-transcript-${f.call.sessionId}`;
await docs.upsertIssueDocument({ issueId: f.call.intakeIssueId!, key, format: "markdown", body: "Older transcript" });
await docs.lockIssueDocument({ issueId: f.call.intakeIssueId!, key, lockedByUserId: f.callerId });
const ended = f.sign({ type: "call.report", session_id: f.call.providerSessionId, organization_id: f.event.organization_id });
await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers);
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
const settle = () => vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 });
try {
await f.service.registerReportWork(coordinator, () => true).ready; await settle();
const [saved] = await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.sessionId, f.call.sessionId!));
expect(saved.status).toBe("available"); expect(saved.transcript).toHaveLength(1);
expect((await docs.getIssueDocumentByKey(f.call.intakeIssueId!, key))!.body).toBe("Older transcript");
await docs.unlockIssueDocument(f.call.intakeIssueId!, key);
await vi.advanceTimersByTimeAsync(saved.nextCheckAt.getTime() - Date.now() + 1000); await settle();
const projected = await docs.getIssueDocumentByKey(f.call.intakeIssueId!, key);
expect(projected!.body).toContain("A test request");
expect(projected!.sourceTrust).toMatchObject({ preset: "low_trust_review", disposition: "quarantined" });
expect(f.transport.callReport).toHaveBeenCalledTimes(2); expect(onError).not.toHaveBeenCalled();
} finally {
await coordinator.stop(); vi.useRealTimers();
await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 2 * 86_400_000) }).where(eq(chatVoiceSessions.companyId, f.companyId));
}
});
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" });
@@ -150,7 +270,6 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
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,
});
@@ -1766,12 +1885,19 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
await db.update(chatVoiceReports).set({transcript: [...detail.report.transcript, {...detail.report.transcript[0]!, id: "legacy-control", index: 1, text: VOICE_RESULT_NOTIFICATION}]}).where(eq(chatVoiceReports.sessionId, first.session.id));
expect((await f.service.callDetail(f.companyId, first.session.id, f.input.caller)).report.transcript).toEqual(detail.report.transcript);
const other = await f.service.start({ ...f.input, idempotencyKey: randomUUID() });
await f.service.end(f.companyId, other.session.id, f.input.caller);
await db.update(chatVoiceSessions).set({ endedAt: new Date(Date.now() - 60_000) }).where(eq(chatVoiceSessions.id, other.session.id));
await db.update(chatVoiceReports).set({ nextCheckAt: new Date(0) }).where(eq(chatVoiceReports.sessionId, first.session.id));
const latest = new Date(Date.now() + 10000);
f.transport.callReport.mockResolvedValueOnce({ ...await f.transport.callReport(), costMicroUsd: "240000", providerUpdatedAt: latest });
const [stored] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, first.session.id));
const baseline = await f.transport.callReport(), latest = new Date(Date.now() + 10000);
let response = { ...baseline, costMicroUsd: "240000", providerUpdatedAt: latest };
f.transport.callReport.mockImplementation(async (providerSessionId?: string) => providerSessionId === stored.providerSessionId ? response : baseline);
await f.service.reconcile(1000);
expect((await f.service.callDetail(f.companyId, first.session.id, f.input.caller)).report.costMicroUsd).toBe("240000");
expect((await f.service.callDetail(f.companyId, other.session.id, f.input.caller)).report.costMicroUsd).toBe("120000");
await db.update(chatVoiceReports).set({ nextCheckAt: new Date(0) }).where(eq(chatVoiceReports.sessionId, first.session.id));
f.transport.callReport.mockResolvedValueOnce({ ...await f.transport.callReport(), costMicroUsd: "1", providerUpdatedAt: new Date(0) });
response = { ...baseline, costMicroUsd: "1", providerUpdatedAt: new Date(0) };
await f.service.reconcile(1000);
expect((await f.service.callDetail(f.companyId, first.session.id, f.input.caller)).report.costMicroUsd).toBe("240000");
expect(await db.select().from(chatVoiceReports).where(eq(chatVoiceReports.sessionId, first.session.id))).toHaveLength(1);
server/src/app.ts
+1 -4
View File
@@ -321,7 +321,6 @@ export function shouldEnablePrivateHostnameGuard(opts: {
}
type ChatReconciliationLane =
| "voice sessions"
| "deliveries"
| "GitHub webhook recovery"
| "run milestones"
@@ -335,7 +334,6 @@ type ChatReconciliationLane =
* next publication sweep for every endpoint.
*/
export function createChatReconciliationCoordinator(input: {
reconcileVoiceSessions: () => Promise<unknown>;
processPendingDeliveries: () => Promise<unknown>;
processFailedGitHubWebhookDeliveries?: () => Promise<unknown>;
projectRunMilestones: () => Promise<number>;
@@ -377,7 +375,6 @@ export function createChatReconciliationCoordinator(input: {
return {
reconcile() {
if (stopped) return;
start("voice sessions", input.reconcileVoiceSessions);
start("deliveries", input.processPendingDeliveries);
if (input.processFailedGitHubWebhookDeliveries) {
start(
@@ -1298,6 +1295,7 @@ export async function createApp(
}
registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
chatChannels.voice.registerReportWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.voice.registerRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.voice.registerReplyWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
@@ -1309,7 +1307,6 @@ export async function createApp(
await chatChannels.processPublicationMaintenance();
};
const chatReconciliation = createChatReconciliationCoordinator({
reconcileVoiceSessions: () => chatChannels.voice.reconcileMaintenance(),
// Chat delivery, action, and Telegram discovery workers now own these
// duties. Keep the coordinator's other maintenance lanes independent.
processPendingDeliveries: async () => undefined,
server/src/services/delivery-work-notifications.ts
+1
View File
@@ -45,6 +45,7 @@ export const DELIVERY_QUEUES = {
chatTelegramMaintenance: "chat-telegram-maintenance",
chatReceivers: "chat-receivers",
voiceSessionRecovery: "voice-session-recovery",
voiceReports: "voice-call-reports",
voiceInboundRetention: "voice-inbound-retention",
voiceInboundCleanup: "voice-inbound-cleanup",
voiceReplies: "voice-reply-delivery",
server/src/services/voice/voice-call-reports.ts
+146
View File
@@ -0,0 +1,146 @@
import { and, eq, gte, inArray, isNotNull, lte, ne, or, sql } from "drizzle-orm";
import { activityLog, chatEndpoints, chatVoiceReports, chatVoiceSessions, issues, type Db } from "@paperclipai/db";
import { logger } from "../../middleware/logger.js";
import { documentService } from "../documents.js";
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
import { voiceRecoveryScope, type VoiceRecoveryScope } from "./voice-recovery-notifications.js";
import { notifyVoiceReports } from "./voice-report-notifications.js";
import type { VoiceSessionRow } from "./voice-session-store.js";
import type { SpekoCallReport } from "./speko-call-report.js";
import { SpekoProviderError } from "./speko-provider.js";
type Entry = { id: string; companyId: string; endpointId: string; connectionId: string; dueAt: number };
type Report = SpekoCallReport;
const WINDOW_MS = 24 * 60 * 60 * 1000;
const eligible = (now: number) => and(inArray(chatVoiceSessions.state, ["ended", "failed", "expired"]),
isNotNull(chatVoiceSessions.providerSessionId), gte(chatVoiceSessions.endedAt, new Date(now - WINDOW_MS)));
/** Reports may change for 24 hours after a call. Available is a presentation
* state, not a promise of immutable provider data. Keep that finite obligation. */
export function voiceCallReports(db: Db, fetchReport: (session: VoiceSessionRow) => Promise<Report>) {
type Cursor = { id: string; endedAt: string };
const cursors = new Map<string, Cursor>();
async function page(scope?: VoiceRecoveryScope, cursor?: Cursor, signal?: AbortSignal) {
signal?.throwIfAborted();
const rows = await db.select({ id: chatVoiceSessions.id, companyId: chatVoiceSessions.companyId,
endpointId: chatVoiceSessions.endpointId, connectionId: chatEndpoints.connectionId,
endedAt: sql<string>`${chatVoiceSessions.endedAt}::text`,
dueAt: sql<string>`least(coalesce(ceil(extract(epoch from ${chatVoiceReports.nextCheckAt}) * 1000), 0), floor(extract(epoch from ${chatVoiceSessions.endedAt}) * 1000) + ${WINDOW_MS} + 1)`,
}).from(chatVoiceSessions)
.innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceSessions.endpointId), eq(chatEndpoints.companyId, chatVoiceSessions.companyId)))
.leftJoin(chatVoiceReports, and(eq(chatVoiceReports.sessionId, chatVoiceSessions.id), eq(chatVoiceReports.companyId, chatVoiceSessions.companyId)))
.where(and(eligible(Date.now()),
scope?.kind === "session" ? eq(chatVoiceSessions.id, scope.id) : undefined,
scope?.kind === "endpoint" ? eq(chatVoiceSessions.endpointId, scope.id) : undefined,
scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined,
scope?.kind === "company" ? eq(chatVoiceSessions.companyId, scope.id) : undefined,
cursor ? sql`(${chatVoiceSessions.endedAt}, ${chatVoiceSessions.id}) > (${cursor.endedAt}::timestamptz, ${cursor.id}::uuid)` : undefined))
.orderBy(chatVoiceSessions.endedAt, chatVoiceSessions.id).limit(100);
signal?.throwIfAborted();
return rows;
}
function entries(rows: Awaited<ReturnType<typeof page>>): Entry[] {
return rows.map(({ endedAt: _cursor, ...row }) => ({ ...row, dueAt: Number(row.dueAt) }));
}
async function load(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise<Entry[]> {
const result: Entry[] = []; let cursor: Cursor | undefined;
for (;;) {
const rows = await page(scope, cursor, signal);
result.push(...entries(rows));
if (rows.length < 100) return result;
cursor = rows[rows.length - 1];
}
}
async function refresh(scope?: VoiceRecoveryScope, signal?: AbortSignal) {
const key = scope ? `${scope.kind}:${scope.id}` : "*";
const rows = await page(scope, cursors.get(key), signal), result = entries(rows), complete = rows.length < 100;
signal?.throwIfAborted();
if (complete) cursors.delete(key); else cursors.set(key, rows[rows.length - 1]);
return { rows: result, complete };
}
async function projectTranscript(session: VoiceSessionRow) {
// Keep both actual spoken sides on this public call's own task. This
// does not publish a document or grant access to any other task.
if (session.callerAuthority !== "guest_intake") return;
await db.transaction(async tx => {
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`voice-transcript:${session.id}`}, 0))`);
const [retained] = await tx.select().from(chatVoiceReports).where(and(eq(chatVoiceReports.companyId, session.companyId), eq(chatVoiceReports.sessionId, session.id)));
if (!retained?.transcript.length) return;
const [task] = await tx.select({sourceTrust: issues.sourceTrust}).from(issues).where(and(eq(issues.companyId, session.companyId), eq(issues.id, session.issueId)));
if (task?.sourceTrust?.preset !== "low_trust_review" || task.sourceTrust.disposition !== "quarantined") return;
const key = `voice-transcript-${session.id}`;
const body = retained.transcript.map(turn => `**${turn.speaker === "caller" ? "Caller" : "Agent"}** · ${turn.startedAt}${turn.interrupted ? " · interrupted" : ""}\n\n${turn.text}`).join("\n\n---\n\n");
const docs = documentService(tx as unknown as Db), previous = await docs.getIssueDocumentByKey(session.issueId, key);
if (previous?.body === body) return;
await docs.upsertIssueDocument({issueId: session.issueId, key, baseRevisionId: previous?.latestRevisionId, title: "Phone conversation transcript", format: "markdown", body, sourceTrust: task.sourceTrust, changeSummary: "Speko spoken conversation, including caller and agent turns"});
await tx.insert(activityLog).values({companyId: session.companyId, actorType: "system", actorId: "speko-report", action: "voice.transcript.updated", entityType: "issue", entityId: session.issueId, details: {sessionId: session.id}});
});
}
async function check(id: string, signal?: AbortSignal): Promise<Entry | null> {
signal?.throwIfAborted();
const claim = await db.transaction(async tx => {
const [session] = await tx.select().from(chatVoiceSessions).where(and(eq(chatVoiceSessions.id, id), eligible(Date.now()))).for("update");
if (!session) return null;
await notifyVoiceReports(tx, { kind: "session", id });
await tx.insert(chatVoiceReports).values({ sessionId: id, companyId: session.companyId, nextCheckAt: new Date() }).onConflictDoNothing();
const [report] = await tx.select().from(chatVoiceReports).where(and(eq(chatVoiceReports.sessionId, id),
eq(chatVoiceReports.companyId, session.companyId), lte(chatVoiceReports.nextCheckAt, new Date()))).for("update");
if (!report) return null;
signal?.throwIfAborted();
const attempts = report.attempts + 1;
const retryAt = new Date(Date.now() + Math.min(3600, 30 * 2 ** Math.min(attempts, 7)) * 1000);
await tx.update(chatVoiceReports).set({ attempts, nextCheckAt: retryAt }).where(eq(chatVoiceReports.sessionId, id));
return { session, attempts, retryAt };
});
if (claim) {
const { session, attempts, retryAt } = claim;
// A callback can advance nextCheckAt while this GET is in flight.
// That callback or a newer attempt owns settlement; never overwrite it.
const owned = () => and(eq(chatVoiceReports.sessionId, id), eq(chatVoiceReports.companyId, session.companyId),
eq(chatVoiceReports.attempts, attempts), eq(chatVoiceReports.nextCheckAt, retryAt));
signal?.throwIfAborted();
try {
const result = await fetchReport(session);
logger.info({ event: "voice.call.report_received", companyId: session.companyId, endpointId: session.endpointId,
issueId: session.issueId, sessionId: id, providerSessionId: session.providerSessionId, attempts,
complete: result.complete, transcriptTurns: result.transcript.length, providerUpdatedAt: result.providerUpdatedAt?.toISOString() ?? null }, "Voice call report received for reconciliation");
await db.transaction(async tx => {
await notifyVoiceReports(tx, { kind: "session", id });
await tx.update(chatVoiceReports).set({ transcript: result.transcript, costMicroUsd: result.costMicroUsd,
durationSeconds: result.durationSeconds, providerUpdatedAt: result.providerUpdatedAt,
status: result.complete ? "available" : "pending", updatedAt: new Date(),
}).where(and(owned(), result.complete ? undefined : ne(chatVoiceReports.status, "available"),
result.providerUpdatedAt ? or(sql`${chatVoiceReports.providerUpdatedAt} is null`, lte(chatVoiceReports.providerUpdatedAt, result.providerUpdatedAt)) : sql`${chatVoiceReports.providerUpdatedAt} is null`));
});
// Project the retained authoritative report, including when a response
// was stale. A projection failure retains the already saved retry.
await projectTranscript(session);
} catch (error) {
logger.warn({ event: "voice.call.report_failed", companyId: session.companyId, endpointId: session.endpointId,
issueId: session.issueId, sessionId: id, providerSessionId: session.providerSessionId, attempts,
errorCode: error instanceof SpekoProviderError ? error.code : "report_reconciliation_failed",
httpStatus: error instanceof SpekoProviderError ? error.httpStatus ?? null : null, retryAt: retryAt.toISOString() }, "Voice call report reconciliation deferred");
if (attempts >= 8) await db.transaction(async tx => {
await notifyVoiceReports(tx, { kind: "session", id });
await tx.update(chatVoiceReports).set({ status: "unavailable" }).where(and(owned(), ne(chatVoiceReports.status, "available")));
});
}
}
return (await load({ kind: "session", id }, signal))[0] ?? null;
}
async function reconcile(limit: number) {
const rows = (await 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 check(row.id);
return rows.length;
}
function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
cursors.clear();
return registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.voiceReports, decodeScope: voiceRecoveryScope,
matches: (entry, scope) => !scope || (scope.kind === "session" ? 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, reconcile, register };
}
server/src/services/voice/voice-recovery-notifications.ts
+2
View File
@@ -1,3 +1,4 @@
import { notifyVoiceReports } from "./voice-report-notifications.js";
import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
import { notifyVoiceReplies } from "./voice-reply-notifications.js";
import type { Db } from "@paperclipai/db";
@@ -20,6 +21,7 @@ export async function notifyVoiceChange(tx: object, scope?: VoiceRecoveryScope)
await notifyVoiceRecovery(tx, scope);
await notifyVoiceReplies(tx, scope);
await notifyVoiceInbound(tx, scope);
if (scope?.kind === "session") await notifyVoiceReports(tx, { kind: "session", id: scope.id });
}
export function withVoiceSessionChange<T>(db: Db, id: string, write: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => PromiseLike<T>, options: { replyDelivery?: boolean } = {}): Promise<T> {
return db.transaction(async tx => {
server/src/services/voice/voice-reply-push.ts
+5 -1
View File
@@ -1,3 +1,4 @@
import { notifyVoiceReports } from "./voice-report-notifications.js";
import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
import { notifyVoiceReplies } from "./voice-reply-notifications.js";
import { recoverVoiceReplyDispatches } from "./voice-reply-dispatch-recovery.js";
@@ -84,7 +85,10 @@ export async function pushVoiceReplies(db: Db, options: {
await notifyVoiceReplies(tx, { kind: "session", id: candidate.id });
await tx.update(chatActions).set({status, result: {errorCode: providerError?.code ?? "voice_authority_unavailable", ...(status === "retry" ? {retryAt: Date.now() + 10_000} : {})}, updatedAt: new Date()}).where(eq(chatActions.id, claim.action.id));
if (status === "call_ended" || status === "unknown") await notifyVoiceRecovery(tx, { kind: "session", id: candidate.id });
if (status === "call_ended") await notifyVoiceInbound(tx, { kind: "session", id: candidate.id });
if (status === "call_ended") {
await notifyVoiceInbound(tx, { kind: "session", id: candidate.id });
await notifyVoiceReports(tx, { kind: "session", id: candidate.id });
}
if (status === "call_ended") await tx.update(chatVoiceSessions).set({state: "ended", endedAt: new Date(), updatedAt: new Date()}).where(and(eq(chatVoiceSessions.id, candidate.id), eq(chatVoiceSessions.companyId, candidate.companyId)));
if (status === "unknown") await tx.update(chatVoiceSessions).set({errorCode: "message_delivery_unknown", updatedAt: new Date()}).where(eq(chatVoiceSessions.id, candidate.id));
});
server/src/services/voice/voice-report-notifications.ts
+5
View File
@@ -0,0 +1,5 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
export function notifyVoiceReports(tx: object, scope: { kind: "session"; id: string }) {
return notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceReports, `${scope.kind}:${scope.id}`);
}
server/src/services/voice/voice-session-service.ts
+15 -66
View File
@@ -5,17 +5,17 @@ import { registerVoiceRecoveryWork } from "./voice-recovery-work.js";
import { notifyVoiceChange, withVoiceSessionChange } from "./voice-recovery-notifications.js";
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
import { logger } from "../../middleware/logger.js";
import { voiceCallReports } from "./voice-call-reports.js";
import { captureVoiceDeliveryDiagnostics } from "./voice-delivery-diagnostics.js";
import { pushVoiceReplies } from "./voice-reply-push.js";
import { voiceInboundService } from "./voice-inbound-service.js";
import { createHash, randomUUID } from "node:crypto";
import { and, asc, desc, eq, inArray, lte, notInArray, or, sql } from "drizzle-orm";
import { chatVoiceReports, chatVoiceCallbacks, chatConversations, chatEndpointResources, chatEndpoints, chatVoiceSessions, issues, toolConnections, activityLog, type Db } from "@paperclipai/db";
import { and, desc, eq, inArray, notInArray, or, sql } from "drizzle-orm";
import { chatVoiceReports, chatVoiceCallbacks, chatConversations, chatEndpointResources, chatEndpoints, chatVoiceSessions, issues, activityLog, type Db } from "@paperclipai/db";
import { VOICE_RESULT_NOTIFICATION, VOICE_REPEAT_NOTIFICATION, type VoiceSession, type VoiceSessionMedia, type VoiceCallHistoryEntry, type VoiceCallReport } from "@paperclipai/shared";
import { conflict, forbidden, notFound, HttpError } from "../../errors.js";
import { instanceSettingsService } from "../instance-settings.js";
import { buildChatCommunicationGuidance } from "../chat-communication-guidance.js";
import { documentService } from "../documents.js";
import { issueService } from "../issues.js";
import { createSpekoProvider, SpekoProviderError } from "./speko-provider.js";
import { serializeVoiceSession, voiceSessionStore, type VoiceCaller, type VoiceSessionRow } from "./voice-session-store.js";
@@ -229,8 +229,16 @@ export function voiceSessionService(db: Db, options: {
function registerReplyWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
return registerVoiceReplyWork(coordinator, replies, canRun);
}
async function reconcileMaintenance(limit = 25) {
await reconcileReports(limit);
const reports = voiceCallReports(db, async session => {
const endpoint = await endpointFor(session.companyId, session.endpointId);
const credentials = await options.credentials(endpoint, session.credentialFingerprint);
const client = provider(credentials.apiKey);
const result = await client.callReport(session.providerSessionId!, [credentials.signingSecret]);
await captureVoiceDeliveryDiagnostics(db, session, client, result.complete);
return result;
});
function registerReportWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
return reports.register(coordinator, canRun);
}
/** Explicit one-shot recovery for diagnostics/tests; never register on a recurring lane. */
async function reconcile(limit = 25) {
@@ -238,7 +246,7 @@ export function voiceSessionService(db: Db, options: {
for (const row of due) await recovery.check(row.id).catch(() => undefined);
await pushReplies(limit);
await inbound.reconcile(limit);
await reconcileReports(limit);
await reports.reconcile(limit);
return due.length;
}
async function pushReplies(limit = 25, companyId?: string, endpointId?: string) {
@@ -270,65 +278,6 @@ export function voiceSessionService(db: Db, options: {
const [report] = await db.select().from(chatVoiceReports).where(and(eq(chatVoiceReports.companyId, companyId), eq(chatVoiceReports.sessionId, session.id)));
return { session: serializeVoiceSession(session), report: reportProjection(report) };
}
async function reconcileReports(limit: number) {
// Repair missed lifecycle callbacks, including later recording-ready reports.
// GET the same call; never create a call or download provider recordings.
const recent = new Date(Date.now() - 24 * 60 * 60 * 1000);
const rows = await db.select({ session: chatVoiceSessions, report: chatVoiceReports }).from(chatVoiceSessions)
.leftJoin(chatVoiceReports, and(eq(chatVoiceReports.sessionId, chatVoiceSessions.id), eq(chatVoiceReports.companyId, chatVoiceSessions.companyId)))
.where(and(inArray(chatVoiceSessions.state, terminalStates), sql`${chatVoiceSessions.providerSessionId} is not null`,
sql`${chatVoiceSessions.endedAt} >= ${recent.toISOString()}`, or(sql`${chatVoiceReports.sessionId} is null`, lte(chatVoiceReports.nextCheckAt, new Date()))))
.orderBy(asc(sql`coalesce(${chatVoiceReports.nextCheckAt}, ${chatVoiceSessions.endedAt})`)).limit(limit);
for (const {session, report} of rows) {
const attempts = (report?.attempts ?? 0) + 1;
const retryAt = new Date(Date.now() + Math.min(3600, 30 * 2 ** Math.min(attempts, 7)) * 1000);
await db.insert(chatVoiceReports).values({ sessionId: session.id, companyId: session.companyId }).onConflictDoNothing();
try {
const endpoint = await endpointFor(session.companyId, session.endpointId);
const credentials = await options.credentials(endpoint, session.credentialFingerprint);
const client = provider(credentials.apiKey);
const result = await client.callReport(session.providerSessionId!, [credentials.signingSecret]);
logger.info({event: "voice.call.report_received", companyId: session.companyId, endpointId: session.endpointId,
issueId: session.issueId, sessionId: session.id, providerSessionId: session.providerSessionId, attempts,
complete: result.complete, transcriptTurns: result.transcript.length, providerUpdatedAt: result.providerUpdatedAt?.toISOString() ?? null}, "Speko call report received for reconciliation");
await captureVoiceDeliveryDiagnostics(db, session, client, result.complete);
if (!result.complete && report?.status === "available") {
await db.update(chatVoiceReports).set({ attempts, nextCheckAt: retryAt }).where(eq(chatVoiceReports.sessionId, session.id));
continue;
}
await db.update(chatVoiceReports).set({
transcript: result.transcript, costMicroUsd: result.costMicroUsd, durationSeconds: result.durationSeconds,
providerUpdatedAt: result.providerUpdatedAt, status: result.complete ? "available" : "pending", attempts, nextCheckAt: retryAt, updatedAt: new Date(),
}).where(and(eq(chatVoiceReports.sessionId, session.id), eq(chatVoiceReports.companyId, session.companyId),
// A stale report must never replace a newer completed report.
result.providerUpdatedAt ? or(sql`${chatVoiceReports.providerUpdatedAt} is null`, lte(chatVoiceReports.providerUpdatedAt, result.providerUpdatedAt)) : sql`${chatVoiceReports.providerUpdatedAt} is null`));
// Keep both actual spoken sides on this public call's own task. This
// does not publish a document or grant access to any other task.
if (session.callerAuthority === "guest_intake") await db.transaction(async tx => {
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`voice-transcript:${session.id}`}, 0))`);
const [retained] = await tx.select().from(chatVoiceReports).where(and(eq(chatVoiceReports.companyId, session.companyId), eq(chatVoiceReports.sessionId, session.id)));
if (!retained?.transcript.length) return;
const [task] = await tx.select({sourceTrust: issues.sourceTrust}).from(issues).where(and(eq(issues.companyId, session.companyId), eq(issues.id, session.issueId)));
if (task?.sourceTrust?.preset !== "low_trust_review" || task.sourceTrust.disposition !== "quarantined") return;
const key = `voice-transcript-${session.id}`;
const body = retained.transcript.map(turn => `**${turn.speaker === "caller" ? "Caller" : "Agent"}** · ${turn.startedAt}${turn.interrupted ? " · interrupted" : ""}\n\n${turn.text}`).join("\n\n---\n\n");
const docs = documentService(tx as unknown as Db), previous = await docs.getIssueDocumentByKey(session.issueId, key);
if (previous?.body === body) return;
await docs.upsertIssueDocument({issueId: session.issueId, key, title: "Phone conversation transcript", format: "markdown", body, sourceTrust: task.sourceTrust, changeSummary: "Speko spoken conversation, including caller and agent turns"});
await tx.insert(activityLog).values({companyId: session.companyId, actorType: "system", actorId: "speko-report", action: "voice.transcript.updated", entityType: "issue", entityId: session.issueId, details: {sessionId: session.id}});
});
// Even a stale provider reply must back off instead of spinning.
await db.update(chatVoiceReports).set({ attempts, nextCheckAt: retryAt }).where(eq(chatVoiceReports.sessionId, session.id));
} catch (error) {
logger.warn({event: "voice.call.report_failed", companyId: session.companyId, endpointId: session.endpointId,
issueId: session.issueId, sessionId: session.id, providerSessionId: session.providerSessionId, attempts,
errorCode: error instanceof SpekoProviderError ? error.code : "report_reconciliation_failed",
httpStatus: error instanceof SpekoProviderError ? error.httpStatus ?? null : null, retryAt: retryAt.toISOString()}, "Speko call report reconciliation deferred");
await db.update(chatVoiceReports).set({ attempts, nextCheckAt: retryAt, ...(report?.status !== "available" && attempts >= 8 ? { status: "unavailable" as const } : {}) }).where(eq(chatVoiceReports.sessionId, session.id));
}
}
return rows.length;
}
async function callbackPreference(companyId: string, endpointId: string, caller: VoiceCaller) {
await endpointFor(companyId, endpointId);
await db.transaction(tx => store.authorizeCaller(tx, companyId, caller));
@@ -351,7 +300,7 @@ export function voiceSessionService(db: Db, options: {
const result = await start({ ...binding, mode: "outbound_phone", phoneNumber: preference.phoneNumber, idempotencyKey: invocationId, maxDurationSeconds: 600 });
return { session: result.session };
});
return { start, inspect, end, tool, reconcile, reconcileMaintenance, registerRecoveryWork, registerReplyWork, pushReplies, inbound, notification: store.notification, callbackPreference, saveCallbackPreference, history, callDetail, unregisterAgentTools };
return { start, inspect, end, tool, reconcile, registerReportWork, registerRecoveryWork, registerReplyWork, pushReplies, inbound, notification: store.notification, callbackPreference, saveCallbackPreference, history, callDetail, unregisterAgentTools };
}
export type VoiceSessionService = ReturnType<typeof voiceSessionService>;