fix(server): fence late voice binding during expiry recovery

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

doc/idle-voice-session-recovery.md

@@ -0,0 +1,97 @@
+# Voice session recovery without idle discovery
+
+The live-session worker replaces the session-discovery portion of the recurring
+voice maintenance call. It uses the existing delivery coordinator, transaction
+signals and shared scheduler. It adds no tables or migrations. Reply delivery,
+inbound admission cleanup, inbound history retention and call reports are separate
+obligations; their existing maintenance remains until their own migrations land.
+
+## Work and deadlines
+
+Existing `chat_voice_sessions` rows are authoritative. Startup reconstructs live
+sessions in UUID pages of 100, joined to their same-company endpoint and connection.
+Terminal history does not create a live-session recovery obligation. With no live
+sessions, the worker also avoids reading experimental settings after reconstruction.
+
+| State | Next obligation |
+| --- | --- |
+| No provider identity yet | Existing `expiresAt` plus two minutes of teardown grace; no provider creation replay |
+| Known provider, ending or invalidated configuration | Immediate cleanup; a recorded cleanup failure waits 30 seconds |
+| Connecting or active call | Earlier of session expiry and `updatedAt + 30 seconds` for current authorization and missed-hangup inspection |
+| Other live state with provider identity | Saved session expiry |
+| Ended, failed or expired | No live-session work; reports have a separate owner |
+
+Provider maximum duration remains the final bound for uncertain creation. A
+late provider response can bind identity while expiry is being considered. The
+expiration write requires that provider identity is still absent, so a newly
+known call retains exact-call cleanup rather than being marked terminal prematurely.
+
+Each due call is reloaded before execution. The voice store checks current
+company, endpoint, credentials, caller, task, guest sandbox and binding authority.
+A 403/404 means authority no longer permits the call. A transient lookup error
+retains a retry; it does not authorize ending an otherwise valid call. The existing
+voice service owns provider inspection and termination, including single-flight
+termination and ambiguous-response confirmation. Recovery never creates a call.
+
+## Producers and transaction ownership
+
+| Owner | Changes that wake live-session recovery |
+| --- | --- |
+| `voice-session-store.ts` | Reservation, guest promotion, connecting-to-active transition and reply cursor writes that change the inspection timestamp |
+| `voice-session-service.ts` | Creation claim, binding, known/unknown creation failure, user end, cleanup outcome and foreground inspection |
+| `voice-inbound-service.ts` | Guest session creation, guest denial and signed terminal lifecycle callbacks |
+| `voice-reply-push.ts` | Confirmed remote end and unknown delivery outcome that changes session state/timestamp |
+| Chat configuration helper | Endpoint configuration, pause, reconnect, generation/credential changes, connection hints and company removal |
+| Generic tool/access services | Voice connection API mutation, removal and credential revocation during access removal |
+| Instance settings | Explicit chat-connector feature enable/disable |
+
+Existing transactions register the awaited signal before the session/configuration
+write. Standalone session updates use the small `withVoiceSessionChange` helper.
+Provider requests remain outside database transactions. Nested writes retain outer
+commit ownership, rollback does not dispatch work, and uncertain commit settlement
+stays with the shared database protocol.
+
+Membership changes, task reassignment, phone preferences and other authorization
+inputs that do not mutate the session or connection remain checked on the existing
+30-second active-call inspection deadline, as well as on use. This slice does not
+promise instant cleanup for every authority mutation or add cross-cutting voice
+hooks throughout those owners. Session activity that changes `updatedAt` retains
+the existing inspection-time semantics. The session expiry is always an upper
+bound for authorized active work.
+
+The generic chat connection hint does not always carry a transport. It can cause
+a bounded empty voice lookup after a chat configuration write. It does not create
+idle polling or a recurring deadline. The shared helper contains only routing;
+voice eligibility, authority and provider policy stay with the voice owner.
+
+## Failure and lifecycle behavior
+
+Refreshes coalesce by scope, with a full refresh covering already pending narrower
+hints. Changes arriving during refresh remain dirty. Failed refreshes block use of
+stale cached entries in that scope. Per-scope and per-call failure backoff starts
+at 30 seconds and caps at five minutes. New hints do not erase failure backoff.
+Batches rotate across companies, process at most 25 calls, and continue an already
+due backlog after one second. A failed provider does not stall independent calls.
+
+Drain and standby retain dirty work without passive SQL. Resume processes that
+work, and restart reconstructs deadlines from the rows. Shutdown joins admitted
+provider work and removes the coordinator's subscriptions and timers. Retained
+terminal history becomes quiet after any already committed hints settle.
+
+Process-local signals support the existing single active database writer and its
+shared clients. Independent SQL writers or another replica need an explicit wake
+contract. This change neither expands Cloud alarm coverage nor relaxes sleep
+eligibility. Active calls and their real deadlines remain obligations; local worker
+quiet is not proof of host suspension or fleet savings.
+
+## Verification
+
+The voice integration suite exercises actual service producers, guest intake,
+signed callbacks, current authority, unknown creation and adjacent reply/report
+behavior. The recovery-owner suite uses real PostgreSQL and controlled provider
+operations for outer commit/rollback, drain, restart, scope failure, mutation during
+refresh, paging, company fairness, shutdown and the late-provider-binding race.
+The latter race failed before the write-time guard and passes afterward.
+
+The dated audit records exact commands, results, initial fixture failures and
+remaining acceptance work: [voice migration audit](plans/2026-10-25-voice-maintenance-sleep-audit.md).

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

