From 8f5e861dbd6ea5c2b45684dff970ea27723a84cb Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 03:17:23 -0500 Subject: [PATCH] fix(server): fence late voice binding during expiry recovery Co-Authored-By: Paperclip --- doc/idle-voice-session-recovery.md | 97 +++++++++++++++++++ ...026-10-25-voice-maintenance-sleep-audit.md | 23 +++++ .../src/__tests__/voice-recovery-work.test.ts | 50 +++++++++- .../voice-session-store.integration.test.ts | 35 ++++++- .../services/voice/voice-session-recovery.ts | 4 +- 5 files changed, 202 insertions(+), 7 deletions(-) create mode 100644 doc/idle-voice-session-recovery.md diff --git a/doc/idle-voice-session-recovery.md b/doc/idle-voice-session-recovery.md new file mode 100644 index 00000000000..187eeef1e85 --- /dev/null +++ b/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). diff --git a/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md b/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md index a1e4489b372..5652484ae2f 100644 --- a/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md +++ b/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. diff --git a/server/src/__tests__/voice-recovery-work.test.ts b/server/src/__tests__/voice-recovery-work.test.ts index c6bc776490c..946209e55cb 100644 --- a/server/src/__tests__/voice-recovery-work.test.ts +++ b/server/src/__tests__/voice-recovery-work.test.ts @@ -19,6 +19,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu let coordinator: ReturnType | 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 () => { - companyId = randomUUID(); endpointId = randomUUID(); agentId = randomUUID(); issueId = randomUUID(); conversationId = randomUUID(); + async function fixture() { + 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(resolve => { finish = resolve; }); diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 372acfbec58..5d28be54b66 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/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()}}); diff --git a/server/src/services/voice/voice-session-recovery.ts b/server/src/services/voice/voice-session-recovery.ts index 91410057935..6685187a4e7 100644 --- a/server/src/services/voice/voice-session-recovery.ts +++ b/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() })