diff --git a/doc/idle-voice-reply-delivery.md b/doc/idle-voice-reply-delivery.md index 1315d2ae946..3cb7b37ef14 100644 --- a/doc/idle-voice-reply-delivery.md +++ b/doc/idle-voice-reply-delivery.md @@ -55,3 +55,10 @@ Signals cover the supported process-local database owner. Independent SQL writer or other processes require their own notification/reconciliation contract. These changes do not extend Cloud wake coverage, permit sleep during a live call, or prove deployed instance/database suspension. + + +Both live-session recovery and reply delivery use `scoped-deadline-work.ts` for +coalescing, fair batching, scope reconstruction, retries and drain retention. +Their adapters keep domain state and provider operations in the voice owner. +The production-wiring regression covers authority returning before live recovery +observes the denial; recovery still ends calls when their authority is absent. diff --git a/doc/plans/2026-10-25-voice-reply-delivery-sleep-audit.md b/doc/plans/2026-10-25-voice-reply-delivery-sleep-audit.md index cc87e82979b..d61f9e1856e 100644 --- a/doc/plans/2026-10-25-voice-reply-delivery-sleep-audit.md +++ b/doc/plans/2026-10-25-voice-reply-delivery-sleep-audit.md @@ -203,3 +203,45 @@ shared-consumer cases (786 selected tests). Workspace typecheck and build passed The final added failure case changed tests only after the typecheck; the workspace build compiled the current tree. No production changes followed the full run. Remote CI/review remains pending until this branch is published. + + +## Review follow-up: shared scheduling and real authority restoration + +Paperclip review at `439d0a86eb` requested one scheduling implementation and a +production-wiring regression. `scoped-deadline-work.ts` now owns coalescing, +scoped reconstruction, bounded fair batches, per-entry/scope backoff and drain +retention for both voice workers. The adapters retain their own scope matching, +load/check contracts, eligibility and provider operations. The helper has no SQL, +provider knowledge, or feature flags; absent consumers register no work. + +The restoration test now registers the real voice service's reply and live-call +recovery workers. It first lets the reply encounter suspended membership, restores +membership before the next live-call authority inspection, and verifies exactly +one send through the production authorization callback. Healthy inspections do +not resend; confirmed hangup leaves both workers SQL-quiet. This deliberately +preserves the documented window: if recovery observes denied authority first, +it closes the call rather than waiting indefinitely for access to return. + +A mutation probe removed only the production `authorityAvailable` invocation. +The new test failed with zero sends instead of one. The production line was +restored in a finally block; the unmodified case passes. This replaces the older +test that called the notification directly. + +Full voice/startup/coordinator/batching selection passed 180 tests on a fresh +PostgreSQL database. Initial parent recovery verification passed nine and failed +two: the extraction captured load methods at registration, bypassing the tests' +later injected refresh failures and concurrent-write barriers. Adapters now call +the service methods at dispatch time, preserving the original boundary. All 11 +parent lifecycle tests pass on their separately isolated database. These failures +are retained here rather than counted as successful initial verification. + +The optional review suggestion to cache the feature flag across a batch is not +applied: current checks retain authority/configuration checks at dispatch, including +changes while earlier provider calls are in flight. Any optimization must preserve +that behavior. Workspace checks and fresh remote acceptance are recorded below +when complete; the prior 5/5 Greptile score does not cover this follow-up yet. + +Workspace `pnpm -r typecheck` and `pnpm build` pass after the review repairs. +The 191 selected tests above are overlapping consumer verification, not an +additional 191 unique cases on top of the earlier 786. Remote acceptance remains +required on the published follow-up head. diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index d9f3ce1ddbe..4192ef75764 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -77,6 +77,50 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication }; } + it("restores blocked replies through the production live-call recovery wiring", async () => { + const f = await pushFixture(); + await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.principalId, f.callerId)); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const settle = () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + try { + // Reply denial is remembered by the service's own schedule. No test calls + // authorityAvailable: only production recovery may reactivate this reply. + await f.service.registerReplyWork(coordinator, () => true).ready; + await settle(); + expect(coordinator.nextWakeAt()).toBeNull(); + expect(f.transport.sendCallMessage).not.toHaveBeenCalled(); + await db.update(companyMemberships).set({ status: "active" }).where(eq(companyMemberships.principalId, f.callerId)); + await f.service.registerRecoveryWork(coordinator, () => true).ready; + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(f.transport.sendCallMessage).toHaveBeenCalledOnce(); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + expect(f.transport.inspectSession).toHaveBeenCalled(); + expect(f.transport.endSession).not.toHaveBeenCalled(); + // Repeated healthy inspections do not republish the delivered answer. + await vi.advanceTimersByTimeAsync(31_000); await settle(); + expect(f.transport.sendCallMessage).toHaveBeenCalledOnce(); + f.transport.inspectSession.mockResolvedValue({ status: "ended", endedAt: new Date().toISOString() }); + await vi.advanceTimersByTimeAsync(31_000); await settle(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(coordinator.nextWakeAt()).toBeNull(); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); transactions.mockRestore(); } + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + 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(); @@ -780,27 +824,6 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await worker.quiet(); expect(worker.onError).not.toHaveBeenCalled(); } finally { await worker.coordinator.stop(); vi.useRealTimers(); } }); - it("reactivates an authority-blocked phone reply when active-call recovery verifies restored access", async () => { - const f = await pushFixture(); - await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.principalId, f.callerId)); - vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); - const worker = await replyWorker(f); - try { - await worker.settled(); await worker.quiet(); expect(f.transport.sendCallMessage).not.toHaveBeenCalled(); - await db.update(companyMemberships).set({ status: "active" }).where(eq(companyMemberships.principalId, f.callerId)); - await db.transaction(async tx => { - await f.store.authorizeSession(tx, f.companyId, f.session.id); - await worker.service.authorityAvailable(tx, f.session.id); - }); - await worker.settled(); expect(f.transport.sendCallMessage).toHaveBeenCalledOnce(); await worker.quiet(); - // Healthy active-call inspections must not wake an already idle queue. - await db.transaction(async tx => { - await f.store.authorizeSession(tx, f.companyId, f.session.id); - await worker.service.authorityAvailable(tx, f.session.id); - }); - expect(worker.coordinator.nextWakeAt()).toBeNull(); await worker.quiet(); expect(worker.onError).not.toHaveBeenCalled(); - } finally { await worker.coordinator.stop(); vi.useRealTimers(); } - }); it("preserves a newly committed phone reply through drain without passive SQL", async () => { const f = await pushFixture(); await f.push(); vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts new file mode 100644 index 00000000000..ee7f532b14e --- /dev/null +++ b/server/src/services/scoped-deadline-work.ts @@ -0,0 +1,90 @@ +import { takeCompanyWorkBatch } from "./company-work-batch.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import type { DeliveryQueue } from "./delivery-work-notifications.js"; + +type ScheduledEntry = { id: string; companyId: string; dueAt: number }; + +/** Coalesced scope reconstruction and fair deadline dispatch. Domain owners + * supply persistence, eligibility, and execution; this helper owns no SQL. */ +export function registerScopedDeadlineWork( + coordinator: ReturnType, + options: { + queue: DeliveryQueue; + decodeScope(hint: string): Scope | undefined; + matches(entry: Entry, scope?: Scope): boolean; + load(scope?: Scope, signal?: AbortSignal): Promise; + check(entry: Entry, signal?: AbortSignal): Promise; + }, + canRun: () => boolean, +) { + const dirty = new Set(["*"]); + const entries = new Map(); + const failures = new Map(); + const scopeFailures = new Map(); + let earliest = 0; + let lastCompany: string | null = null; + function failed(map: typeof failures, id: string) { + const count = Math.min((map.get(id)?.count ?? 0) + 1, 5); + map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) }); + } + function dirtyScopes() { + return [...dirty].map(hint => { try { return options.decodeScope(hint); } catch { return undefined; } }); + } + function next() { + let nextAt: number | null = null; + const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; + 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 (!unavailable.some(scope => options.matches(entry, scope))) + take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); + return nextAt; + } + const worker = coordinator.register(options.queue, { + retryMs: 30_000, + pausePassiveMaintenance: true, + onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); }, + run: async signal => { + if (!canRun() || Date.now() < earliest) return; + earliest = Date.now() + 1000; + const errors: unknown[] = []; + 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.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; } + const hint = requested[index]; + let scope: Scope | undefined; + let decoded = false; + try { + scope = options.decodeScope(hint); decoded = true; + const rows = await options.load(scope, signal), retained = new Set(rows.map(row => row.id)); + for (const entry of entries.values()) if (options.matches(entry, scope) && !retained.has(entry.id)) { + entries.delete(entry.id); failures.delete(entry.id); + } + for (const entry of rows) entries.set(entry.id, entry); + 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 => !unavailable.some(scope => options.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)); + const batch = takeCompanyWorkBatch(due, 25, lastCompany); lastCompany = batch.lastCompany; + for (const entry of batch.entries) { + if (signal.aborted || !canRun()) break; + try { + const current = await options.check(entry, signal); + if (current) entries.set(entry.id, current); else entries.delete(entry.id); + failures.delete(entry.id); + } catch (error) { failed(failures, entry.id); errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, `${options.queue} could not settle`); + }, + hasPending: async () => dirty.size > 0, + nextRunAt: next, + }); + return worker; +} diff --git a/server/src/services/voice/voice-recovery-work.ts b/server/src/services/voice/voice-recovery-work.ts index 85da30c2d31..be140f51d57 100644 --- a/server/src/services/voice/voice-recovery-work.ts +++ b/server/src/services/voice/voice-recovery-work.ts @@ -1,4 +1,4 @@ -import { takeCompanyWorkBatch } from "../company-work-batch.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 type { VoiceRecoveryEntry, voiceSessionRecovery } from "./voice-session-recovery.js"; @@ -12,74 +12,11 @@ export function registerVoiceRecoveryWork( service: ReturnType, canRun: () => boolean, ) { - const dirty = new Set(["*"]); - const entries = new Map(); - const failures = new Map(); - const scopeFailures = new Map(); - let earliest = 0; - let lastCompany: string | null = null; - function failed(map: typeof failures, id: string) { - const count = Math.min((map.get(id)?.count ?? 0) + 1, 5); - map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) }); - } - function dirtyScopes() { - return [...dirty].map(hint => { try { return voiceRecoveryScope(hint); } catch { return undefined; } }); - } - function next() { - let nextAt: number | null = null; - const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; - 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 (!unavailable.some(scope => matches(entry, scope))) - take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); - return nextAt; - } - const worker = coordinator.register(DELIVERY_QUEUES.voiceSessionRecovery, { - retryMs: 30_000, - pausePassiveMaintenance: true, - onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); }, - run: async signal => { - if (!canRun() || Date.now() < earliest) return; - earliest = Date.now() + 1000; - const errors: unknown[] = []; - 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.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; } - const hint = requested[index]; - let scope: VoiceRecoveryScope | undefined; - let decoded = false; - try { - scope = voiceRecoveryScope(hint); decoded = true; - const rows = await service.load(scope, signal), retained = new Set(rows.map(row => row.id)); - for (const entry of entries.values()) if (matches(entry, scope) && !retained.has(entry.id)) { - entries.delete(entry.id); failures.delete(entry.id); - } - for (const entry of rows) entries.set(entry.id, entry); - 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 => !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)); - const batch = takeCompanyWorkBatch(due, 25, lastCompany); lastCompany = batch.lastCompany; - for (const entry of batch.entries) { - if (signal.aborted || !canRun()) break; - try { - const current = await service.check(entry.id, signal); - if (current) entries.set(entry.id, current); else entries.delete(entry.id); - failures.delete(entry.id); - } catch (error) { failed(failures, entry.id); errors.push(error); } - } - if (errors.length) throw new AggregateError(errors, "Voice session recovery could not settle"); - }, - hasPending: async () => dirty.size > 0, - nextRunAt: next, - }); - return worker; + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.voiceSessionRecovery, + decodeScope: voiceRecoveryScope, + matches, + load: (scope, signal) => service.load(scope, signal), + check: (entry, signal) => service.check(entry.id, signal), + }, canRun); } diff --git a/server/src/services/voice/voice-reply-work.ts b/server/src/services/voice/voice-reply-work.ts index a9917518814..1703414d182 100644 --- a/server/src/services/voice/voice-reply-work.ts +++ b/server/src/services/voice/voice-reply-work.ts @@ -1,4 +1,4 @@ -import { takeCompanyWorkBatch } from "../company-work-batch.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 type { VoiceReplyEntry, voiceReplySchedule } from "./voice-reply-schedule.js"; @@ -12,74 +12,11 @@ export function registerVoiceReplyWork( service: ReturnType, canRun: () => boolean, ) { - const dirty = new Set(["*"]); - const entries = new Map(); - const failures = new Map(); - const scopeFailures = new Map(); - let earliest = 0; - let lastCompany: string | null = null; - function failed(map: typeof failures, id: string) { - const count = Math.min((map.get(id)?.count ?? 0) + 1, 5); - map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) }); - } - function dirtyScopes() { - return [...dirty].map(hint => { try { return voiceReplyScope(hint); } catch { return undefined; } }); - } - function next() { - let nextAt: number | null = null; - const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; - 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 (!unavailable.some(scope => matches(entry, scope))) - take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); - return nextAt; - } - const worker = coordinator.register(DELIVERY_QUEUES.voiceReplies, { - retryMs: 30_000, - pausePassiveMaintenance: true, - onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); }, - run: async signal => { - if (!canRun() || Date.now() < earliest) return; - earliest = Date.now() + 1000; - const errors: unknown[] = []; - 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.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; } - const hint = requested[index]; - let scope: VoiceReplyScope | undefined; - let decoded = false; - try { - scope = voiceReplyScope(hint); decoded = true; - const rows = await service.load(scope, signal), retained = new Set(rows.map(row => row.id)); - for (const entry of entries.values()) if (matches(entry, scope) && !retained.has(entry.id)) { - entries.delete(entry.id); failures.delete(entry.id); - } - for (const entry of rows) entries.set(entry.id, entry); - 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 => !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)); - const batch = takeCompanyWorkBatch(due, 25, lastCompany); lastCompany = batch.lastCompany; - for (const entry of batch.entries) { - if (signal.aborted || !canRun()) break; - try { - const current = await service.check(entry, signal); - if (current) entries.set(entry.id, current); else entries.delete(entry.id); - failures.delete(entry.id); - } catch (error) { failed(failures, entry.id); errors.push(error); } - } - if (errors.length) throw new AggregateError(errors, "Voice reply delivery could not settle"); - }, - hasPending: async () => dirty.size > 0, - nextRunAt: next, - }); - return worker; + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.voiceReplies, + decodeScope: voiceReplyScope, + matches, + load: (scope, signal) => service.load(scope, signal), + check: (entry, signal) => service.check(entry, signal), + }, canRun); }