@@ -141,3 +141,26 @@ Outstanding before publication: complete producer/authorization-change audit,
 current-state races, cross-company fairness and failed-scope coverage, affected
 shared producer consumers, full workspace verification and remote gates. The
 other four voice duties and every remaining original stack item remain required.
+
+
+## Producer and race audit completion
+
+See [the live-session contract](../idle-voice-session-recovery.md) for the complete
+producer mapping, existing transaction ownership and retained active authorization
+inspection contract. All production session writers are covered. Phone preferences,
+membership and task authority remain checked on the existing active-call deadline
+and on use; no new recurring empty inventory scan replaces their absence of signals.
+
+A late provider binding between grace-period discovery and expiration was reproduced:
+the previous update marked a connecting call expired despite its new provider ID.
+The expiration update now requires provider identity to remain absent. The regression
+proves the known call stays live for exact-provider cleanup. All 11 recovery lifecycle
+cases pass, including company fairness and failed-scope fencing.
+
+Expanded real-service coverage passes for endpoint pause, generic voice connection
+disable, task reassignment and signed guest lifecycle settlement. Shared consumer
+verification passes 469 tool-access, 18 access-removal, and 207 settings/provider
+cases. Workspace typechecking passes. A fresh final voice/receiver/startup selection
+and the full local test command are running; build/publication/remote gates remain.
+No full local matrix result is claimed from targeted tests. Initial fixture errors
+and the reproduced provider-binding race remain recorded above and in test logs.

server/src/__tests__/voice-recovery-work.test.ts

@@ -19,6 +19,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
   let enabled = true;
   const errors: unknown[] = [];
+  const fixtureCompanies: string[] = [];
   const authorize = vi.fn(async (_session: typeof chatVoiceSessions.$inferSelect) => {}), inspect = vi.fn(async (_session: typeof chatVoiceSessions.$inferSelect): Promise<{ endedAt?: unknown }> => ({ endedAt: true }));
   const close = vi.fn(async (session: typeof chatVoiceSessions.$inferSelect) => {
     await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)));
@@ -28,10 +29,10 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
     await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true });
   }, 60_000);
