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 4c8dc6659a4..5ed3fa1fca8 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1237,3 +1237,12 @@ full-refresh failure/backoff case. Server typecheck and build pass. Prior broad consumer/workspace verification remains recorded above; these 45 overlap that coverage and are not an additional unique total. Publication requires fresh remote CI and review. The full migration inventory remains outstanding. + + +Voice recovery checkpoint: receiver repair `1604a9fcba` is published to #17572 +and incorporated into the unpublished voice child. The complete 93-case voice +suite, eight new recovery lifecycle tests, and 16 startup wiring cases pass. +See the voice audit for precise coverage and remaining work. Receiver CI +40175242445 is live with no failures at inspection; no final acceptance or merge +is claimed. All remaining reasons, older #17490 failure, and final master +reconciliation stay in scope. 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 fbce8e14676..a1e4489b372 100644 --- a/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md +++ b/doc/plans/2026-10-25-voice-maintenance-sleep-audit.md @@ -2,7 +2,7 @@ Baseline: receiver parent #17572, head 643f4766227e15006b684d2cc7bea7ddd5cc2147. Branch: codex/sleep-21-voice-session-recovery. -Status: owner audit only. No voice polling removal is implemented or verified. +Status: implementation and local integration/lifecycle verification in progress; unpublished. ## Separate the reasons before changing the timer @@ -112,3 +112,32 @@ requests suppressing rediscovery after discovery-initiated stops. Its optional full-refresh coalescing observation will also be addressed in the owning parent before this branch is published. Preserve all unfinished voice changes in this commit while repairing that parent. + + +## Expanded lifecycle checkpoint + +Receiver review repairs are published in #17572 at `1604a9fcba` and merged into +this child. The receiver CI run 40175242445 remains live, with no failing checks +at inspection. The voice worker now also coalesces pending scopes under a full +refresh and retains its backoff when narrower signals arrive. + +The complete existing voice session integration suite passes all 93 tests on a +fresh disposable PostgreSQL database. This includes the four initial real-service +producer cases and the adjacent reply, inbound, callback and report consumers. +Eight new recovery-owner integration tests pass: configured terminal history, +outer rollback/commit, first write during drain, transient authority failure +without ending a valid call, restart with the same saved deadline, mutation during +full refresh, a 105-session paged backlog with one failed provider, and shutdown +joining admitted inspection. Their provider operations are controlled; real +service authorization/producer coverage remains in the 93-case suite. + +The 16 startup wiring tests also pass. Initial new-suite failures were fixture +errors (missing channel purpose) and a silence assertion starting before the +committed cleanup_pending refresh had settled. Both were corrected; the final +retry check proves no SQL during the actual backoff. No production failure was +concealed or treated as a successful run. + +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. diff --git a/server/src/__tests__/voice-recovery-work.test.ts b/server/src/__tests__/voice-recovery-work.test.ts new file mode 100644 index 00000000000..c6bc776490c --- /dev/null +++ b/server/src/__tests__/voice-recovery-work.test.ts @@ -0,0 +1,139 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { agents, chatConversations, chatEndpoints, chatVoiceSessions, 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 { instanceSettingsService } from "../services/instance-settings.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { voiceSessionRecovery } from "../services/voice/voice-session-recovery.js"; +import { registerVoiceRecoveryWork } from "../services/voice/voice-recovery-work.js"; +import { notifyVoiceRecovery, withVoiceSessionChange } from "../services/voice/voice-recovery-notifications.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("voice session recovery lifecycle", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string, endpointId: string, agentId: string, issueId: string, conversationId: string; + let coordinator: ReturnType | undefined; + let enabled = true; + const errors: unknown[] = []; + 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))); + }); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("voice-recovery-"); + 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(); + 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" }); + 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 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); + 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 = {}, 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: "active", 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 = voiceSessionRecovery(db, { authorize, inspect, close })) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + await registerVoiceRecoveryWork(coordinator, service, () => enabled).ready; + return service; + } + async function settled() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); }, { timeout: 5000 }); + } + async function quiet(ms = 120_000) { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(ms); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + } + it("leaves configured terminal history quiet", async () => { + await session({ state: "ended", endedAt: new Date() }); await start(); await settled(); await quiet(); expect(inspect).not.toHaveBeenCalled(); + }); + it("waits for outer commit and ignores a rolled-back nested producer", async () => { + await start(); await settled(); + await expect(db.transaction(async tx => { await session({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(60_000); expect(inspect).not.toHaveBeenCalled(); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await settled(); await quiet(); + await db.transaction(async tx => { await session({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(60_000); expect(inspect).not.toHaveBeenCalled(); }); + await settled(); expect(inspect).toHaveBeenCalledOnce(); await quiet(); + }); + it("preserves the first session during drain without passive database reads", async () => { + await start(); await settled(); enabled = false; await session(); await quiet(60_000); expect(inspect).not.toHaveBeenCalled(); + enabled = true; await settled(); expect(inspect).toHaveBeenCalledOnce(); await quiet(); + }); + it("retries unavailable authority without ending a valid provider call", async () => { + const row = await session(); authorize.mockRejectedValueOnce(new Error("authority database unavailable")); + await start(); expect(errors).toHaveLength(1); expect(close).not.toHaveBeenCalled(); expect(inspect).not.toHaveBeenCalled(); + // Persisting cleanup_pending emits a committed hint; let that refresh settle first. + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now() + 20_000); + await quiet(10_000); await settled(); + expect(inspect).toHaveBeenCalledOnce(); expect(close).not.toHaveBeenCalled(); + expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, row.id)))[0].state).toBe("ended"); + errors.length = 0; await quiet(); + }); + it("reconstructs a saved session deadline after restart", async () => { + const due = Date.now() + 60_000; + await session({ updatedAt: new Date(due - 30_000) }); await start(); + expect(coordinator!.nextWakeAt()).toBe(due); await coordinator!.stop(); + await start(); expect(coordinator!.nextWakeAt()).toBe(due); await quiet(10_000); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); expect(inspect).toHaveBeenCalledOnce(); await quiet(); + }); + it("preserves a session committed during a full refresh", async () => { + const service = voiceSessionRecovery(db, { authorize, inspect, close }), original = service.load; + await start(service); await settled(); let inserted = false; + service.load = async (...args) => { const rows = await original(...args); if (!inserted) { inserted = true; await session(); } return rows; }; + await db.transaction(tx => notifyVoiceRecovery(tx)); await settled(); expect(inspect).toHaveBeenCalledOnce(); await quiet(); + }); + it("pages saved sessions and drains healthy work while one provider fails", async () => { + const failed = await session(); + for (let index = 0; index < 104; index++) await session(); + let broken = true; + inspect.mockImplementation(async row => { + if (row.id === failed.id && broken) throw new Error("one provider unavailable"); + return { endedAt: true }; + }); + await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + const rows = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.companyId, companyId)); + expect(rows.filter(row => row.state === "ended")).toHaveLength(104); + }, { timeout: 5000 }); + expect(inspect.mock.calls.filter(([row]) => row.id === failed.id)).toHaveLength(1); + expect(errors).toHaveLength(1); + broken = false; await settled(); errors.length = 0; await quiet(); + }); + it("joins admitted provider inspection at shutdown", async () => { + await start(); await settled(); + let finish!: () => void; const gate = new Promise(resolve => { finish = resolve; }); + inspect.mockImplementationOnce(async () => { await gate; return { endedAt: true }; }); + await session(); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(inspect).toHaveBeenCalledOnce(); }); + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(5000); expect(stopped).toBe(false); + finish(); await stopping; expect(stopped).toBe(true); + }); +}); diff --git a/server/src/services/voice/voice-recovery-work.ts b/server/src/services/voice/voice-recovery-work.ts index 356f00d852a..85da30c2d31 100644 --- a/server/src/services/voice/voice-recovery-work.ts +++ b/server/src/services/voice/voice-recovery-work.ts @@ -28,9 +28,9 @@ export function registerVoiceRecoveryWork( function next() { let nextAt: number | null = null; const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; - for (const hint of dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); + for (const hint of dirty.has("*") ? ["*"] : dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); const unavailable = dirtyScopes(); - for (const entry of entries.values()) if (entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope))) + for (const entry of entries.values()) if (!unavailable.some(scope => matches(entry, scope))) take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); return nextAt; } @@ -42,10 +42,11 @@ export function registerVoiceRecoveryWork( if (!canRun() || Date.now() < earliest) return; earliest = Date.now() + 1000; const errors: unknown[] = []; - const requested = [...dirty].filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); - for (const hint of requested) dirty.delete(hint); + const requested = (dirty.has("*") ? ["*"] : [...dirty]).filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); + const covered = requested.includes("*") ? [...dirty] : requested; + for (const hint of covered) dirty.delete(hint); for (let index = 0; index < requested.length; index++) { - if (signal.aborted || !canRun()) { for (const hint of requested.slice(index)) dirty.add(hint); return; } + if (signal.aborted || !canRun()) { for (const hint of requested.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; } const hint = requested[index]; let scope: VoiceRecoveryScope | undefined; let decoded = false; @@ -56,16 +57,16 @@ export function registerVoiceRecoveryWork( entries.delete(entry.id); failures.delete(entry.id); } for (const entry of rows) entries.set(entry.id, entry); - scopeFailures.delete(hint); + if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint); } catch (error) { const retryHint = decoded ? hint : "*"; dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error); } } const unavailable = dirtyScopes(); - const due = [...entries.values()].filter(entry => entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope)) + const due = [...entries.values()].filter(entry => !unavailable.some(scope => matches(entry, scope)) && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) - .sort((a, b) => a.dueAt! - b.dueAt! || a.id.localeCompare(b.id)); + .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)); const batch = takeCompanyWorkBatch(due, 25, lastCompany); lastCompany = batch.lastCompany; for (const entry of batch.entries) { if (signal.aborted || !canRun()) break;