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

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/idle-voice-call-reports.md

@@ -0,0 +1,76 @@
+# 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, schema, provider endpoint, or independent timer.
+
+## 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. Committed hints reload their
+session, endpoint, connection, or company scope. 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.
+
+Signals cover supported writes through the owning database client in this
+process. Independent SQL writers and another process need their own notification
+contract. 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).

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -1331,3 +1331,22 @@ Slice 24 is published as [#17583](https://github.com/paperclipai/paperclip/pull/
 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.

doc/plans/2026-10-25-voice-call-report-sleep-audit.md

@@ -3,7 +3,7 @@
 Baseline: voice history retention PR #17583 at
 9c38ddfe5349ef1872306a91c7e4c5c1bc083259.
 Branch: codex/sleep-25-voice-call-reports.
-Status: owner audit; implementation, verification and publication remain required.
+Status: implementation and local acceptance complete; publication and remote gates pending.
 
 ## The obligation
 
@@ -97,3 +97,45 @@ 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.

server/src/__tests__/app-private-hostname-gate.test.ts

@@ -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

@@ -0,0 +1,210 @@
+import { randomUUID } from "node:crypto";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { eq, sql } from "drizzle-orm";
+import { 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-");
+    db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+  }, 60_000);
+  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 = 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("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 quiet(60_000); expect(await report(call.id)).toBeUndefined();
+    enabled = true; await flush(); expect(fetchReport).toHaveBeenCalledOnce();
+    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 () => {
+    const broken = await session(); for (let i = 0; i < 103; i++) await session();
+    const other = await fixture(); const last = await session(other);
+    fetchReport.mockImplementation(async call => { if (call.id === broken.id) throw new Error("unavailable"); return result(); });
+    await start(); expect(fetchReport.mock.calls.slice(0, 25).some(([call]) => call.id === last.id)).toBe(true);
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(fetchReport).toHaveBeenCalledTimes(105); }, { timeout: 5000 });
+    expect((await report(broken.id)).attempts).toBe(1); await flush(); await quiet();
+  });
+  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

@@ -81,6 +81,77 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication };
   }
 
+  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("limits a new pre-call wake to its admission instead of scanning endpoint history", async () => {
     const f = await inboundFixture(); const hints: string[] = [];
     const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.voiceInboundRetention,
@@ -102,7 +173,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,
     });

server/src/app.ts

@@ -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

@@ -42,6 +42,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

@@ -0,0 +1,126 @@
+import { and, eq, gt, 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>) {
+  async function load(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise<Entry[]> {
+    const entries: Entry[] = []; let cursor: string | undefined;
+    for (;;) {
+      signal?.throwIfAborted();
+      const rows = await db.select({ id: chatVoiceSessions.id, companyId: chatVoiceSessions.companyId,
+        endpointId: chatVoiceSessions.endpointId, connectionId: chatEndpoints.connectionId,
+        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 ? gt(chatVoiceSessions.id, cursor) : undefined)).orderBy(chatVoiceSessions.id).limit(100);
+      entries.push(...rows.map(row => ({ ...row, dueAt: Number(row.dueAt) })));
+      if (rows.length < 100) return entries;
+      cursor = rows[rows.length - 1].id;
+    }
+  }
+  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) {
+    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, check: (entry, signal) => check(entry.id, signal),
+    }, canRun);
+  }
+  return { load, check, reconcile, register };
+}

server/src/services/voice/voice-recovery-notifications.ts

@@ -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);
+  await notifyVoiceReports(tx, scope);
 }
 export function withVoiceSessionChange<T>(db: Db, id: string, write: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => PromiseLike<T>, options: { replyDelivery?: boolean } = {}): Promise<T> {
   return db.transaction(async tx => {

server/src/services/voice/voice-reply-push.ts

@@ -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

@@ -0,0 +1,6 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
+import type { VoiceRecoveryScope } from "./voice-recovery-notifications.js";
+
+export function notifyVoiceReports(tx: object, scope?: VoiceRecoveryScope) {
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceReports, scope ? `${scope.kind}:${scope.id}` : "*");
+}

server/src/services/voice/voice-session-service.ts

@@ -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>;