-  beforeEach(async () => {
+  async function fixture() {
-    companyId = randomUUID(); endpointId = randomUUID(); agentId = randomUUID(); issueId = randomUUID(); conversationId = randomUUID();
+    const companyId = randomUUID(), endpointId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), conversationId = randomUUID();
+    fixtureCompanies.push(companyId);
     const applicationId = randomUUID(), connectionId = randomUUID();
-    enabled = true; errors.length = 0; authorize.mockClear(); inspect.mockReset().mockResolvedValue({ endedAt: true }); close.mockClear();
     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" });
@@ -39,16 +40,23 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     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; authorize.mockClear(); inspect.mockReset().mockResolvedValue({ endedAt: true }); close.mockClear();
+    ({ 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(); });
@@ -127,6 +135,42 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     expect(errors).toHaveLength(1);
     broken = false; await settled(); errors.length = 0; await quiet();
   });
+  it("includes another company in the first batch despite a large older backlog", async () => {
+    for (let index = 0; index < 30; index++) await session({ updatedAt: new Date(Date.now() - 120_000) });
+    const other = await fixture();
+    const row = await session(other);
+    await start();
+    expect(inspect.mock.calls.slice(0, 25).some(([candidate]) => candidate.id === row.id)).toBe(true);
+    await settled(); expect(inspect).toHaveBeenCalledTimes(31); await quiet();
+  });
+  it("does not inspect stale cached work until a failed scope refresh succeeds", async () => {
+    const row = await session({ updatedAt: new Date(Date.now() - 20_000) });
+    const service = voiceSessionRecovery(db, { authorize, inspect, close }), original = service.load;
+    await start(service); service.load = async () => { throw new Error("scope unavailable"); };
+    await db.transaction(tx => notifyVoiceRecovery(tx, { kind: "session", id: row.id }));
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors).toHaveLength(1); });
+    await quiet(15_000); expect(inspect).not.toHaveBeenCalled();
+    service.load = original; await settled(); expect(inspect).toHaveBeenCalledOnce(); errors.length = 0; await quiet();
+  });
+  it("does not expire a provider identity bound after the grace-period read", async () => {
+    const row = await session({ state: "creation_unknown", providerSessionId: null,
+      createdAt: new Date(Date.now() - 240_000), expiresAt: new Date(Date.now() - 180_000) });
+    const service = voiceSessionRecovery(db, { authorize, inspect, close });
+    const providerSessionId = randomUUID(), transact = db.transaction.bind(db);
+    const transaction = vi.spyOn(db, "transaction").mockImplementationOnce(async (callback, options) => {
+      await transact(async tx => {
+        await notifyVoiceRecovery(tx, { kind: "session", id: row.id });
+        await tx.update(chatVoiceSessions).set({ providerSessionId, state: "connecting", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, row.id));
+      });
+      return transact(callback, options);
+    });
+    try { await service.check(row.id); } finally { transaction.mockRestore(); }
+    const [bound] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, row.id));
+    expect(bound).toMatchObject({ state: "connecting", providerSessionId });
+    await service.check(row.id);
+    expect(close).toHaveBeenCalledWith(expect.objectContaining({ id: row.id, providerSessionId }));
+    expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, row.id)))[0].state).toBe("ended");
+  });
   it("joins admitted provider inspection at shutdown", async () => {
     await start(); await settled();
     let finish!: () => void; const gate = new Promise<void>(resolve => { finish = resolve; });

server/src/__tests__/voice-session-store.integration.test.ts

@@ -1,3 +1,4 @@
+import { toolAccessService } from "../services/tool-access.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { idleWorkSnapshot } from "../services/task-admission.js";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
@@ -71,7 +72,7 @@ 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", "revocation", "feature disable"] as const)("wakes live recovery from a real start and settles after %s", async reason => {
+  it.each(["hangup", "revocation", "feature disable", "endpoint pause", "connection disable", "task reassignment"] as const)("wakes live recovery from a real start and settles after %s", async reason => {
     await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true });
     const f = await nativeServiceFixture();
     vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
@@ -96,7 +97,14 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
       if (reason === "hangup") f.transport.inspectSession.mockResolvedValue({ status: "ended", endedAt: new Date().toISOString() });
       if (reason === "revocation") await db.update(companyMemberships).set({ status: "removed" }).where(eq(companyMemberships.principalId, f.callerId));
       if (reason === "feature disable") await instanceSettingsService(db).updateExperimental({ enableChatConnectors: false });
-      else await vi.advanceTimersByTimeAsync(31_000);
+      if (reason === "endpoint pause") {
+        const chat = chatChannelService(db, { heartbeat: { wakeup: vi.fn() } });
+        try { await chat.configure(f.endpointId, { action: "pause" }, f.callerId); }
+        finally { await chat.shutdown(); }
+      }
+      if (reason === "connection disable") await toolAccessService(db).updateConnection(f.connectionId, { enabled: false });
+      if (reason === "task reassignment") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId));
+      if (["hangup", "revocation", "task reassignment"].includes(reason)) await vi.advanceTimersByTimeAsync(31_000);
       await settle(true);
       const [row] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, started.session.id));
       expect(row.state).toBe("ended");
