diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index 7b8ef115430..12f8dada5c4 100644 --- a/doc/idle-sleep-safety.md +++ b/doc/idle-sleep-safety.md @@ -81,6 +81,17 @@ Independent SQL writers and additional processes need an explicit wake protocol; process-local notifications do not provide cross-replica delivery. Worker quiet alone is not evidence that the host or database actually suspended. +## Native reply recovery + +Committed native replies use the shared delivery coordinator and its persisted +retry deadlines. One pass checks at most four independent tasks at a time. +Checks for the same company and task run in candidate order within the batch. +Other scoped deadline workers remain serial unless their owner enables concurrency. +Drain retains unstarted checks, and shutdown waits for every admitted check before +releasing tracked work. Completed history creates no recurring discovery reads; +new committed work reactivates the existing notification path. Pending reply +obligations keep their existing sleep blockers. + ## Scheduled routine wake protocol Historical failed ephemeral leases and terminal custom-image setup sessions can diff --git a/server/src/__tests__/scoped-deadline-work.test.ts b/server/src/__tests__/scoped-deadline-work.test.ts index fda914dd13e..7c6792c287b 100644 --- a/server/src/__tests__/scoped-deadline-work.test.ts +++ b/server/src/__tests__/scoped-deadline-work.test.ts @@ -37,3 +37,166 @@ it("preserves a subsecond replacement deadline when the owner does not require p expect(check).toHaveBeenCalledOnce(); expect(onError).not.toHaveBeenCalled(); }); + +type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; + +function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) { + vi.useFakeTimers(); + const onError = vi.fn(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + let canRun = true; + let releaseAll = false; + const releases = new Map void>(); + const signals: AbortSignal[] = []; + const started: string[] = []; + const active = new Set(); + const completed = new Set(); + let maximumActive = 0; + const check = vi.fn(async (entry: Entry, signal?: AbortSignal) => { + started.push(entry.id); + signals.push(signal!); + active.add(entry.id); + maximumActive = Math.max(maximumActive, active.size); + if (!releaseAll) await new Promise(resolve => releases.set(entry.id, resolve)); + active.delete(entry.id); + completed.add(entry.id); + return null; + }); + const load = vi.fn(async () => entries.filter(entry => !completed.has(entry.id))); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, onError }); + const options = { + queue: DELIVERY_QUEUES.nativeResponsePresentation, + concurrentChecks: concurrent ? { limit, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` } : undefined, + decodeScope: () => undefined, + matches: () => true, + load, + check, + }; + const worker = registerScopedDeadlineWork(coordinator, options, () => canRun); + return { + worker, check, load, started, active, signals, onError, entries, + maximumActive: () => maximumActive, + pause: () => { canRun = false; }, + resume: () => { canRun = true; worker.wake(); }, + release: (id: string) => releases.get(id)?.(), + releaseAll: () => { releaseAll = true; for (const resolve of releases.values()) resolve(); }, + }; +} + +const entry = (id: string, issueId = id, companyId = "company"): Entry => ({ id, issueId, companyId, dueAt: 0 }); + +it("checks independent tasks concurrently with a bounded number of active checks", async () => { + const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index)))); + try { + await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 }); + expect(f.maximumActive()).toBe(4); + f.release("1"); + await vi.waitFor(() => expect(f.started).toHaveLength(5), { timeout: 100 }); + expect(f.active.has("0")).toBe(true); + expect(f.maximumActive()).toBe(4); + } finally { f.releaseAll(); await f.worker.ready; } + expect(f.started).toHaveLength(6); + expect(new Set(f.started).size).toBe(6); + expect(coordinator!.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(60_000); + expect(f.load).toHaveBeenCalledOnce(); + expect(f.check).toHaveBeenCalledTimes(6); + f.entries.push(entry("fresh")); + f.worker.invalidate(); + await vi.advanceTimersByTimeAsync(1); + expect(f.load).toHaveBeenCalledTimes(2); + expect(f.check).toHaveBeenCalledTimes(7); + expect(f.started.at(-1)).toBe("fresh"); + expect(coordinator!.nextWakeAt()).toBeNull(); + expect(f.onError).not.toHaveBeenCalled(); +}); + +it("serializes each task in candidate order while other tasks and companies progress", async () => { + const f = concurrentFixture([ + entry("1", "shared"), entry("2", "shared"), entry("3", "other"), entry("4", "shared", "other-company"), + ]); + try { + await vi.waitFor(() => expect(new Set(f.started)).toEqual(new Set(["1", "3", "4"])), { timeout: 100 }); + f.release("3"); + await vi.waitFor(() => expect(f.active.has("3")).toBe(false), { timeout: 100 }); + expect(f.started).not.toContain("2"); + f.release("1"); + await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 }); + expect(f.started.filter(id => id === "1" || id === "2")).toEqual(["1", "2"]); + expect(f.active.has("1")).toBe(false); + } finally { f.releaseAll(); await f.worker.ready; } + expect(f.maximumActive()).toBe(3); + expect(f.onError).not.toHaveBeenCalled(); +}); + +it("preserves global candidate order and serial dispatch unless concurrency is enabled", async () => { + const f = concurrentFixture([entry("1", "shared"), entry("2", "other"), entry("3", "shared")], 4, false); + try { + await vi.waitFor(() => expect(f.started).toEqual(["1"]), { timeout: 100 }); + f.release("1"); + await vi.waitFor(() => expect(f.started).toEqual(["1", "2"]), { timeout: 100 }); + } finally { f.releaseAll(); await f.worker.ready; } + expect(f.started).toEqual(["1", "2", "3"]); + expect(f.maximumActive()).toBe(1); +}); + +it("retains unstarted checks through drain and resumes without idle discovery reads", async () => { + const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index)))); + try { + await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 }); + f.pause(); + } finally { f.releaseAll(); await f.worker.ready; } + expect(f.started).toHaveLength(4); + await vi.advanceTimersByTimeAsync(60_000); + expect(f.load).toHaveBeenCalledOnce(); + expect(f.started).toHaveLength(4); + f.resume(); + await vi.advanceTimersByTimeAsync(1); + expect(f.started).toHaveLength(6); + expect(coordinator!.nextWakeAt()).toBeNull(); + expect(f.onError).not.toHaveBeenCalled(); +}); + +it("stops starting checks and awaits every admitted check during shutdown", async () => { + const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index)))); + let stopped = false; + let stop: Promise | undefined; + try { + await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 }); + stop = coordinator!.stop().then(() => { stopped = true; }); + expect(f.signals.every(signal => signal.aborted)).toBe(true); + expect(stopped).toBe(false); + f.release("0"); + await vi.waitFor(() => expect(f.active.has("0")).toBe(false), { timeout: 100 }); + expect(stopped).toBe(false); + expect(f.started).toHaveLength(4); + } finally { f.releaseAll(); await f.worker.ready; await stop; } + expect(stopped).toBe(true); + expect(f.started).toHaveLength(4); +}); + +it("keeps failures isolated and retries their deadline without repeating successful checks", async () => { + vi.useFakeTimers(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + const onError = vi.fn(); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError }); + const entries = [entry("1", "shared"), entry("2", "shared"), entry("3")]; + const load = vi.fn(async () => entries); + const check = vi.fn(async (_entry: Entry) => null).mockRejectedValueOnce(new Error("temporary failure")); + const options = { + queue: DELIVERY_QUEUES.nativeResponsePresentation, + concurrentChecks: { limit: 4, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` }, + decodeScope: () => undefined, matches: () => true, load, check, + }; + const worker = registerScopedDeadlineWork(coordinator, options, () => true); + await worker.ready; + expect(check.mock.calls.map(([entry]) => entry.id).sort()).toEqual(["1", "2", "3"]); + expect(onError).toHaveBeenCalledOnce(); + expect(onError.mock.calls[0][0]).toBeInstanceOf(AggregateError); + await vi.advanceTimersByTimeAsync(29_999); + expect(check).toHaveBeenCalledTimes(3); + await vi.advanceTimersByTimeAsync(1); + expect(check.mock.calls.map(([entry]) => entry.id)).toEqual(["1", "3", "2", "1"]); + expect(load).toHaveBeenCalledOnce(); + expect(coordinator.nextWakeAt()).toBeNull(); +}); diff --git a/server/src/services/native-runtime/native-response-presentation-work.ts b/server/src/services/native-runtime/native-response-presentation-work.ts index cad63e0f53c..25a2c65a4dd 100644 --- a/server/src/services/native-runtime/native-response-presentation-work.ts +++ b/server/src/services/native-runtime/native-response-presentation-work.ts @@ -56,6 +56,7 @@ export function nativeResponsePresentationWork(db: Db) { function register(coordinator: ReturnType, canRun: () => boolean) { const worker = registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.nativeResponsePresentation, + concurrentChecks: { limit: 4, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` }, decodeScope: (hint: string): NativeResponsePresentationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), matches: (entry: Entry, scope?: NativeResponsePresentationScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId) && (!scope.agentId || entry.agentId === scope.agentId), load: (scope, signal) => load(scope, signal, canRun), diff --git a/server/src/services/presentation-ownership-read.test.ts b/server/src/services/presentation-ownership-read.test.ts index c08037e78af..2059284758a 100644 --- a/server/src/services/presentation-ownership-read.test.ts +++ b/server/src/services/presentation-ownership-read.test.ts @@ -10,6 +10,7 @@ import { getConversationOwnershipBlocker } from "./conversation-continuation.js" import { isExternalChatPresentationContext } from "./heartbeat-run-summary.js"; import { finalizeNativeRun, recoverCommittedNativeChatResponse, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js"; import { nativeResponsePresentationWork } from "./native-runtime/native-response-presentation-work.js"; +import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; @@ -24,10 +25,12 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read }, 60_000); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); - async function fixture(patch: Partial = {}) { - const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); - await db.insert(companies).values({ id: companyId, name: "Read fixture", issuePrefix: companyId.slice(0, 8) }); - await db.insert(agents).values({ id: agentId, companyId, name: "Read agent", adapterType: "codex_local" }); + async function fixture(patch: Partial = {}, identity?: { companyId: string; agentId: string }) { + const companyId = identity?.companyId ?? randomUUID(), agentId = identity?.agentId ?? randomUUID(), issueId = randomUUID(); + if (!identity) { + await db.insert(companies).values({ id: companyId, name: "Read fixture", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Read agent", adapterType: "codex_local" }); + } await db.insert(issues).values({ id: issueId, companyId, title: "Read task", status: "blocked", assigneeAgentId: agentId }); const payload = randomBytes(128 * 1024).toString("base64"); const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, nativeIssueId: issueId, @@ -138,10 +141,10 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read expect(await repairCommittedNativeChatResponse(db, { companyId: f.companyId, issueId: other.id, runId: f.run.id })).toBe(false); }); - async function unpublishedAcceptedReply() { + async function unpublishedAcceptedReply(id?: string, identity?: { companyId: string; agentId: string }) { const payload = randomBytes(128 * 1024).toString("base64"); - const f = await fixture({ runtimeMode: "native", status: "running", finishedAt: null, - contextSnapshot: { source: "internal", prompt: payload, paperclipWake: { prompt: payload } } }); + const f = await fixture({ id, runtimeMode: "native", status: "running", finishedAt: null, + contextSnapshot: { source: "internal", prompt: payload, paperclipWake: { prompt: payload } } }, identity); const [contract] = await db.insert(completionContracts).values({ companyId: f.companyId, issueId: f.issueId, revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", @@ -173,6 +176,40 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read return { ...f, before, reply }; } + it("saves an independent task reply while an earlier task waits for its database lock", async () => { + const suffix = randomUUID().replaceAll("-", "").slice(0, 12); + const first = await unpublishedAcceptedReply(`00000000-0000-4000-8000-${suffix}`); + const second = await unpublishedAcceptedReply(`00000000-0000-4001-8000-${suffix}`, first); + let release!: () => void; + let locked!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const acquired = new Promise(resolve => { locked = resolve; }); + const blocker = db.transaction(async tx => { + await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, first.run.id)).for("update"); + locked(); await hold; + }); + await Promise.race([acquired, blocker]); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + try { + await vi.waitFor(async () => { + const rows = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, second.run.id)); + expect(rows).toHaveLength(1); + expect(rows[0].body).toBe(second.reply); + }, { timeout: 5000 }); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, first.run.id))).toHaveLength(0); + } finally { + release(); await blocker; await worker.ready; worker.stop(); await coordinator.stop(); + } + for (const f of [first, second]) { + const rows = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id)); + expect(rows).toHaveLength(1); + expect(rows[0].body).toBe(f.reply); + } + expect(onError).not.toHaveBeenCalled(); + }); + async function repairFixture(patch: Partial) { diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index b46841e2a91..8c94a863ccd 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -12,6 +12,7 @@ export function registerScopedDeadlineWork( queue: DeliveryQueue; batchSize?: number; minimumPassIntervalMs?: number; + concurrentChecks?: { limit: number; key(entry: Entry): string }; decodeScope(hint: string): Scope | undefined; matches(entry: Entry, scope?: Scope): boolean; load(scope?: Scope, signal?: AbortSignal): Promise; @@ -19,6 +20,9 @@ export function registerScopedDeadlineWork( }, canRun: () => boolean, ) { + if (options.concurrentChecks && (!Number.isInteger(options.concurrentChecks.limit) || options.concurrentChecks.limit < 1)) { + throw new RangeError("Concurrent check limit must be a positive integer"); + } const dirty = new Set(["*"]); const versions = new Map(); const scans = new Map }>(); @@ -94,14 +98,30 @@ export function registerScopedDeadlineWork( && 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, options.batchSize ?? 25, lastCompany); lastCompany = batch.lastCompany; + const groups = new Map(); 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); } + const key = options.concurrentChecks?.key(entry) ?? ""; + const group = groups.get(key) ?? []; + group.push(entry); groups.set(key, group); } + const pending = [...groups.values()]; + let nextGroup = 0; + const checkGroups = async () => { + while (!signal.aborted && canRun()) { + const group = pending[nextGroup++]; + if (!group) return; + for (const entry of group) { + if (signal.aborted || !canRun()) return; + 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); } + } + } + }; + const checks = await Promise.allSettled(Array.from({ length: Math.min(options.concurrentChecks?.limit ?? 1, pending.length) }, checkGroups)); + for (const result of checks) if (result.status === "rejected") errors.push(result.reason); if (errors.length) throw new AggregateError(errors, `${options.queue} could not settle`); }, // Retain scopes/deadlines for resume without retaining a settled write's