diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 47893cb2783..6c79515d99f 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1272,3 +1272,17 @@ checks pass; remote acceptance is pending. Slice 23 starts the inbound admission cleanup audit in `2026-10-25-voice-inbound-cleanup-sleep-audit.md`. It explicitly separates retention and reports and requires proof of approval/cleanup handoff, termination confirmation, and protection against deleting unfinished cleanup. + + +Voice reply review repairs are published at #17578 head `90caaf3fa9`: shared scoped +worker, actual authority-restoration wiring regression, 191 overlapping selected +consumer tests and workspace types/build passing. Removing the real notification +reproduces the lost reply. Both review threads are resolved; fresh CI/review remain. + +Inbound cleanup slice 23 now makes approval/session handoff atomic, after a real +expiry regression reproduced termination of an already authorized call. All 110 +voice integration cases and server typechecking pass. This slice is not published: +locked cleanup claims, legacy handoff recovery, bound-guest ownership, termination +confirmation, retention protection and scheduler/producer migration remain required. +The full original inventory, older #17490 E2E failure, and stack/master reconciliation +remain outstanding. No stack PR has been merged by this work. diff --git a/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md b/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md index 1adc795f33d..2110c23b6e5 100644 --- a/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md +++ b/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md @@ -2,7 +2,7 @@ Baseline: voice reply PR #17578, head 439d0a86eb75c129c30742a8cc24bf18e2a6ebf3. Branch: codex/sleep-23-voice-inbound-cleanup. -Status: owner audit; implementation and verification remain required. +Status: atomic approval handoff implemented and verified; cleanup worker, remaining ownership cases, and publication remain required. ## One reason in this slice @@ -112,3 +112,39 @@ live at publication; no check failure was present at the latest inspection. Latest-head checks, Greptile score and review findings still require acceptance. No stack PR is authorized for merge by the active goal. The remaining original ledger, older #17490 failure, and final master reconciliation remain required. + + +## Atomic handoff checkpoint + +The parent reply review repairs are published at `90caaf3fa929141a13cc17f54d6be846d6e93217` +and merged into this branch. Both voice workers now share the scoped deadline +mechanism; the next inbound worker should reuse it rather than copy a third loop. +Parent CI run 40180022971 is live; no failure was present at the latest inspection. +Its old 439d head passed CI/Greptile, but fresh-head acceptance remains required. + +A real-service regression paused after inbound session reservation committed but +before its result reached `decide`, then advanced beyond the admission expiry. +The original sweep called provider termination on that authorized session. This +failed in `/tmp/sleep23-handoff-before.log`. The reservation now writes admission +state=approved and sessionId in the same transaction as insertion or guest +promotion. `decide` reads that binding after start, rather than creating it in a +second transaction. No provider call is moved inside the transaction. + +Both ordinary admission and guest-to-private promotion are covered. All 110 voice +integration cases pass on a fresh PostgreSQL database, and server typechecking +passes. An initial full run passed 109 and failed the guest test's assertion of +no termination calls globally: advancing ten minutes also expired unrelated +retained fixture admissions. The assertion now checks the exact provider call +whose handoff is under test, preserving legitimate unrelated cleanup. The full +rerun passes. No broad workspace/remote verification is claimed for this child. + +This is a prerequisite, not the complete cleanup migration. Still required: +- Recheck and claim a discovered admission under its row lock before provider I/O; + atomic new handoff alone does not protect a stale discovery snapshot. +- Reconstruct legacy interrupted handoffs written before this atomic change. +- Define one termination owner for bound guest sessions, and settlement signals. +- Preserve unconfirmed/failed termination debt and credential/account isolation. +- Prevent history retention from deleting unfinished cleanup. +- Wire all producer notifications, actual deadlines, scoped worker registration, + lifecycle tests and remove only the cleanup portion of periodic reconciliation. +- Keep history retention and call reports as separate forthcoming PRs. diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 4192ef75764..16af29cb463 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -24,6 +24,7 @@ import { chatActions, instanceUserRoles, principalPermissionGrants, environments import { configureSpekoSessionTools } from "../services/voice/speko-tool-setup.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { chatChannelService } from "../services/chat-channels.js"; +import { voiceInboundService } from "../services/voice/voice-inbound-service.js"; import { voiceSessionService } from "../services/voice/voice-session-service.js"; import { captureVoiceDeliveryDiagnostics } from "../services/voice/voice-delivery-diagnostics.js"; import { parseSpekoDeliveryDiagnostics } from "../services/voice/speko-delivery-diagnostics.js"; @@ -1256,6 +1257,40 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(repeated?.latestRevisionId).toBe(first?.latestRevisionId); expect(await docs.listIssueDocuments(f.call.intakeIssueId!)).toHaveLength(1); }); + it.each([false, true])("hands inbound cleanup ownership to the authorized session atomically (guest=%s)", async guest => { + const f = await inboundFixture(guest); + if (guest) await f.tool("submit_request", { text: "A public request before private approval" }); + let release!: () => void, bound!: () => void; + const bindingReached = new Promise(resolve => { bound = resolve; }); + const releaseApproval = new Promise(resolve => { release = resolve; }); + const inbound = voiceInboundService(db, { + allowLocalBoard: false, credentials: f.credentials, provider: () => f.transport, + voicePrompt: "Test voice", lowTrustVoicePrompt: "Test guest voice", + start: async input => { + const result = await f.service.start(input); + bound(); await releaseApproval; + return result; + }, + }); + vi.useFakeTimers({ toFake: ["Date"] }); + const approval = inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, + { approve: true, approvalCode: f.call.approvalCode, issueId: f.issueId }); + try { + await Promise.race([bindingReached, approval.then(() => { throw new Error("Approval did not reach session binding"); })]); + // The session has committed but its caller has not received the result. + // Expiry or process death here must not leave two cleanup owners. + vi.setSystemTime(f.call.expiresAt.getTime() + 1); + await inbound.reconcile(1000); + expect(f.transport.endSession).not.toHaveBeenCalledWith(f.call.providerSessionId); + const [call] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)); + expect(call).toMatchObject({ state: "approved" }); + expect(call.sessionId).not.toBeNull(); + const [session] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, call.sessionId!)); + expect(session).toMatchObject({ providerSessionId: f.call.providerSessionId, callerId: f.callerId, issueId: f.issueId }); + release(); + await expect(approval).resolves.toMatchObject({ state: "approved", sessionId: session.id }); + } finally { release(); await approval.catch(() => undefined); vi.useRealTimers(); } + }); it("requires approval of the specific incoming call before delivering any private work", async () => { const f = await inboundFixture(); expect(await f.service.inbound.lifecycle(f.endpoint.publicId, f.signed.body, f.signed.headers)).toEqual(f.response); diff --git a/server/src/services/voice/voice-inbound-service.ts b/server/src/services/voice/voice-inbound-service.ts index 8b94722a366..36b60dbccb1 100644 --- a/server/src/services/voice/voice-inbound-service.ts +++ b/server/src/services/voice/voice-inbound-service.ts @@ -244,7 +244,7 @@ export function voiceInboundService(db: Db, options: { if (!input.approve) { await client.endSession(admission.providerSessionId).catch(() => undefined); return projection(admission); } if ((await client.inspectSession(admission.providerSessionId)).endedAt) { await db.update(chatVoiceInboundCalls).set({state: "ended", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, callId)); throw conflict("The caller already hung up"); } const result = await options.start({companyId, endpointId, issueId: admission.requestedIssueId ?? undefined, caller, idempotencyKey: `inbound:${admission.id}`, maxDurationSeconds: 600, mode: "inbound_phone", providerSessionId: admission.providerSessionId, toolToken: token(row, admission.providerSessionId, admission.generation, credentials.signingSecret)}); - [admission] = await db.update(chatVoiceInboundCalls).set({state: "approved", sessionId: result.session.id, updatedAt: new Date()}).where(and(eq(chatVoiceInboundCalls.id, callId), eq(chatVoiceInboundCalls.state, "approving"))).returning(); + [admission] = await db.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.id, callId), eq(chatVoiceInboundCalls.state, "approved"), eq(chatVoiceInboundCalls.sessionId, result.session.id))); if (!admission) throw conflict("This call ended during approval"); return projection(admission); } diff --git a/server/src/services/voice/voice-session-store.ts b/server/src/services/voice/voice-session-store.ts index d9247856a5f..05d54e5856a 100644 --- a/server/src/services/voice/voice-session-store.ts +++ b/server/src/services/voice/voice-session-store.ts @@ -175,9 +175,18 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n const context = await endpointContext(tx, input.companyId, input.endpointId); await authorizeConnection(tx, context, input.caller); if (context.generation < 1 || input.maxDurationSeconds < 30 || input.maxDurationSeconds > 1800 || !Number.isInteger(input.maxDurationSeconds)) throw conflict("Voice setup or duration is invalid"); + let inboundAdmissionId: string | undefined; + async function bindInboundSession(sessionId: string) { + if (!inboundAdmissionId) return; + // Admission expiry must relinquish cleanup in the same commit that + // establishes an authorized owner for the existing provider call. + await tx.update(chatVoiceInboundCalls).set({ state: "approved", sessionId, updatedAt: new Date() }) + .where(eq(chatVoiceInboundCalls.id, inboundAdmissionId)); + } if (input.mode === "inbound_phone") { const [admission] = await tx.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, input.companyId), eq(chatVoiceInboundCalls.endpointId, input.endpointId), eq(chatVoiceInboundCalls.providerSessionId, input.providerSessionId ?? ""))).for("update"); if (!admission || admission.state !== "approving" || admission.approvedByUserId !== input.caller.id || admission.callerAuthority !== input.caller.authority || admission.expiresAt.getTime() <= Date.now() || admission.generation !== context.generation || admission.credentialFingerprint !== context.fingerprint || admission.toolTokenHash !== voiceTokenHash(input.toolToken ?? "") || input.idempotencyKey !== `inbound:${admission.id}`) throw forbidden("This live call has not been approved for private work"); + inboundAdmissionId = admission.id; } const binding = await input.prepareBinding(tx); const task = await authorizeTask(tx, input.companyId, binding.issueId, input.caller); @@ -197,6 +206,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n await notifyVoiceChange(tx, { kind: "session", id: guest.id }); const [promoted] = await tx.update(chatVoiceSessions).set({...binding, callerId: input.caller.id, callerAuthority: input.caller.authority, approvedByUserId: input.caller.id, idempotencyKey: input.idempotencyKey, requestFingerprint: input.requestFingerprint, updatedAt: new Date()}).where(eq(chatVoiceSessions.id, guest.id)).returning(); await tx.insert(activityLog).values({companyId: input.companyId, actorType: "user", actorId: input.caller.id, action: "voice.guest.private_call_approved", entityType: "voice_session", entityId: guest.id, details: {issueId: binding.issueId}}); + await bindInboundSession(promoted.id); return {session: promoted, created: false as const}; } } @@ -207,6 +217,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n callerId: input.caller.id, callerAuthority: input.caller.authority, mode: input.mode ?? "browser", state: input.providerSessionId ? "connecting" : "reserved", providerSessionId: input.providerSessionId ?? null, approvedByUserId: input.mode === "inbound_phone" ? input.caller.id : null, generation: context.generation, credentialFingerprint: context.fingerprint, toolTokenHash: voiceTokenHash(token), idempotencyKey: input.idempotencyKey, requestFingerprint: input.requestFingerprint, createdAt, expiresAt: new Date(createdAt.getTime() + input.maxDurationSeconds * 1000) }).returning(); await tx.insert(activityLog).values({ companyId: input.companyId, actorType: "user", actorId: input.caller.id, action: "voice.session.requested", entityType: "voice_session", entityId: session.id, details: { endpointId: input.endpointId, issueId: session.issueId, mode: input.mode ?? "browser" } }); + await bindInboundSession(session.id); return { session, created: true as const, token }; }); }