@@ -128,6 +136,29 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     } finally { await coordinator.stop(); f.service.unregisterAgentTools(); vi.useRealTimers(); }
   });
 
+  it("recovers a real guest session and settles from its signed lifecycle event", async () => {
+    const f = await inboundFixture(true);
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    const settle = async () => vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); });
+    try {
+      await f.service.registerRecoveryWork(coordinator, () => true).ready;
+      expect(coordinator.nextWakeAt()).toBeNull();
+      await f.tool("submit_request", { text: "Help draft a public question" });
+      await settle(); expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now());
+      await vi.advanceTimersByTimeAsync(31_000); await settle();
+      expect(f.transport.inspectSession).toHaveBeenCalledWith(f.event.session_id);
+      expect(f.transport.endSession).not.toHaveBeenCalled();
+      const ended = f.sign({ type: "call.status", call_id: f.event.call_id, status: "ended" });
+      await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers);
+      await settle(); expect(coordinator.nextWakeAt()).toBeNull();
+      const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
+      try { await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); }
+      finally { reads.mockRestore(); writes.mockRestore(); }
+      expect(onError).not.toHaveBeenCalled();
+    } finally { await coordinator.stop(); f.service.unregisterAgentTools(); vi.useRealTimers(); }
+  });
+
   it("reports signing-secret presence for interrupted setup without returning credential refs", async () => {
     const f = await fixture();
     const chat = chatChannelService(db, {heartbeat: {wakeup: vi.fn()}});

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

@@ -1,4 +1,4 @@
-import { and, eq, gt, inArray, sql } from "drizzle-orm";
+import { and, eq, gt, inArray, isNull, sql } from "drizzle-orm";
 import { chatEndpoints, chatVoiceSessions, toolConnections, type Db } from "@paperclipai/db";
 import { HttpError } from "../../errors.js";
 import { instanceSettingsService } from "../instance-settings.js";
@@ -75,7 +75,7 @@ export function voiceSessionRecovery(db: Db, operations: {
         } else if (session.providerSessionId) await operations.close(session);
         else if (session.expiresAt.getTime() + 120_000 < Date.now()) {
           await withVoiceSessionChange(db, id, tx => tx.update(chatVoiceSessions).set({ state: "expired", endedAt: new Date(), updatedAt: new Date() })
-            .where(and(eq(chatVoiceSessions.id, id), inArray(chatVoiceSessions.state, [...LIVE_VOICE_STATES]))));
+            .where(and(eq(chatVoiceSessions.id, id), isNull(chatVoiceSessions.providerSessionId), inArray(chatVoiceSessions.state, [...LIVE_VOICE_STATES]))));
         }
       } catch (error) {
         await withVoiceSessionChange(db, id, tx => tx.update(chatVoiceSessions).set({ errorCode: "cleanup_pending", updatedAt: new Date() })