diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 13f3868869a..e8c98c6f6e5 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2660,6 +2660,13 @@ An upgrade readiness report contains preliminary native evidence, not permission to remove a deployment volume. Keep the checkpoint pool bounded and measure all pools when qualifying database cost; the current size of two is a default. +Committed native Agent Chat turns recover their unanswered final reply through +the existing response presentation worker. Recovery verifies the accepted result +and final turn, agent, responsible user, native session and conversation generation. +It stores the exact final message before the normal conversation settlement moves +the chat to waiting. Later user input remains active. Recovery does not run the +agent again, replace a selected reply or restore a deleted reply. + For retained Codex event tails, keep source drain/accounting and target admission separate. `native-staged-attachment-recovery.ts` validates private source/target header evidence; `native-recovery-usage-delivery.ts` binds the source's accounting diff --git a/doc/plans/2026-10-30-run-dispatch-sleep.md b/doc/plans/2026-10-30-run-dispatch-sleep.md new file mode 100644 index 00000000000..8ad35c128fc --- /dev/null +++ b/doc/plans/2026-10-30-run-dispatch-sleep.md @@ -0,0 +1,43 @@ +# Run dispatch without idle retry scans + +The run-dispatch module owns retry admission, its durable `scheduled_retry_at` +deadline, and atomic promotion. The heartbeat queue owns concurrency, current +execution authority, and dispatch. The shared coordinator owns notification +coalescing, nonoverlap, transaction uncertainty, drain/resume and shutdown. +The heartbeat facade and `index.ts` only connect these owners. + +A scheduled retry is a future obligation. Startup reconstructs its next stored +deadline through the run-dispatch adapter. Writes that create, move, cancel or +invalidate retries notify the retry worker in their transaction. Promotion uses +the existing gate and compare-and-set, and its committed write wakes ordinary +queued dispatch. A queued run behind occupied capacity waits for the owning +run's lifecycle transition. Configuration and authority changes also wake it. +Retry-schedule notifications are separate from ordinary queue eligibility. +`notifyRunDispatch` wakes only queued dispatch; ordinary issue, agent, budget +and company edits cannot repeatedly scan the global retry schedule. +`notifyScheduledRetryDispatch` wakes retry reconciliation as well as dispatch +for actual retry creation, rescheduling and cancellation. Its producers include +bounded and operator-requested retries, recovery retries, native capacity and +safe-replacement successors, and the existing retry-cancellation transition +boundary. Shared runtime changes do not impose a spacing floor on a saved deadline. + +An idle queue has no recurring read or timer-driven dispatch attempt. Notifications batch for one second and carry the changed company to queued dispatch. Known retry deadlines still preempt batching. Cancellation-owned retries stay with their cancellation worker and do not retain a promotion deadline. + +The supported writer contract is service writes through the shared database +connection scope, including caller-owned and nested transactions. Existing +explicit domain notification boundaries cover lifecycle, issue review state, +agent configuration/status, company lifecycle and scheduling settings. Direct +SQL writers and independent replicas must provide an explicit ingress signal or +restart reconstruction; a process-local signal does not observe their commits. +This change adds no permanent database listener or write interception. + +The remaining periodic recovery chain still owns deferred comment receipts, +execution-wait cancellation recovery, orphan/resource recovery, stranded issue +assignments, dependency wake repair, task watchdogs, silent active runs and stale +locks. Those duties retain their backstop until their separate producer/deadline +migrations are qualified. This slice removes only the periodic scheduled-retry +promotion call. Ordinary queued dispatch can still be reached by that backstop; +the existing atomic claim and per-agent start lock prevent duplicate execution. + +These workers use internal process deadlines. They do not register Cloud wake +alarms and do not relax the durable idle-sleep safety proof. diff --git a/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts b/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts index 23a465d4b56..7d7d89ecd80 100644 --- a/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts @@ -47,6 +47,9 @@ async function openOwnedWire(url: string): Promise<{ wire: PrpWireConnection; cl const size = initial === 126 ? buffer.readUInt16BE(2) : initial === 127 ? Number(buffer.readBigUInt64BE(2)) : initial; const offset = initial === 126 ? 4 : initial === 127 ? 10 : 2; if (buffer.length < offset + size) return; + // Upgrade head bytes can precede the caller's message subscription. + // Retain complete data frames until the control-plane listener exists. + if ((opcode === 1 || opcode === 2) && listeners.length === 0) return; const payload = buffer.subarray(offset, offset + size); buffer = buffer.subarray(offset + size); if (opcode === 8) { socket.destroy(); finish(); return; } if (opcode === 9) send(payload, 10); @@ -56,12 +59,45 @@ async function openOwnedWire(url: string): Promise<{ wire: PrpWireConnection; cl socket.on("data", bytes => { buffer = Buffer.concat([buffer, bytes]); parse(); }); socket.on("close", finish); socket.on("error", finish); const close = () => { socket.destroy(); finish(); }; resolvePeer({ wire: { sendJson: value => send(Buffer.from(JSON.stringify(value)), 1), close, - onJson: listener => { listeners.push(listener); }, onClose: listener => { closes.push(() => listener({})); } }, closed, close }); + onJson: listener => { listeners.push(listener); parse(); }, onClose: listener => { closes.push(() => listener({})); } }, closed, close }); queueMicrotask(parse); }); req.end(); }); } +it("delivers a WebSocket message coalesced with the real HTTP upgrade response", async () => { + const message = { type: "hello", sentinel: "coalesced-upgrade" }; + const server = createServer(); + const sockets = new Set(); + server.on("connection", socket => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); }); + server.on("upgrade", (req, socket) => { + const accept = createHash("sha1").update(String(req.headers["sec-websocket-key"]) + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").digest("base64"); + const payload = Buffer.from(JSON.stringify(message)); + socket.write(Buffer.concat([Buffer.from(`HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`), + Buffer.from([0x81, payload.length]), payload])); + }); + let peer: Awaited> | undefined; + const messages: unknown[] = []; + try { + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { server.off("error", reject); resolve(); }); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing coalesced-upgrade endpoint"); + peer = await openOwnedWire(`ws://127.0.0.1:${address.port}/`); + peer.wire.onJson(value => messages.push(value)); + await vi.waitFor(() => expect(messages).toEqual([message]), { timeout: 1000 }); + } finally { + peer?.close(); + const closing = server.listening + ? new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())) + : Promise.resolve(); + for (const socket of sockets) socket.destroy(); + await closing; + } +}); + // Real encrypted PRP and real Runner/provider processes; no model or server imports. it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every durable boundary without another process", async mode => { const root = await mkdtemp(join(tmpdir(), "runner-staged-handoff-")); @@ -226,6 +262,7 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d coreWarm: currentCore?.store.state.warmTransition?.phase, header: checkpoint && inspectStagedRunAttachmentHeader(checkpoint.records.get("header")!)?.phase, runnerRun: physical.runId, runnerWarm: physical.warmTransition?.phase, lifecycle: physical.lifecycle, failureCode: physical.recoverableFailure?.code, failureKeys: Object.keys(physical.recoverableFailure ?? {}), + physicalFailure: physical.recoverableFailure, })}`, { cause: error }); } finally { release(); diff --git a/server/src/__tests__/agent-timer-deadlines.test.ts b/server/src/__tests__/agent-timer-deadlines.test.ts index 4c4c0d324a9..f925fb8dfd7 100644 --- a/server/src/__tests__/agent-timer-deadlines.test.ts +++ b/server/src/__tests__/agent-timer-deadlines.test.ts @@ -597,8 +597,10 @@ describe.skipIf(!support.supported)("agent timer owned deadlines", () => { await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } } }); await vi.advanceTimersByTimeAsync(2000); - - expect(idleWorkSnapshot().active).toBe(1); + // The producer committed: retain its resume deadline without keeping the + // instance busy throughout the drain. Uncertain writes still hold admission. + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now()); const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction"); try { await vi.advanceTimersByTimeAsync(60_000); @@ -608,6 +610,11 @@ describe.skipIf(!support.supported)("agent timer owned deadlines", () => { stopTaskDrain(); await vi.advanceTimersByTimeAsync(1000); await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + await vi.advanceTimersByTimeAsync(coordinator!.nextWakeAt()! - Date.now()); + await vi.waitFor(() => expect(f.enqueueWakeup).toHaveBeenCalledOnce()); + expect(f.enqueueWakeup).toHaveBeenCalledWith(f.agent.id, expect.objectContaining({ source: "timer", reason: "heartbeat_timer" })); + await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: false } } }); + await quiet(); }); it("does not inspect timer rows during startup drain, then reconstructs them on resume", async () => { diff --git a/server/src/__tests__/deadline-work.test.ts b/server/src/__tests__/deadline-work.test.ts index 7b852977345..c00daeb2f00 100644 --- a/server/src/__tests__/deadline-work.test.ts +++ b/server/src/__tests__/deadline-work.test.ts @@ -1,4 +1,5 @@ -import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { installDatabaseWorkSignals, signalDatabaseWork, databaseWorkPending } from "../../../packages/db/src/work-signals.js"; import { afterEach, expect, it, vi } from "vitest"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { registerDeadlineWork } from "../services/deadline-work.js"; @@ -39,3 +40,132 @@ it.each([ expect(run).toHaveBeenCalledTimes(2); expect(onError).not.toHaveBeenCalled(); }); + +it("releases a settled first-write hold during drain and retains a probe-free resume wake", async () => { + vi.useFakeTimers(); + let draining = false; + const owner = installDatabaseWorkSignals({ + execute: vi.fn(async () => [{ status: "committed" }]), + transaction: async (callback: (tx: object) => Promise) => callback({ execute: async () => [{ xid: "1" }] }), + }); + const run = vi.fn(async () => {}), next = vi.fn(async () => null); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError: vi.fn() }); + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !draining, retryMs: 30_000 }); + await worker.ready; + draining = true; + const before = idleWorkSnapshot().active; + await owner.transaction(async tx => { await signalDatabaseWork(tx, DELIVERY_QUEUES.agentTimers); }); + await vi.advanceTimersByTimeAsync(0); + expect(idleWorkSnapshot().active).toBe(before); + expect(coordinator.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(3 * 30_000); + expect(run).toHaveBeenCalledOnce(); expect(next).toHaveBeenCalledOnce(); expect(owner.execute).not.toHaveBeenCalled(); + draining = false; + await vi.advanceTimersByTimeAsync(30_000); + expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull(); +}); + +it("retains failed work through drain and preserves a future domain deadline", async () => { + vi.useFakeTimers(); + let draining = false; + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + const run = vi.fn(async () => { if (run.mock.calls.length === 1) { draining = true; throw new Error("retry debt"); } }); + const deadline = Date.now() + 600_000; + const next = vi.fn(async () => deadline); + const onError = vi.fn(); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError }); + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !draining, retryMs: 30_000 }); + await worker.ready; + await vi.advanceTimersByTimeAsync(90_000); + expect(run).toHaveBeenCalledOnce(); expect(next).not.toHaveBeenCalled(); expect(owner.execute).not.toHaveBeenCalled(); + draining = false; await vi.advanceTimersByTimeAsync(30_000); + expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBe(deadline); + draining = true; await vi.advanceTimersByTimeAsync(30_000); + expect(coordinator.nextWakeAt()).toBe(deadline); expect(onError).toHaveBeenCalledOnce(); +}); + +it("does no domain probes in warm standby and reconstructs after admission opens", async () => { + vi.useFakeTimers(); + let standby = true; + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + const run = vi.fn(async () => {}), next = vi.fn(async () => null); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !standby, canReconcile: () => !standby, onError: vi.fn() }); + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !standby, retryMs: 30_000 }); + await worker.ready; await vi.advanceTimersByTimeAsync(90_000); + expect(run).not.toHaveBeenCalled(); expect(next).not.toHaveBeenCalled(); expect(owner.execute).not.toHaveBeenCalled(); + standby = false; await vi.advanceTimersByTimeAsync(30_000); + expect(run).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull(); +}); + +it("does not release an ambiguous database intent while dispatch is drained", async () => { + vi.useFakeTimers(); + let draining = false, committed = false; + const owner = installDatabaseWorkSignals({ + execute: vi.fn(async () => [{ status: committed ? "committed" : "in progress" }]), + transaction: async (callback: (tx: object) => Promise) => { + await callback({ execute: async () => [{ xid: "2" }] }); + throw new Error("lost acknowledgement"); + }, + }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError: vi.fn() }); + const run = vi.fn(async () => {}); + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next: async () => null, canRun: () => !draining, retryMs: 30_000 }); + await worker.ready; draining = true; + const before = idleWorkSnapshot().active; + await expect(owner.transaction(async tx => { await signalDatabaseWork(tx, DELIVERY_QUEUES.agentTimers); })).rejects.toThrow("lost acknowledgement"); + await vi.advanceTimersByTimeAsync(0); + expect(idleWorkSnapshot().active).toBe(before + 1); + expect(databaseWorkPending(owner, DELIVERY_QUEUES.agentTimers)).toBe(true); + committed = true; await vi.advanceTimersByTimeAsync(30_000); + expect(databaseWorkPending(owner, DELIVERY_QUEUES.agentTimers)).toBe(false); + expect(idleWorkSnapshot().active).toBe(before); + draining = false; await vi.advanceTimersByTimeAsync(30_000); + expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull(); +}); + +it("backs off an expired deadline when only the domain is drained and resumes promptly", async () => { + vi.useFakeTimers(); + let paused = false; + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + const run = vi.fn(async () => {}); + const domainCanRun = vi.fn(() => !paused); + const deadline = Date.now() + 1_000; + const next = vi.fn(async () => run.mock.calls.length === 1 ? deadline : null); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() }); + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { + run, next, canRun: domainCanRun, retryMs: 30_000, + }); + await worker.ready; + paused = true; + const before = idleWorkSnapshot(); + await vi.advanceTimersByTimeAsync(1_000); + expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000); + const attempts = domainCanRun.mock.calls.length; + await vi.advanceTimersByTimeAsync(90_000); + expect(domainCanRun.mock.calls.length - attempts).toBeLessThanOrEqual(12); + // Four attempts (expiry plus three retries), each begins and ends tracked work. + expect(idleWorkSnapshot().generation - before.generation).toBe(8); + expect(idleWorkSnapshot().active).toBe(before.active); + expect(run).toHaveBeenCalledOnce(); expect(next).toHaveBeenCalledOnce(); expect(owner.execute).not.toHaveBeenCalled(); + paused = false; + await vi.advanceTimersByTimeAsync(30_000); + expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull(); +}); + +it("retains a prompt paused wake when a new request preempts a far-future deadline", async () => { + vi.useFakeTimers(); + let paused = false; + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + const run = vi.fn(async () => {}); + const deadline = Date.now() + 86_400_000; + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() }); + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { + run, next: async () => run.mock.calls.length === 1 ? deadline : null, canRun: () => !paused, retryMs: 30_000, + }); + await worker.ready; paused = true; worker.wake(); + await vi.advanceTimersByTimeAsync(0); + expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000); + expect(owner.execute).not.toHaveBeenCalled(); + paused = false; await vi.advanceTimersByTimeAsync(30_000); + expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull(); +}); diff --git a/server/src/__tests__/execution-control-work.test.ts b/server/src/__tests__/execution-control-work.test.ts index 0046d8574aa..14cd9456eb5 100644 --- a/server/src/__tests__/execution-control-work.test.ts +++ b/server/src/__tests__/execution-control-work.test.ts @@ -383,11 +383,11 @@ describe.skipIf(!support.supported)("execution-control durable work", () => { const due = new Date(Date.now() + 120_000), earlier = new Date(Date.now() + 60_000); for (const at of [due, earlier]) { await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns) - .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id))); + .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id)), { companyId }); await settle(); expect(c.nextWakeAt()).toBe(at.getTime()); } await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns) - .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id))); + .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id)), { companyId }); await quiet(c); }); it("delivers a reconciled continuation when marking and resolving commit together", async () => { diff --git a/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts new file mode 100644 index 00000000000..a9e120387c8 --- /dev/null +++ b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts @@ -0,0 +1,128 @@ +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { PgDialect } from "drizzle-orm/pg-core"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyRunDispatch, notifyScheduledRetryDispatch } from "../services/run-dispatch-notifications.js"; +import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js"; + +const cleanup: Array<() => Promise> = []; +afterEach(async () => { + for (const stop of cleanup.splice(0)) await stop(); + vi.useRealTimers(); +}); +function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean } = {}) { + vi.useFakeTimers(); + vi.setSystemTime(1_000_000); + const dialect = new PgDialect(); + const execute = vi.fn(async (query: any) => { + if (dialect.sqlToQuery(query).sql.includes("pg_current_xact_id")) return [{ xid: "42" }]; + throw new Error("unexpected database query"); + }); + const db: { execute: typeof execute; transaction: (callback: (tx: any) => Promise) => Promise } = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise): Promise { + return callback({ execute, transaction: db.transaction }); + } }); + const state = { due: input.due ?? null as number | null, suppressed: input.suppressed ?? false, + draining: input.draining ?? false, standby: false, fail: false }; + const promote = vi.fn(async () => { + if (state.fail) throw new Error("database unavailable"); + if (state.due !== null && state.due <= Date.now()) state.due = null; + }); + const next = vi.fn(async () => state.due); + const dispatch = vi.fn(async (_scope?: { companyId: string }) => {}); + const coordinator = createDeliveryWorkCoordinator({ owner: db, + canRun: () => !state.draining && !state.standby, canReconcile: () => !state.standby, onError: vi.fn() }); + const worker = registerRunDispatchWork(coordinator, { + promoteDueScheduledRetries: promote, nextScheduledRetryAt: next, + resumeQueuedRunsAfterDrain: dispatch, schedulingSuppressed: async () => state.suppressed, + }, () => !state.draining && !state.standby); + cleanup.push(() => coordinator.stop()); + return { db, state, promote, next, dispatch, coordinator, worker, + notifyRetry: () => db.transaction(tx => notifyScheduledRetryDispatch(tx)), + notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)), + settle: () => vi.advanceTimersByTimeAsync(1_000) }; +} + +describe("run dispatch obligations", () => { + it.each([false, true])("settles empty/feature-suppressed state and stops all owner work (suppressed=%s)", async suppressed => { + const f = fixture({ suppressed }); await f.worker.ready; + const calls = [f.promote.mock.calls.length, f.next.mock.calls.length, f.dispatch.mock.calls.length]; + expect(f.coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(20 * 60_000); + expect([f.promote.mock.calls.length, f.next.mock.calls.length, f.dispatch.mock.calls.length]).toEqual(calls); + expect(f.db.execute).not.toHaveBeenCalled(); + }); + it("restores an exact deadline and does no earlier recovery work", async () => { + const f = fixture({ due: 1_010_000 }); await f.worker.ready; + expect(f.coordinator.nextWakeAt()).toBe(1_010_000); + await vi.advanceTimersByTimeAsync(9_999); expect(f.promote).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); expect(f.promote).toHaveBeenCalledTimes(2); + expect(f.state.due).toBeNull(); expect(f.coordinator.nextWakeAt()).toBeNull(); + expect(f.dispatch).toHaveBeenCalledOnce(); + }); + it("rearms earlier work and cancels a future deadline on a committed producer change", async () => { + const f = fixture({ due: 1_600_000 }); await f.worker.ready; + f.state.due = 1_005_000; await f.notifyRetry(); await f.settle(); + expect(f.coordinator.nextWakeAt()).toBe(1_005_000); + f.state.due = null; await f.notifyRetry(); await f.settle(); + expect(f.coordinator.nextWakeAt()).toBeNull(); + const before = f.next.mock.calls.length; + await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.next).toHaveBeenCalledTimes(before); + }); + it("batches repeated tenant writes and passes only their company to queued dispatch", async () => { + const f = fixture(); await f.worker.ready; + for (let index = 0; index < 100; index++) { + const source = { companyId: "company-a", issueId: `issue-${index % 2}`, + runId: `run-${index}`, agentId: `agent-${index}`, interactionBody: "not a dispatch key" }; + await f.notify(source); + await vi.advanceTimersByTimeAsync(10); + } + expect(f.promote).toHaveBeenCalledOnce(); + expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2); + await f.settle(); + expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" }); + for (const [scope] of f.dispatch.mock.calls.slice(1)) expect(scope).toEqual({ companyId: "company-a" }); + expect(f.coordinator.nextWakeAt()).toBeNull(); + const count = f.dispatch.mock.calls.length; + await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.dispatch).toHaveBeenCalledTimes(count); + }); + it("remembers the first producer arriving during drain without owner SQL", async () => { + const f = fixture(); await f.worker.ready; f.state.draining = true; + await f.notifyRetry(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000); + expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce(); + expect(idleWorkSnapshot().active).toBe(0); + f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000); + expect(f.promote).toHaveBeenCalledTimes(2); expect(f.dispatch).toHaveBeenCalledTimes(2); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("releases a failed producer hold during drain and retains recovery for resume", async () => { + const f = fixture(); await f.worker.ready; + f.state.fail = true; await f.notifyRetry(); await f.settle(); + expect(f.promote).toHaveBeenCalledTimes(2); + f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000); + expect(f.promote).toHaveBeenCalledTimes(2); expect(f.next).toHaveBeenCalledOnce(); + expect(idleWorkSnapshot().active).toBe(0); + f.state.fail = false; f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000); + expect(f.promote).toHaveBeenCalledTimes(3); expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("preserves a notification during a nonoverlapping pass", async () => { + const f = fixture(); await f.worker.ready; + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + f.promote.mockImplementationOnce(() => held); + await f.notifyRetry(); await f.settle(); + await f.notifyRetry(); await f.settle(); + expect(f.promote).toHaveBeenCalledTimes(2); + release(); await f.settle(); + await vi.advanceTimersByTimeAsync(30_000); + expect(f.promote).toHaveBeenCalledTimes(3); expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("does not dispatch on warm standby and cancels deadlines at shutdown", async () => { + const f = fixture({ due: 1_010_000 }); f.state.standby = true; await f.worker.ready; + await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).not.toHaveBeenCalled(); + f.state.standby = false; await vi.advanceTimersByTimeAsync(30_000); + expect(f.promote).toHaveBeenCalledOnce(); expect(f.state.due).toBeNull(); + f.state.due = Date.now() + 60_000; await f.notifyRetry(); await f.settle(); await f.worker.stop(); + await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2); + }); +}); diff --git a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts new file mode 100644 index 00000000000..1c0e5005a04 --- /dev/null +++ b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts @@ -0,0 +1,284 @@ +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns, issues } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { issueService } from "../services/issues.js"; +import { materializeNativeProviderCapacityRetry } from "../services/native-runtime/native-provider-capacity-retry.js"; +import { agentService } from "../services/agents.js"; +import { budgetService } from "../services/budgets.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { notifyRunDispatch } from "../services/run-dispatch-notifications.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RUN_DISPATCH_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); + +describe.skipIf(!support.supported)("run dispatch production wiring (controlled local process adapter)", () => { + let database: Awaited> | undefined; + let db: ReturnType; + let heartbeat: ReturnType; + let coordinator: ReturnType | undefined; + let worker: ReturnType["registerRunDispatchWork"]> | undefined; + let reads: ReturnType; + let draining = false; + const errors: unknown[] = []; + const retryQueries: string[] = []; + beforeAll(async () => { + if (!external) database = await startEmbeddedPostgresTestDatabase("run-dispatch-work-"); + db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + db.$client.options.debug = (_connection, query) => { + if (query.startsWith("select") && query.includes('"heartbeat_runs"') && + query.includes("startupCancellation") && query.includes('"scheduled_retry_at"')) retryQueries.push(query); + }; + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + }, 90_000); + beforeEach(async () => { + await db.execute(sql`TRUNCATE companies CASCADE`); + heartbeat = heartbeatService(db); draining = false; errors.length = 0; + reads = vi.spyOn(db, "select"); + }); + afterEach(async () => { + await worker?.stop(); await coordinator?.stop(); + await heartbeat.drainActiveRunExecutions(); + worker = undefined; coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); + + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + worker = heartbeat.registerRunDispatchWork(coordinator, () => !draining); + await worker.ready; + } + async function fixture() { + const companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Dispatch", issuePrefix: `D${companyId.slice(0, 7)}`, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "dispatch-user" }); + const [agent] = await db.insert(agents).values({ companyId, name: "Controlled process", role: "engineer", status: "idle", + adapterType: "process", adapterConfig: { command: process.execPath, args: ["-e", "console.log('dispatch-proof')"], cwd: process.cwd() }, + runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } }, + }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, status: "failed", + responsibleUserId: "dispatch-user", errorCode: "codex_transient_upstream", finishedAt: new Date(), + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, contextSnapshot: {}, + }).returning(); + return { companyId, agentId: agent.id, runId: run.id }; + } + async function quiet() { + await heartbeat.drainActiveRunExecutions(); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull(), { timeout: 5_000 }); + const count = reads.mock.calls.length; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + try { + await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(reads).toHaveBeenCalledTimes(count); + } finally { vi.useRealTimers(); } + } + async function status(id: string) { + return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]); + } + async function completed(id: string) { + await vi.waitFor(async () => expect(["running", "succeeded"]).toContain((await status(id))?.status), { timeout: 10_000 }); + await heartbeat.drainActiveRunExecutions(); + const completed = await status(id); + expect(completed.status).toBe("succeeded"); + expect(completed.stdoutExcerpt).toContain("dispatch-proof"); + } + it("keeps fresh empty, configured idle and retained terminal history quiet", async () => { + await start(); await quiet(); + await fixture(); await worker!.stop(); await coordinator!.stop(); + await start(); await quiet(); + }); + it.each(["agent", "issue"] as const)("does not scan global scheduled retries for repeated ordinary %s edits", async kind => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Edit source", status: "backlog" }).returning(); + await start(); await quiet(); + const before = retryQueries.length; + expect(before).toBeGreaterThan(0); // The passive observer saw real startup retry SQL. + for (let index = 0; index < 12; index++) { + if (kind === "agent") await agentService(db).update(f.agentId, { name: `Edited agent ${index}` }); + else await issueService(db).update(issue.id, { title: `Edited issue ${index}` }); + await new Promise(resolve => setTimeout(resolve, 100)); + } + await quiet(); + expect(retryQueries).toHaveLength(before); + }); + it.each([false, true])("observes a native capacity retry only after its transaction settles (rollback=%s)", async rollback => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Capacity retry", status: "in_progress", + assigneeAgentId: f.agentId }).returning(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.runId)); + await start(); await quiet(); + let retryId = ""; + const writing = db.transaction(async tx => { + retryId = await materializeNativeProviderCapacityRetry({ tx: tx as unknown as typeof db, ...f, issueId: issue.id }); + expect(databaseWorkPending(db, DELIVERY_QUEUES.scheduledRetryDispatch)).toBe(true); + expect(await status(retryId)).toBeUndefined(); + if (rollback) throw new Error("native capacity rollback"); + }); + if (rollback) { + await expect(writing).rejects.toThrow("native capacity rollback"); + await quiet(); expect(await status(retryId)).toBeUndefined(); + } else { + await writing; + const retry = await status(retryId); + expect(retry.status).toBe("scheduled_retry"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(retry.scheduledRetryAt!.getTime()), { timeout: 5_000 }); + await heartbeat.cancelRun(retryId, "fixture teardown", { suppressImmediateRecovery: true }); + await quiet(); + } + }); + it("ordinary retry creation wakes the worker, respects the exact deadline, dispatches and settles", async () => { + const f = await fixture(); await start(); await quiet(); + const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 4_000, random: () => 0 }); + expect(result.outcome).toBe("scheduled"); + if (result.outcome !== "scheduled") throw new Error("Expected retry creation"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(result.dueAt.getTime()), { timeout: 5_000 }); + expect((await status(result.run.id)).status).toBe("scheduled_retry"); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(Date.now()).toBeLessThan(result.dueAt.getTime()); + expect((await status(result.run.id)).status).toBe("scheduled_retry"); + await completed(result.run.id); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.retryOfRunId, f.runId)))).toHaveLength(1); + await quiet(); + }); + it("reconstructs an overdue retry after process restart without a periodic tick", async () => { + const f = await fixture(); + const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { now: new Date(Date.now() - 120_000), delayMs: 60_000 }); + if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation"); + heartbeat = heartbeatService(db); + await start(); await completed(scheduled.run.id); await quiet(); + }); + it("cancels a future retry and removes its old deadline", async () => { + const f = await fixture(); await start(); + const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 600_000 }); + if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()), { timeout: 5_000 }); + await heartbeat.cancelRun(scheduled.run.id, "operator stop", { suppressImmediateRecovery: true }); + await quiet(); expect((await status(scheduled.run.id)).status).toBe("cancelled"); + }); + it("retains the first retry written during drain and catches it up after resume", async () => { + const f = await fixture(); await start(); await quiet(); draining = true; + const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 200 }); + if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation"); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + const count = reads.mock.calls.length; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + try { + await vi.advanceTimersByTimeAsync(90_000); + expect(reads).toHaveBeenCalledTimes(count); + expect(idleWorkSnapshot().active).toBe(0); + } finally { vi.useRealTimers(); } + expect((await status(scheduled.run.id)).status).toBe("scheduled_retry"); + draining = false; worker!.wake(); + await completed(scheduled.run.id); await quiet(); + }); + it("uses a committed budget raise to resume a queued run in the company it paused", async () => { + const f = await fixture(); + const budgets = budgetService(db); + await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, amount: 100, + metric: "billed_cents", windowKind: "lifetime", hardStopEnabled: true }, "dispatch-user"); + const [cost] = await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agentId, + provider: "fixture", biller: "fixture", model: "controlled", inputTokens: 0, outputTokens: 0, + costCents: 125, occurredAt: new Date() }).returning(); + await budgets.evaluateCostEvent(cost); + expect((await db.select().from(companies).where(eq(companies.id, f.companyId)))[0].status).toBe("paused"); + const [queued] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "queued", responsibleUserId: "dispatch-user", invocationSource: "automation", contextSnapshot: {} }).returning(); + await start(); await quiet(); expect((await status(queued.id)).status).toBe("queued"); + await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, amount: 10_000, + metric: "billed_cents", windowKind: "lifetime", hardStopEnabled: true }, "dispatch-user"); + await completed(queued.id); await quiet(); + }); + it("clearing an agent error wakes saved queue work only after its own transaction commits", async () => { + const f = await fixture(); await start(); await quiet(); + await db.update(agents).set({ status: "error" }).where(eq(agents.id, f.agentId)); + const [queued] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "queued", responsibleUserId: "dispatch-user", invocationSource: "automation", contextSnapshot: {} }).returning(); + let writing: ReturnType["clearError"]> | undefined; + try { + await db.transaction(async blocker => { + await blocker.select().from(agents).where(eq(agents.id, f.agentId)).for("update"); + writing = agentService(db).clearError(f.agentId); + void writing.catch(() => {}); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(databaseWorkPending(db, DELIVERY_QUEUES.queuedRunDispatch)).toBe(false); + expect((await status(queued.id)).status).toBe("queued"); + expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].status).toBe("error"); + }); + } finally { await writing; } + await completed(queued.id); await quiet(); + }); + it("leaves cancellation-owned retries quiet even when their old promotion time is overdue", async () => { + const f = await fixture(); + const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { now: new Date(Date.now() - 60_000), delayMs: 1_000 }); + if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation"); + await db.update(heartbeatRuns).set({ resultJson: { startupCancellation: { requestedAt: new Date().toISOString() }, + cancellation: { expected: true, reason: "operator stop", source: "operator" } } }).where(eq(heartbeatRuns.id, scheduled.run.id)); + await start(); await quiet(); + expect((await status(scheduled.run.id)).status).toBe("scheduled_retry"); + expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] }); + await quiet(); + }); + it("rejects an error clear without publishing work when the lifecycle is no longer ready", async () => { + const f = await fixture(); await start(); await quiet(); + await db.update(agents).set({ status: "error", lifecycleState: "paused" }).where(eq(agents.id, f.agentId)); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + try { + await expect(agentService(db).clearError(f.agentId)).rejects.toThrow("Only agents in error status"); + expect(coordinator!.nextWakeAt()).toBeNull(); + const count = reads.mock.calls.length; + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).toHaveBeenCalledTimes(count); + } finally { vi.useRealTimers(); } + expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].status).toBe("error"); + await quiet(); + }); + it("keeps a foreign company's future retry intact while concurrent workers claim one due retry", async () => { + const selected = await fixture(), foreign = await fixture(); await start(); + const later = await heartbeat.scheduleBoundedRetry(foreign.runId, { delayMs: 600_000 }); + const due = await heartbeat.scheduleBoundedRetry(selected.runId, { delayMs: 500 }); + if (later.outcome !== "scheduled" || due.outcome !== "scheduled") throw new Error("Expected retries"); + const other = heartbeatService(db); + const competing = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const competingWorker = other.registerRunDispatchWork(competing, () => true); + try { + await competingWorker.ready; await completed(due.run.id); await other.drainActiveRunExecutions(); + expect((await status(later.run.id)).status).toBe("scheduled_retry"); + expect((await status(later.run.id)).scheduledRetryAt).toEqual(later.dueAt); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, selected.runId))).toHaveLength(1); + await heartbeat.cancelRun(later.run.id, "fixture teardown", { suppressImmediateRecovery: true }); + await quiet(); + } finally { await competingWorker.stop(); await competing.stop(); await other.drainActiveRunExecutions(); } + }); + it("holds nested producer work until outer commit and rolls back without a successor", async () => { + const f = await fixture(); await start(); await quiet(); + let release!: () => void; const held = new Promise(resolve => { release = resolve; }); + let queuedId = ""; + const writing = db.transaction(async tx => { + await tx.transaction(async nested => { + await notifyRunDispatch(nested); + const [row] = await nested.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + responsibleUserId: "dispatch-user", status: "queued", contextSnapshot: {} }).returning(); + queuedId = row.id; + }); + await held; + }); + await vi.waitFor(() => expect(queuedId).not.toBe("")); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(await status(queuedId)).toBeUndefined(); + release(); await writing; await completed(queuedId); await quiet(); + await expect(db.transaction(async tx => { + await tx.transaction(async nested => { await notifyRunDispatch(nested); await nested.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "queued" }); }); + throw new Error("rollback outer"); + })).rejects.toThrow("rollback outer"); + await quiet(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(2); + }); +}); diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index 210a88a8fd7..d617f8cdb0e 100644 --- a/server/src/__tests__/managed-skill-content.test.ts +++ b/server/src/__tests__/managed-skill-content.test.ts @@ -1,3 +1,5 @@ +import { createServer } from "node:http"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js"; import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js"; import { eq } from "drizzle-orm"; @@ -77,6 +79,93 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () .toMatchObject({ content: "complete root guide" }); }); + it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => { + const companyId = await company(); + const objects = new Map(); + const server = createServer((request, response) => { + void (async () => { + const key = new URL(request.url!, "http://fixture").pathname; + if (request.method === "PUT") { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + let body = Buffer.concat(chunks); + if (request.headers["content-encoding"]?.includes("aws-chunked")) { + const decoded: Buffer[] = []; + let offset = 0; + while (offset < body.length) { + const end = body.indexOf("\r\n", offset); + const size = Number.parseInt(body.subarray(offset, end).toString().split(";")[0]!, 16); + if (size === 0) break; + decoded.push(body.subarray(end + 2, end + 2 + size)); offset = end + 2 + size + 2; + } + body = Buffer.concat(decoded); + } + objects.set(key, body); response.writeHead(200, { etag: '"fixture"' }).end(); return; + } + const body = objects.get(key); + if (!body) { response.writeHead(404).end(); return; } + response.writeHead(200, { "content-length": body.length, "content-type": "application/octet-stream", etag: '"fixture"' }); + response.end(request.method === "HEAD" ? undefined : body); + })().catch(() => { response.writeHead(500).end(); }); + }); + await new Promise(resolve => { server.listen(0, "127.0.0.1", resolve); }); + const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected fixture address"); + const endpoint = `http://127.0.0.1:${address.port}`; + const credentials = { AWS_ACCESS_KEY_ID: "fixture-access-key", AWS_SECRET_ACCESS_KEY: "fixture-secret-key", AWS_SESSION_TOKEN: "" }; + const previousCredentials = Object.fromEntries(Object.keys(credentials).map(key => [key, process.env[key]])); + Object.assign(process.env, credentials); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createS3StorageProvider({ + bucket: "managed-skill-fixture", region: "us-east-1", endpoint, forcePathStyle: true, + }))); + const key = "cloud-runtime-config:v1"; + try { + await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const run = (phase: string) => promisify(execFile)(process.execPath, [ + "--import", fileURLToPath(import.meta.resolve("tsx")), + fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)), + `--phase=${phase}`, `--company=${companyId}`, "--limit=50", + ], { env: { ...process.env, ...credentials, DATABASE_URL: database.connectionString, + PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", + PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64), PAPERCLIP_STORAGE_PROVIDER: "s3", + PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-fixture", PAPERCLIP_STORAGE_S3_REGION: "us-east-1", + PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "" }, + timeout: 60_000, maxBuffer: 1024 * 1024 }); + await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration"); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0); + await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }); + const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!); + expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]); + const before = [...objects.entries()].map(([key, body]) => [key, Buffer.from(body)] as const); + const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(verification.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()]).toEqual(before); + const archiveKey = [...objects.keys()].find(key => key.endsWith("/files.tar.gz"))!; + const archive = objects.get(archiveKey)!; + const corrupted = Buffer.from(archive); corrupted[0] = corrupted[0]! ^ 1; + objects.set(archiveKey, corrupted); + const corrupt = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(corrupt.results).toEqual([expect.objectContaining({ status: "corrupt_package" })]); + objects.set(archiveKey, archive); + const manifestKey = [...objects.keys()].find(key => key.endsWith("/manifest.json"))!; + const manifest = objects.get(manifestKey)!; objects.delete(manifestKey); + const unavailable = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(unavailable.results).toEqual([expect.objectContaining({ status: "unavailable_or_corrupt_package" })]); + objects.set(manifestKey, manifest); + const recovered = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); + expect(recovered.results).toEqual([expect.objectContaining({ status: "remote_storage" })]); + expect([...objects.entries()].sort()).toEqual([...before].sort()); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))) + .toEqual([expect.objectContaining({ general: { version: 1, revision: 0, config: null } })]); + } finally { + await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key)); + vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createLocalDiskStorageProvider(path.join(root, "objects")))); + for (const [key, value] of Object.entries(previousCredentials)) { + if (value === undefined) delete process.env[key]; else process.env[key] = value; + } + server.closeAllConnections(); await new Promise(resolve => { server.close(() => resolve()); }); + } + }, 180_000); + it.each([ { storage: "legacy local", env: { PAPERCLIP_STORAGE_PROVIDER: "local_disk" }, expectedStatus: "local_storage" }, { storage: "shared S3", env: { PAPERCLIP_STORAGE_PROVIDER: "s3", diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bbb020c865d..e00c8519df7 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -134,6 +134,7 @@ const { })), reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })), reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), + registerRunDispatchWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerRetainedNativeSessionCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), @@ -923,6 +924,18 @@ describe("startServer feedback export wiring", () => { if (enabled) expect(heartbeatServiceMock.registerCapturedInstructionCommitWork.mock.calls[0][1]()).toBe(true); }); + it.each([false, true])("registers retry and queued dispatch only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) { + const [coordinator, admitted] = heartbeatServiceMock.registerRunDispatchWork.mock.calls[0]; + const app = await createAppMock.mock.results.at(-1)!.value; + expect(coordinator).toBe(app.locals.deliveryWork); + expect(admitted()).toBe(true); + } + }); + it.each([false, true])("registers retry cancellation recovery only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -1328,6 +1341,7 @@ describe("startServer feedback export wiring", () => { expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); + expect(heartbeatServiceMock.promoteDueScheduledRetries).not.toHaveBeenCalled(); expect(retiredDetector).not.toHaveBeenCalled(); } finally { delete (runtime as Partial).reconcileProductivityReviews; diff --git a/server/src/index.ts b/server/src/index.ts index c127753889c..cd8bb951946 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1778,7 +1778,6 @@ async function startServerWithDatabaseTeardown( } } - const promotion = await timeStartupRecovery("scheduled_retries", () => heartbeat.promoteDueScheduledRetries()); await timeStartupRecovery("queued_runs", () => heartbeat.resumeQueuedRuns()); const recoveredGoalActions = await timeStartupRecovery("goal_actions", () => heartbeat.recoverPendingSessionGoalActions()); if ( @@ -1799,7 +1798,6 @@ async function startServerWithDatabaseTeardown( } const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); if ( - promotion.promoted > 0 || reconciled.assignmentDispatched > 0 || reconciled.dispatchRequeued > 0 || reconciled.continuationRequeued > 0 || @@ -1808,7 +1806,7 @@ async function startServerWithDatabaseTeardown( reconciled.escalated > 0 ) { logger.warn( - { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled }, + { ...reconciled }, "startup heartbeat recovery changed assigned issue state", ); } @@ -1847,6 +1845,8 @@ async function startServerWithDatabaseTeardown( await startupHeartbeatRecovery; } + if (config.heartbeatSchedulerEnabled) await heartbeat.registerRunDispatchWork(deliveryWork, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; @@ -2049,12 +2049,10 @@ async function startServerWithDatabaseTeardown( // persisted queued work is still being driven forward. runPeriodicRecovery(() => heartbeat .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }) - .then(() => heartbeat.promoteDueScheduledRetries()) - .then(async (promotion) => { + .then(async () => { await heartbeat.resumeQueuedRuns(); const reconciled = await heartbeat.reconcileStrandedAssignedIssues({ skipExternalConversationSettlement: true }); if ( - promotion.promoted > 0 || reconciled.assignmentDispatched > 0 || reconciled.dispatchRequeued > 0 || reconciled.continuationRequeued > 0 || @@ -2063,7 +2061,7 @@ async function startServerWithDatabaseTeardown( reconciled.escalated > 0 ) { logger.warn( - { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled }, + { ...reconciled }, "periodic heartbeat recovery changed assigned issue state", ); } diff --git a/server/src/modules/agent-lifecycle/adapters/records.ts b/server/src/modules/agent-lifecycle/adapters/records.ts index 1c59758dd91..96892987b01 100644 --- a/server/src/modules/agent-lifecycle/adapters/records.ts +++ b/server/src/modules/agent-lifecycle/adapters/records.ts @@ -80,22 +80,20 @@ export function agentRecords(db: Db, effects: LifecycleEffects) { throw conflict("Only agents in error status can have their error cleared"); } - const updated = await db - .update(agents) - .set({ - status: "idle", - pauseReason: null, - pausedAt: null, - errorReason: null, - updatedAt: new Date(), - }) - .where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready"))) - .returning() - .then((rows) => rows[0] ?? null); + const updated = await effects.transaction(db, existing.companyId, async tx => { + const [current] = await tx.select({ status: agents.status, lifecycleState: agents.lifecycleState }) + .from(agents).where(and(eq(agents.id, id), eq(agents.companyId, existing.companyId))).for("no key update"); + if (current?.status !== "error" || current.lifecycleState !== "ready") { + throw conflict("Only agents in error status can have their error cleared"); + } + await effects.beforeStatusChange(tx, existing.companyId, id, "error", "idle"); + const [saved] = await tx.update(agents).set({ + status: "idle", pauseReason: null, pausedAt: null, errorReason: null, updatedAt: new Date(), + }).where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready"))).returning(); + if (!saved) throw conflict("Only agents in error status can have their error cleared"); + return saved; + }); - if (!updated) { - throw conflict("Only agents in error status can have their error cleared"); - } return getById(updated.id); }, diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index cf9946bafe9..b05f94e1a4b 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -511,6 +511,15 @@ export function createPostgresRunDispatchAdapter( return decideScheduledRetryGate(factsResult.facts, input.now); } + async function nextScheduledRetryAt(input: { cutoff: Date | null }): Promise { + const [row] = await db.select({ at: heartbeatRuns.scheduledRetryAt }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.status, "scheduled_retry"), + sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') is distinct from 'object'`, + input.cutoff ? gte(heartbeatRuns.createdAt, input.cutoff) : undefined)) + .orderBy(asc(heartbeatRuns.scheduledRetryAt)).limit(1); + return row?.at?.getTime() ?? null; + } + async function listDueRetries(input: ListDueRetriesInput): Promise { const rows = await db .select() @@ -518,6 +527,7 @@ export function createPostgresRunDispatchAdapter( .where( and( eq(heartbeatRuns.status, "scheduled_retry"), + sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') is distinct from 'object'`, lte(heartbeatRuns.scheduledRetryAt, input.now), input.cutoff ? gte(heartbeatRuns.createdAt, input.cutoff) : undefined, ), @@ -1414,6 +1424,7 @@ export function createPostgresRunDispatchAdapter( return { evaluateScheduledRetryGate, + nextScheduledRetryAt, listDueRetries, cancelStaleQueuedRun, dispatchResolvedInteractionIfCurrent, diff --git a/server/src/modules/run-dispatch/application/ports.ts b/server/src/modules/run-dispatch/application/ports.ts index 83e41fd3da6..be6faf39ca7 100644 --- a/server/src/modules/run-dispatch/application/ports.ts +++ b/server/src/modules/run-dispatch/application/ports.ts @@ -24,6 +24,7 @@ export type EvaluateScheduledRetryGateInput = { /** Read-only scheduled-retry operations exposed to application use cases. */ export interface ScheduledRetryReader { + nextScheduledRetryAt(input: { cutoff: Date | null }): Promise; evaluateScheduledRetryGate(input: EvaluateScheduledRetryGateInput): Promise; listDueRetries(input: ListDueRetriesInput): Promise; } diff --git a/server/src/modules/run-dispatch/application/use-cases.test.ts b/server/src/modules/run-dispatch/application/use-cases.test.ts index 58e5ca6f8df..1cce6637ba0 100644 --- a/server/src/modules/run-dispatch/application/use-cases.test.ts +++ b/server/src/modules/run-dispatch/application/use-cases.test.ts @@ -12,6 +12,7 @@ function fakeReader(dueRuns: DueRetryRun[] = []): ScheduledRetryReader & { evalu const evaluateCalls: unknown[] = []; return { evaluateCalls, + async nextScheduledRetryAt() { return null; }, async evaluateScheduledRetryGate(input) { evaluateCalls.push(input); return { allowed: true }; diff --git a/server/src/modules/run-dispatch/index.ts b/server/src/modules/run-dispatch/index.ts index 5293ca0ae64..d44d559618a 100644 --- a/server/src/modules/run-dispatch/index.ts +++ b/server/src/modules/run-dispatch/index.ts @@ -67,6 +67,7 @@ export function createRunDispatch(db: Db, deps: RunDispatchDeps = {}) { }); return { + nextScheduledRetryAt: (input: { cutoff: Date | null }) => adapter.nextScheduledRetryAt(input), evaluateScheduledRetryGate: createEvaluateScheduledRetryGate({ reader: adapter }), promoteScheduledRetry, promoteDueScheduledRetries: createPromoteDueScheduledRetries({ diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index b888ee51fb3..1df4420615f 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "../../../services/run-dispatch-notifications.js"; import { notifyExternalConversationSettlement } from "../../../services/external-conversation-settlement-notifications.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; @@ -637,6 +638,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .returning() .then((rows) => rows[0]); + await notifyRunDispatch(tx, { companyId }); const queuedRun = await tx .insert(heartbeatRuns) .values({ @@ -715,6 +717,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .returning() .then((rows) => rows[0]); + await notifyRunDispatch(tx, { companyId }); const queuedRun = await tx .insert(heartbeatRuns) .values({ diff --git a/server/src/services/agent-configuration.ts b/server/src/services/agent-configuration.ts index 4139dc6dfe0..dc10efaa082 100644 --- a/server/src/services/agent-configuration.ts +++ b/server/src/services/agent-configuration.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifySkillVersionPinChange } from "./skill-version-retention-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; @@ -83,6 +84,7 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura existing.adapterConfig, ); } + await notifyRunDispatch(db, { companyId: existing.companyId }); if (normalizedPatch.runtimeConfig !== undefined || normalizedPatch.reportsTo !== undefined) { await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); } diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 0a6dcaf0fc9..59357fcc5de 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { isAgentStatusInvokable } from "@paperclipai/shared"; import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; @@ -38,7 +39,10 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li }, notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle), async beforeStatusChange(tx, companyId, agentId, before, after) { - if (before !== after) await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + if (before !== after) { + await notifyRunDispatch(tx, { companyId }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + } if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) { await notifyNativeResponsePresentation(tx, { companyId, agentId }); await notifyNativeReviewResponse(tx, { companyId, agentId }); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 3a15d489f0a..21ae37eda03 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyApprovalWork } from "./approval-work-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { addCents, compareCents, normalizeCents } from "@paperclipai/shared"; @@ -300,6 +301,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } if (policy.scopeType === "project") { + await notifyRunDispatch(db, { companyId: policy.companyId }); await db .update(projects) .set({ @@ -313,6 +315,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; + await notifyRunDispatch(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); @@ -359,6 +362,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } if (policy.scopeType === "project") { + await notifyRunDispatch(db, { companyId: policy.companyId }); await db .update(projects) .set({ @@ -373,6 +377,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; + await notifyRunDispatch(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 73535f5434a..524936395b5 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyIssueReviewState } from "./issue-review-state-notifications.js"; import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; @@ -11,6 +12,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type ChatRunMilestoneScope = { companyId: string; issueId?: string; runId?: string }; export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { + if (!scope.issueId && !scope.runId) await notifyRunDispatch(tx, scope); if (scope.issueId || scope.runId) { await notifyIssueReviewState(tx, { companyId: scope.companyId, issueId: scope.issueId, runId: scope.issueId ? undefined : scope.runId }); diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 0f2ed547a79..a9a8f59998c 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -332,6 +333,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { const renamedPrefix = await resolveRenamedIssuePrefix(tx, id, companyPatch); if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { + await notifyRunDispatch(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); @@ -488,6 +490,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { const wasAlreadyArchived = existing.status === "archived"; if (!wasAlreadyArchived) { + await notifyRunDispatch(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts index 306f340d9fc..208b208e051 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -26,7 +26,7 @@ export function registerDeadlineWork( pausePassiveMaintenance: true, onSettled: hints => { requested++; task.onSettled?.(hints); }, run: async signal => { - if (!task.canRun()) { requested++; next = null; return; } + if (!task.canRun()) { requested++; return; } if (Date.now() < earliest) { next = earliest; return; } if (Date.now() < coalesceUntil && (next === null || next > Date.now())) { next = next === null ? coalesceUntil : Math.min(next, coalesceUntil); @@ -39,8 +39,6 @@ export function registerDeadlineWork( if (signal.aborted || !task.canRun()) { next = null; return; } const deadline = await task.next(); - - if (!task.canRun()) { next = null; return; } earliest = deadline === null || deadline > Date.now() ? (task.minimumSpacingMs ? Date.now() + task.minimumSpacingMs : 0) @@ -49,8 +47,13 @@ export function registerDeadlineWork( coalesceUntil = Date.now() + (task.notificationCoalesceMs ?? 0); reconciled = generation; }, - hasPending: async () => requested !== reconciled, - nextRunAt: () => next, + // A settled producer must release its admission hold during drain. Keep + // the unreconciled generation and an in-memory resume wake, without asking + // the domain to probe its database while dispatch is closed. + hasPending: async () => task.canRun() && requested !== reconciled, + nextRunAt: () => !task.canRun() && requested !== reconciled + ? Date.now() + task.retryMs + : next, }); return { ...worker, wake: () => { requested++; worker.wake(); } }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9195f1fcb12..7388bdb8861 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,8 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + scheduledRetryDispatch: "scheduled-retry-dispatch", + queuedRunDispatch: "queued-run-dispatch", runEvidenceMigration: "run-evidence-migration", providerTraceRetention: "provider-trace-retention", nativeRunnerEvidenceRetention: "native-runner-evidence-retention", diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 22c8913bbcd..1839176769f 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -1,7 +1,11 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js"; -export async function notifyExecutionRunWork(transaction: object) { +export type ExecutionWorkScope = { companyId: string; agentId?: string; runId?: string }; + +export async function notifyExecutionRunWork(transaction: object, scope?: ExecutionWorkScope) { + if (scope) await notifyRunDispatch(transaction, scope); for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) { await notifyDeliveryWork(transaction, queue); @@ -12,8 +16,10 @@ export function writeExecutionControl( db: Db | Parameters[0]>[0], queue: DeliveryQueue | readonly DeliveryQueue[], write: (tx: Db) => PromiseLike, + scope: ExecutionWorkScope | undefined, ): Promise { return db.transaction(async transaction => { + await notifyRunDispatch(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors); for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic); return write(transaction as unknown as Db); diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index 2f2a435982d..621fd210cfc 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -196,7 +196,7 @@ export async function reconcileAbandonedExecutionControl( await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); const [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index f3f5a3e4310..d56e7d926f1 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -546,7 +546,7 @@ export async function settleUnrecoverableExecutions( companyId: run.companyId, issueId: task.id, runId: run.id, }).limit(1).for("update"); if (!hold) return; - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); const note = "Exact session recovery stopped. The recorded workspace recovery holds this execution without replaying its actions."; await tx.update(issueRecoveryActions).set({ status: "cancelled", outcome: "cancelled", resolvedAt: now, updatedAt: now, wakePolicy: null, monitorPolicy: null, @@ -654,7 +654,7 @@ export async function settleUnrecoverableExecutions( replay: "not_authorized", }, }); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); await tx .update(heartbeatRuns) .set({ executionStatusDeliveryId: randomUUID() }) diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 074ffc3bdf5..084a25e457f 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -475,7 +475,7 @@ export async function admitExplicitNativeContinuation(input: { }).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.queuedCommentInterruptId), eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"))); if (cancelledStartupIds.size) await db.transaction(async tx => { - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId, agentId }); for (const runId of cancelledStartupIds) { await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 4b449d2477d..0fdefcf3109 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js"; import { NATIVE_RETAINED_OWNER_WAIT } from "./native-runtime/native-controller-recovery.js"; import { externalConversationSettlementWork } from "./external-conversation-settlement-work.js"; import { nativeRecoveryNoticeWork } from "./native-runtime/native-recovery-notice-work.js"; @@ -1003,6 +1004,7 @@ export function heartbeatService( finalizeAiConnectionBusyDeferral, finalizeWorkspaceBusyDeferral, promoteDueScheduledRetries, + nextScheduledRetryAt, retryScheduledRetryNow, scheduleBoundedRetry, } = createHeartbeatRetries(db, { @@ -1337,11 +1339,11 @@ export function heartbeatService( return heartbeatQueue.dispatchPendingNativeStatusWakeups(...args); } - function resumeQueuedRunsAfterDrain() { + function resumeQueuedRunsAfterDrain(...args: Parameters) { for (const pending of nativeRecoveryDeadlines.values()) { pending.rearm(); } - const promise = heartbeatQueue.resumeQueuedRunsAfterDrain(); + const promise = heartbeatQueue.resumeQueuedRunsAfterDrain(...args); activeWakeupPromises.add(promise); void promise.then( () => { activeWakeupPromises.delete(promise); }, @@ -3712,7 +3714,7 @@ export function heartbeatService( await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx .update(heartbeatRuns) .set({ executionControlDeadlineAt: null }) - .where(eq(heartbeatRuns.id, run.id))); + .where(eq(heartbeatRuns.id, run.id)), { companyId: run.companyId }); } nativeOwnershipHeld = nativeOwnershipHeld || @@ -4234,6 +4236,13 @@ export function heartbeatService( computeTaskDrain, applyTaskDrain, + registerRunDispatchWork: (coordinator: Parameters[0], canRun: () => boolean) => + registerRunDispatchWork(coordinator, { + promoteDueScheduledRetries, + nextScheduledRetryAt, + resumeQueuedRunsAfterDrain, + schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed, + }, () => canRun() && readTaskDrain(new Date()) === null), promoteDueScheduledRetries, retryScheduledRetryNow, diff --git a/server/src/services/heartbeat/dispatch-work.ts b/server/src/services/heartbeat/dispatch-work.ts new file mode 100644 index 00000000000..5b1f728688a --- /dev/null +++ b/server/src/services/heartbeat/dispatch-work.ts @@ -0,0 +1,58 @@ +import type { RunDispatchScope } from "../run-dispatch-notifications.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { registerDeadlineWork } from "../deadline-work.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; + +export function registerRunDispatchWork( + coordinator: ReturnType, + owner: { + promoteDueScheduledRetries(): Promise; + nextScheduledRetryAt(): Promise; + resumeQueuedRunsAfterDrain(scope?: RunDispatchScope): Promise; + schedulingSuppressed(): Promise; + }, + canRun: () => boolean, +) { + let suppressed = false; + const retries = registerDeadlineWork(coordinator, DELIVERY_QUEUES.scheduledRetryDispatch, { + retryMs: 30_000, + continuationMs: 1_000, + notificationCoalesceMs: 1_000, + canRun, + run: async signal => { + suppressed = await owner.schedulingSuppressed(); + if (!signal.aborted && !suppressed && canRun()) await owner.promoteDueScheduledRetries(); + }, + next: () => suppressed ? Promise.resolve(null) : owner.nextScheduledRetryAt(), + }); + const scopes = new Set(["*"]); + const queue = registerDeadlineWork(coordinator, DELIVERY_QUEUES.queuedRunDispatch, { + retryMs: 30_000, + notificationCoalesceMs: 1_000, + canRun, + onSettled: hints => { for (const hint of hints ?? ["*"]) scopes.add(hint); }, + run: async signal => { + const requested = scopes.has("*") ? ["*"] : [...scopes]; + if (requested[0] === "*") scopes.clear(); + else for (const hint of requested) scopes.delete(hint); + for (let index = 0; index < requested.length; index++) { + if (signal.aborted || !canRun()) { + for (const hint of requested.slice(index)) scopes.add(hint); + return; + } + try { + await owner.resumeQueuedRunsAfterDrain(requested[index] === "*" ? undefined : JSON.parse(requested[index]) as RunDispatchScope); + } catch (error) { + for (const hint of requested.slice(index)) scopes.add(hint); + throw error; + } + } + }, + next: async () => null, + }); + return { + ready: Promise.all([retries.ready, queue.ready]), + wake() { scopes.add("*"); retries.wake(); queue.wake(); }, + async stop() { await Promise.all([retries.stop(), queue.stop()]); }, + }; +} diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index a41b486b647..64ca0b84ffd 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "../run-dispatch-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -1972,7 +1973,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // Release admission promptly without first scanning unrelated recovery work. // Every run still passes the ordinary per-agent claim and dispatch gates. - async function resumeQueuedRunsAfterDrain() { + async function resumeQueuedRunsAfterDrain(scope?: { companyId: string }) { if ((await getSchedulingSuppression()).suppressed) return; const cutoff = await getWorktreeExecutionCutoff(); const queuedRuns = await db @@ -1982,6 +1983,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende .where( and( eq(heartbeatRuns.status, "queued"), + scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, eq(companies.status, "active"), cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined, ), @@ -4632,6 +4634,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } const queueOutcome = await db.transaction(async (tx) => { + await notifyRunDispatch(tx, { companyId: agent.companyId }); await tx.execute( sql`select id from agents where id = ${agentId} and company_id = ${agent.companyId} for update`, ); diff --git a/server/src/services/heartbeat/retries.test.ts b/server/src/services/heartbeat/retries.test.ts index a557871c553..3d7e9a1ec34 100644 --- a/server/src/services/heartbeat/retries.test.ts +++ b/server/src/services/heartbeat/retries.test.ts @@ -35,6 +35,7 @@ function callbacks() { getWorktreeExecutionCutoff: vi.fn(async (): Promise => null), applyRunDispatchPostCommitEffects: vi.fn<(effects: PostCommitEffect[]) => void>(), runDispatch: { + nextScheduledRetryAt: vi.fn(async () => null), evaluateScheduledRetryGate: vi.fn(async () => ({ allowed: true })), promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [], postCommitEffects: [] })), promoteScheduledRetry: vi.fn(async () => ({ outcome: "not_promoted" })), @@ -183,7 +184,9 @@ describe("heartbeat retry module callbacks", () => { }); }); -const support = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HEARTBEAT_RETRIES_TEST_DATABASE_URL"); +const support = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; if (!support.supported) console.warn(`Skipping retry PostgreSQL tests: ${support.reason ?? "unsupported host"}`); @@ -193,11 +196,11 @@ describePostgres("heartbeat retry module database wiring", () => { const now = new Date("2026-10-08T12:00:00Z"); beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("heartbeat-retries-"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("heartbeat-retries-"); + db = createDb(external ?? database.connectionString, { maxConnections: 6 }); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); - afterAll(async () => { await database?.cleanup(); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); afterEach(async () => { await db.delete(issues); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 946f5e309f8..9eda1347ffa 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js"; import { infrastructureRetryDelayMs } from "../system-recovery-state.js"; import { @@ -77,7 +78,7 @@ type PlanApprovalResumeFailure = { export interface HeartbeatRetryDependencies { hasCaseHandoffForRun: (run: Pick) => Promise; runDispatch: Pick, - "evaluateScheduledRetryGate" | "promoteDueScheduledRetries" | "promoteScheduledRetry">; + "evaluateScheduledRetryGate" | "promoteDueScheduledRetries" | "promoteScheduledRetry" | "nextScheduledRetryAt">; getAgent: (agentId: string) => Promise; getRun: ( runId: string, @@ -997,6 +998,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen retryContextSnapshot.previousRunId = continuation.previousRunId; } + await notifyScheduledRetryDispatch(tx, { companyId: run.companyId }); const wakeupRequest = await tx .insert(agentWakeupRequests) .values({ @@ -1643,6 +1645,10 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen }); } + async function nextScheduledRetryAt() { + return runDispatch.nextScheduledRetryAt({ cutoff: await getWorktreeExecutionCutoff() }); + } + async function promoteDueScheduledRetries(now = new Date()) { const cutoff = await getWorktreeExecutionCutoff(); const result = await runDispatch.promoteDueScheduledRetries({ @@ -1750,6 +1756,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen }; const updated = await db.transaction(async (tx) => { + await notifyScheduledRetryDispatch(tx, { companyId: scheduled.run.companyId }); const row = await tx .update(heartbeatRuns) .set({ @@ -1901,6 +1908,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen finalizeAiConnectionBusyDeferral, finalizeWorkspaceBusyDeferral, promoteDueScheduledRetries, + nextScheduledRetryAt, retryScheduledRetryNow, scheduleBoundedRetry: async ( diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 00a27c1faf2..9d3b30367d2 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "../run-dispatch-notifications.js"; import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; @@ -1044,7 +1045,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl if (previous && isHeartbeatRunTerminalStatus(status)) await notifyRetainedNativeSessionCleanup(tx, { companyId: previous.companyId, runId: previous.id }); if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, companyId ? { companyId } : undefined); if (previous) await notifyChatRunMilestones(tx, { companyId: previous.companyId, runId: previous.id, issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined }); return write(tx); @@ -2447,6 +2448,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl .returning() .then((rows) => rows[0]); + await notifyRunDispatch(tx, { companyId: run.companyId }); const queuedRun = await tx .insert(heartbeatRuns) .values({ diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index 092639894cf..88c8c336539 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -983,7 +983,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "running"), ), - )); + ), { companyId: run.companyId }); const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded"; await recordWorkspaceFinalize(workspaceFinalizeStatus); if (adapterResult.nativeFinalization) { @@ -1063,7 +1063,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "running"), ), - )); + ), { companyId: run.companyId }); if ( issueRef && context.resumeSessionGoalHeartbeat === true && diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 0208653a86f..174c39d6125 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js"; import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; @@ -630,6 +631,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); if (patch.enableWorktreeRunExecution !== undefined) { + await notifyRunDispatch(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyExternalAgentSettlement(tx); await notifyInstructionCommit(tx); diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 5fb28bcb6d8..48fb001c0b4 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -392,7 +392,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx.update(issueRecoveryActions).set({ nextAction: input.nextAction, updatedAt: now }) .where(and(eq(issueRecoveryActions.id, existing.id), inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]), - sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning()); + sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning(), { companyId: input.companyId }); if (updated) return toReadModel(updated); } return existing; @@ -467,7 +467,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { ), ) .returning(); - }); + }, { companyId: input.companyId }); if (!exhausted) { return retryUpsertSourceScoped(input, retryCount); } @@ -539,7 +539,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { ), ) .returning(); - }); + }, { companyId: input.companyId }); if (!updated) { return retryUpsertSourceScoped(input, retryCount); } @@ -554,7 +554,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { .insert(issueRecoveryActions) .values(buildInsertValues(input, ownerType, now)) .returning(); - }); + }, { companyId: input.companyId }); return toReadModel(created!); } catch (error) { if (!isUniqueRecoveryActionConflict(error)) throw error; @@ -608,7 +608,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { }) .where(and(...predicates)) .returning(); - }); + }, { companyId: input.companyId }); return updated ? toReadModel(updated) : null; } diff --git a/server/src/services/issue-review-state-notifications.ts b/server/src/services/issue-review-state-notifications.ts index 679b7e37475..d9d4d756816 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/server/src/services/issue-review-state-notifications.ts @@ -1,9 +1,11 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) { + await notifyRunDispatch(tx, scope); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyExternalConversationSettlement(tx, scope); await notifyNativeReviewResponse(tx, scope); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 49cba7d020d..cc9293db472 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -10606,6 +10607,7 @@ export function issueService(db: Db) { : values.parentId ? `issue:${values.parentId}` : undefined); } if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyRunDispatch(tx, { companyId: values.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const [issue] = await tx.insert(issues).values(values).returning(); await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId }); @@ -11494,6 +11496,7 @@ export function issueService(db: Db) { } if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) { + await notifyRunDispatch(tx, { companyId: receiptExisting.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); } if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt && @@ -11958,6 +11961,7 @@ export function issueService(db: Db) { } await notifyDecisionTargetIssueRemoval(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyRunDispatch(tx, owner ? { companyId: owner.companyId } : undefined); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); removedIssue = await tx .delete(issues) @@ -12086,6 +12090,7 @@ export function issueService(db: Db) { : isNull(issues.executionRunId); const updated = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyRunDispatch(tx, { companyId: issueCompany.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx .update(issues) @@ -12215,6 +12220,7 @@ export function issueService(db: Db) { const staleExecutionRunId = current.executionRunId; const adopted = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyRunDispatch(tx, { companyId: issueCompany.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx .update(issues) @@ -12470,6 +12476,7 @@ export function issueService(db: Db) { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyRunDispatch(tx, { companyId: existing.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx .update(issues) @@ -12530,6 +12537,7 @@ export function issueService(db: Db) { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id }); } await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyRunDispatch(tx, { companyId: existing.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx .update(issues) diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 427704de63f..c462cdf1fca 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -135,7 +135,7 @@ export async function terminalizeLegacyExecution(input: { }; if (issueId) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: current.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id }); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: current.companyId }); let [updated] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/native-runtime/agent-chat-response-recovery.test.ts b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts new file mode 100644 index 00000000000..1c2cc4399a3 --- /dev/null +++ b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts @@ -0,0 +1,149 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, authUsers, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues, + nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; +import { finalizeNativeRun, recoverCommittedNativeChatResponse } from "./native-run-finalizer.js"; +import { nativeResponsePresentationWork } from "./native-response-presentation-work.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_CHAT_RESPONSE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => { + let database: Awaited> | undefined; + let db: Db; + beforeAll(async () => { + if (!external) database = await startEmbeddedPostgresTestDatabase("agent-chat-response-"); + db = createDb(external ?? database!.connectionString); + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function fixture(conversation = true) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const sessionId = randomUUID(), runnerId = randomUUID(), userId = `fixture-${randomUUID()}`, wakeId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Conversation owner", email: `${userId}@example.test`, + createdAt: new Date("2020-01-01T00:00:00Z"), updatedAt: new Date("2020-01-01T00:00:00Z") }); + await db.insert(companies).values({ id: companyId, name: "Reply recovery fixture", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Conversation agent", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Conversation reply", status: "in_progress", assigneeAgentId: agentId, + ...(conversation ? { conversationAgentId: agentId, conversationUserId: userId, conversationState: "active", conversationSessionGeneration: 0 } : {}) }); + await db.insert(issueComments).values({ id: wakeId, companyId, issueId, authorUserId: userId, body: "Hire one teammate and tell me when ready.", + createdAt: new Date("2020-01-01T00:00:00Z") }); + const [contract] = await db.insert(completionContracts).values({ companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", completionAuthority: "server_arbiter", + incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: { revision: "test", objective: "Answer the user", criteria: [] }, + canonicalSha256: "fixture-conversation-contract", createdByActorType: "system", createdByActorId: "test" }).returning(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, nativeSessionId: sessionId, + runnerInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256, + status: "running", runtimeMode: "native", responsibleUserId: userId, + contextSnapshot: { issueId, source: "issue.comment", wakeCommentId: wakeId, ...(conversation ? { conversationSessionGeneration: 0 } : {}) } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const port = new NativeRunCoordinatorStore(db, { companyId, issueId, runId, agentId, normalizedSessionId: sessionId, + runnerSourceInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256, + completionContractRevision: "test", completionContractCriterionIds: [] }); + const reply = "The teammate is hired and ready for your next instruction."; + const prpEvent = { runId, normalizedSessionId: sessionId, turnId: "hire-turn", sourceEventId: "fixture-final-answer", + payload: { kind: "agentMessage", channel: "final", providerPhase: "final", text: reply } }; + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 170, eventType: "item.completed", payload: { prpEvent } }); + await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", + summary: "A completed hire, with its final message preserved separately.", completionClaim: { contractRevision: "test", + objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], attentionRequests: [], artifacts: [] }, + terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" }, turnId: "hire-turn" }); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false }); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date("2020-01-01T00:02:00Z") }).where(eq(heartbeatRuns.id, runId)); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("committed"); + const input = { companyId, issueId, runId }; + return { ...input, input, agentId, userId, wakeId, prpEvent, reply }; + } + const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId)); + const issue = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0]; + + it("finds a committed final Agent Chat turn, persists the exact reply once and only then settles waiting", async () => { + const f = await fixture(); + expect(await comments(f.runId)).toEqual([]); + expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" }); + const work = nativeResponsePresentationWork(db), entries = await work.load(f.input); + expect(entries).toEqual([expect.objectContaining({ id: f.runId })]); + await expect(work.check(entries[0])).resolves.toBeNull(); + expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply, authorAgentId: f.agentId, createdByRunId: f.runId })]); + expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" }); + expect(await work.load(f.input)).toEqual([]); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toHaveLength(1); + }); + it("keeps a later user message actionable after recovering the earlier exact reply", async () => { + const f = await fixture(); + await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorUserId: f.userId, body: "Another instruction", + createdAt: new Date("2020-01-01T00:01:00Z") }); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true); + expect(await comments(f.runId)).toHaveLength(1); + expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" }); + }); + it("does not duplicate a reply under concurrent recovery", async () => { + const f = await fixture(); + expect((await Promise.all([recoverCommittedNativeChatResponse(db, f.input), recoverCommittedNativeChatResponse(db, f.input)])).sort()) + .toEqual([false, true]); + expect(await comments(f.runId)).toHaveLength(1); + }); + it.each(["skip", "generation", "agent", "user", "execution", "waiting", "turn", "event-run", "session", "channel", "kind", "digest", "partial"])( + "keeps %s mismatches unchanged instead of publishing an unauthorized or incomplete reply", async fault => { + const f = await fixture(); + if (["skip", "generation"].includes(fault)) await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, + source: "issue.comment", wakeCommentId: f.wakeId, conversationSessionGeneration: fault === "generation" ? 1 : 0, + ...(fault === "skip" ? { skipIssueComment: true } : {}) } }).where(eq(heartbeatRuns.id, f.runId)); + if (fault === "user") await db.update(heartbeatRuns).set({ responsibleUserId: "another-user" }).where(eq(heartbeatRuns.id, f.runId)); + if (fault === "agent") { + const other = randomUUID(); await db.insert(agents).values({ id: other, companyId: f.companyId, name: "Other agent" }); + await db.update(issues).set({ conversationAgentId: other, assigneeAgentId: other }).where(eq(issues.id, f.issueId)); + } + if (fault === "execution") { + const other = randomUUID(); await db.insert(heartbeatRuns).values({ id: other, companyId: f.companyId, agentId: f.agentId }); + await db.update(issues).set({ executionRunId: other }).where(eq(issues.id, f.issueId)); + } + if (fault === "waiting") await db.update(issues).set({ executionRunId: null, status: "in_review", conversationState: "waiting" }).where(eq(issues.id, f.issueId)); + if (["turn", "event-run", "session", "channel", "kind"].includes(fault)) { + const event = structuredClone(f.prpEvent); + if (fault === "turn") event.turnId = "foreign-turn"; + if (fault === "event-run") event.runId = randomUUID(); + if (fault === "session") event.normalizedSessionId = randomUUID(); + if (fault === "channel") event.payload.channel = "analysis"; + if (fault === "kind") event.payload.kind = "reasoning"; + await db.update(heartbeatRunEvents).set({ payload: { prpEvent: event } }).where(and(eq(heartbeatRunEvents.runId, f.runId), eq(heartbeatRunEvents.seq, 170))); + } + if (fault === "digest") await db.update(nativeRunResults).set({ canonicalSha256: "invalid-digest" }).where(eq(nativeRunResults.runId, f.runId)); + if (fault === "partial") await db.update(nativeRunFinalizations).set({ phase: "copying_back" }).where(eq(nativeRunFinalizations.runId, f.runId)); + const before = await issue(f.issueId); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toEqual([]); expect(await issue(f.issueId)).toEqual(before); + }, + ); + it.each([false, true])("preserves a selected reply without replacing or resurrecting it (deleted=%s)", async deleted => { + const f = await fixture(); + const [saved] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: f.agentId, + createdByRunId: f.runId, body: "Already selected", ...(deleted ? { deletedAt: new Date("2020-01-01T00:03:00Z") } : {}) }).returning(); + await db.update(heartbeatRuns).set({ resultJson: { presentationDecision: { commentId: saved.id, commentAction: "create" } } }).where(eq(heartbeatRuns.id, f.runId)); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toEqual([saved]); + }); + it("never crosses company scope and keeps ordinary internal task response recovery intact", async () => { + const f = await fixture(false); + await expect(recoverCommittedNativeChatResponse(db, { ...f.input, companyId: randomUUID() })).resolves.toBe(false); + const work = nativeResponsePresentationWork(db), entries = await work.load(f.input); + expect(entries).toHaveLength(1); await expect(work.check(entries[0])).resolves.toBeNull(); + expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply })]); + expect((await issue(f.issueId)).conversationAgentId).toBeNull(); + }); + it("rolls back a partial reply write without changing the waiting state or consuming recovery", async () => { + const f = await fixture(), abort = new Error("fixture-commit-aborted"); + const failing = new Proxy(db, { get(target, key) { return key === "transaction" + ? (callback: Parameters[0]) => db.transaction(async tx => { await callback(tx); throw abort; }) + : Reflect.get(target, key); } }); + await expect(recoverCommittedNativeChatResponse(failing, f.input)).rejects.toBe(abort); + expect(await comments(f.runId)).toEqual([]); + expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" }); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true); + expect(await comments(f.runId)).toHaveLength(1); + }); +}); diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index 5c3268f6afc..a1c6c914b33 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -65,7 +65,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick> '{terminal,turnTerminalState}' = 'completed'`, or( and(eq(heartbeatRuns.status, "succeeded"), - sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`, + or(sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`, + and(eq(statusDecisions.reasonCode, "conversation_turn_finished"), + eq(issues.conversationAgentId, heartbeatRuns.agentId), eq(issues.conversationUserId, heartbeatRuns.responsibleUserId), + eq(issues.status, "in_progress"), eq(issues.conversationState, "active"), + or(isNull(issues.executionRunId), eq(issues.executionRunId, heartbeatRuns.id)), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issues.id}::text`, + sql`jsonb_typeof(${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration') = 'number'`, + sql`${heartbeatRuns.contextSnapshot}->>'conversationSessionGeneration' = ${issues.conversationSessionGeneration}::text`, + exists(db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId), eq(heartbeatRunEvents.runId, heartbeatRuns.id), + eq(heartbeatRunEvents.agentId, heartbeatRuns.agentId), eq(heartbeatRunEvents.eventType, "item.completed"), + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,runId}' = ${heartbeatRuns.id}::text`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,normalizedSessionId}' = ${heartbeatRuns.nativeSessionId}::text`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,turnId}' = ${nativeRunResults.turnId}`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,text}' ~ '[^[:space:]]'`, + ))))), sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`, sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`), and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`, diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 53ccc1b07f7..a337ef27ea2 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -924,7 +924,7 @@ export async function claimNativeRestartRecoveries(input: { recycledProviderPids: providerProcesses.recycledPids, stderrTail: row.run.stderrExcerpt, }); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: row.run.companyId }); await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 202d2b4eac3..d39a3b682bb 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -29,7 +29,7 @@ import { readPersistedNativeFailureEvidence } from "./native-provider-failure-ev import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js"; import { findSupersededExternalReviewRequests } from "./superseded-external-review-requests.js"; import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js"; -import { conversationNativeDecision, isConversation } from "../agent-conversations.js"; +import { conversationNativeDecision, isConversation, settleConversationTurnForRun } from "../agent-conversations.js"; import { issueTreeControlService } from "../issue-tree-control.js"; import { randomUUID } from "node:crypto"; import { preserveNativeWorkspaceExportLease } from "./native-workspace-export-resume.js"; @@ -401,7 +401,7 @@ async function recordRetryableFailure(input: { if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); } - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: input.run.companyId }); await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); await tx .update(nativeRunFinalizations) @@ -747,6 +747,7 @@ async function recoverCommittedNativeChatResponseInScope( ): Promise { let agentId: string | null = null; let materialized = false; + let conversationReply = false; const entered = beginRunPhase("response.transaction.entry"); try { materialized = await measureRunPhase("response.transaction", () => db.transaction(async (transaction) => { @@ -789,6 +790,7 @@ async function recoverCommittedNativeChatResponseInScope( id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + responsibleUserId: heartbeatRuns.responsibleUserId, nativeIssueId: heartbeatRuns.nativeIssueId, nativeSessionId: heartbeatRuns.nativeSessionId, completionContractId: heartbeatRuns.completionContractId, @@ -803,6 +805,8 @@ async function recoverCommittedNativeChatResponseInScope( contextSnapshot: sql | null>`case when ${heartbeatRuns.contextSnapshot} is null then null else jsonb_build_object( 'skipIssueComment', ${heartbeatRuns.contextSnapshot}->'skipIssueComment', + 'issueId', ${heartbeatRuns.contextSnapshot}->'issueId', + 'conversationSessionGeneration', ${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration', 'source', ${heartbeatRuns.contextSnapshot}->'source', 'externalChatContinuation', ${heartbeatRuns.contextSnapshot}->'externalChatContinuation', 'paperclipWake', jsonb_build_object('externalInteractionContinuation', @@ -909,12 +913,16 @@ async function recoverCommittedNativeChatResponseInScope( agentId = run.agentId; return true; } - // Copyback can be owned by reconciliation before the live heartbeat - // reaches its presentation step. Recover a completed internal task's - // exact final reply from the same accepted turn, without rerunning work - // or granting external-chat publication authority. + const conversation = isConversation(issue); + const currentConversation = conversation && decision.reasonCode === "conversation_turn_finished" + && issue.conversationAgentId === run.agentId && issue.conversationUserId === run.responsibleUserId + && issue.status === "in_progress" && issue.conversationState === "active" + && (!issue.executionRunId || issue.executionRunId === run.id) + && run.contextSnapshot?.issueId === issue.id + && typeof run.contextSnapshot.conversationSessionGeneration === "number" + && run.contextSnapshot.conversationSessionGeneration === issue.conversationSessionGeneration; if ( - !isConversation(issue) && + (!conversation || currentConversation) && run.status === "succeeded" && record(run.contextSnapshot).skipIssueComment !== true && !isExternalChatPresentationContext(run.contextSnapshot) && @@ -929,6 +937,9 @@ async function recoverCommittedNativeChatResponseInScope( eq(heartbeatRunEvents.companyId, input.companyId), eq(heartbeatRunEvents.runId, run.id), eq(heartbeatRunEvents.eventType, "item.completed"), + conversation ? and(eq(heartbeatRunEvents.agentId, run.agentId), + sql`${heartbeatRunEvents.payload}->'prpEvent'->>'runId' = ${run.id}`, + sql`${heartbeatRunEvents.payload}->'prpEvent'->>'normalizedSessionId' = ${run.nativeSessionId}`) : undefined, sql`${heartbeatRunEvents.payload}->'prpEvent'->>'turnId' = ${accepted.turnId}`, sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`, sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`, @@ -963,6 +974,7 @@ async function recoverCommittedNativeChatResponseInScope( })}::jsonb`, updatedAt: new Date(), }).where(eq(heartbeatRuns.id, run.id)); agentId = run.agentId; + conversationReply = conversation; return true; } if ( @@ -1039,12 +1051,14 @@ async function recoverCommittedNativeChatResponseInScope( } finally { entered(); } - if (materialized && agentId) + if (materialized && agentId) { + if (conversationReply) await settleConversationTurnForRun(db, input.companyId, input.runId); publishChatPublicationCommitSignal({ ...input, agentId, eventType: "run.presentation.resolved", }); + } return materialized; } @@ -1659,7 +1673,7 @@ export async function finalizeNativeRun(input: { if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); if (input.projectRunStatus) await notifyUnsafeWorkspaceExportRunChange(tx, run); if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId }); const [after] = await projectRun(tx); return after ? { before, after } : null; diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts index 5708272da2c..2542b8bb1e3 100644 --- a/server/src/services/native-runtime/native-run-projection.ts +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -138,7 +138,7 @@ export async function projectCommittedRun(input: { await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, input.run); await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: input.run.companyId }); await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); const [after] = await tx .update(heartbeatRuns) diff --git a/server/src/services/native-runtime/native-safe-replacement.test.ts b/server/src/services/native-runtime/native-safe-replacement.test.ts index 422b23dba6b..491bae17b2b 100644 --- a/server/src/services/native-runtime/native-safe-replacement.test.ts +++ b/server/src/services/native-runtime/native-safe-replacement.test.ts @@ -1,3 +1,4 @@ +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { getExecutionBlocker } from "../execution-blocker.js"; import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js"; import { buildExecutionContinuation } from "../execution-continuation.js"; @@ -577,7 +578,12 @@ const support = externalDatabaseUrl }); it("persists exactly one linked successor under competing sweepers and restarts", async () => { const source = await seed(2); + const owner = heartbeatService(db); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const worker = owner.registerRunDispatchWork(coordinator, () => true); + await worker.ready; const now = new Date(); + try { await Promise.all([ reconcileSafeNativeReplacements(db, now), reconcileSafeNativeReplacements(db, now), @@ -608,6 +614,10 @@ const support = externalDatabaseUrl .where(eq(nativeRunFinalizations.runId, source.runId)) )[0]?.failureDetail?.successorRunId, ).toBe(children[0]!.id); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(children[0]!.scheduledRetryAt!.getTime()), { timeout: 5_000 }); + } finally { + await worker.stop(); await coordinator.stop(); await owner.drainActiveRunExecutions(); + } }); it("retries status publication after a crash without dispatching provider work", async () => { await deliverExecutionStatuses(db); diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index 7fb789f3304..12b556fa779 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; @@ -267,7 +268,7 @@ export async function reconcileSafeNativeReplacements( current.failureDetail?.replacementDenied ) return; - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); await tx .update(nativeRunFinalizations) .set({ @@ -409,7 +410,8 @@ export async function reconcileSafeNativeReplacements( idempotencyKey: `native-safe-replacement:${run.id}`, }) .returning(); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: run.companyId }); + await notifyScheduledRetryDispatch(tx, { companyId: run.companyId }); await tx.insert(heartbeatRuns).values({ id: successorRunId, companyId: run.companyId, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 37abb1b88be..9cec1870142 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -9902,7 +9902,7 @@ async function executePaperclipNativeSessionWithinScope( eq(nativeRunFinalizations.runId, input.execution.binding.runId), eq(nativeRunFinalizations.leaseOwner, leaseOwner), ), - )); + ), { companyId: input.execution.binding.companyId }); const failedAtMs = Date.now(); const executionFailureMessage = redactIdentityText( @@ -9954,7 +9954,7 @@ async function executePaperclipNativeSessionWithinScope( eq(nativeRunFinalizations.leaseOwner, leaseOwner), eq(nativeRunFinalizations.attempt, attempt), isNull(nativeRunFinalizations.resultId), - )).returning({ runId: nativeRunFinalizations.runId })); + )).returning({ runId: nativeRunFinalizations.runId }), { companyId: input.execution.binding.companyId }); if (settled) { await stoppedLeaseRenewal; if (warmSessionId !== null && lifecyclePolicy.mode === "warm") { @@ -10041,7 +10041,7 @@ async function executePaperclipNativeSessionWithinScope( ), ) .for("update"); - await notifyExecutionRunWork(tx); + await notifyExecutionRunWork(tx, { companyId: input.execution.binding.companyId }); await notifyNativeFinalizationWork(tx, input.execution.binding); const updated = await tx .update(nativeRunFinalizations) diff --git a/server/src/services/native-runtime/native-superseded-retry.ts b/server/src/services/native-runtime/native-superseded-retry.ts index e98672826b0..1bb6df9bd50 100644 --- a/server/src/services/native-runtime/native-superseded-retry.ts +++ b/server/src/services/native-runtime/native-superseded-retry.ts @@ -39,7 +39,7 @@ export async function retireSupersededNativeRetry(db: Db, input: { )).limit(1); if (!source) return false; - await notifyExecutionRunWork(db); + await notifyExecutionRunWork(db, { companyId: run.companyId }); await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null, leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: "blocked", failureCode: "native_retry_superseded", failureDetail: { ...coordinator.failureDetail, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 991fa8f63b5..e3596791942 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,4 @@ +import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -2716,6 +2717,7 @@ export function recoveryService( const now = new Date(); const retryAt = readProviderQuotaRetryAt(input.latestRun, now); return database.transaction(async (tx) => { + await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ @@ -3316,6 +3318,7 @@ export function recoveryService( created = Boolean(enqueuedRun); } else { scheduledRun = await db.transaction(async (tx) => { + await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts new file mode 100644 index 00000000000..276edacab23 --- /dev/null +++ b/server/src/services/run-dispatch-notifications.ts @@ -0,0 +1,14 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type RunDispatchScope = { companyId: string }; + +export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { + const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); +} + +/** Only a retry row or deadline change needs to re-read the saved retry schedule. */ +export async function notifyScheduledRetryDispatch(transaction: object, scope?: RunDispatchScope) { + await notifyRunDispatch(transaction, scope); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch); +} diff --git a/server/src/services/scheduled-retry-cancellation-notifications.ts b/server/src/services/scheduled-retry-cancellation-notifications.ts index 8a8b4fe1a2f..945bb40e14b 100644 --- a/server/src/services/scheduled-retry-cancellation-notifications.ts +++ b/server/src/services/scheduled-retry-cancellation-notifications.ts @@ -1,7 +1,9 @@ +import { notifyScheduledRetryDispatch } from "./run-dispatch-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type ScheduledRetryCancellationScope = { companyId: string; runId?: string }; export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) { + await notifyScheduledRetryDispatch(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*"); } diff --git a/server/src/services/system-recovery.ts b/server/src/services/system-recovery.ts index 8dba366e1af..a554a765c4c 100644 --- a/server/src/services/system-recovery.ts +++ b/server/src/services/system-recovery.ts @@ -71,7 +71,7 @@ export async function updateSystemRecovery(db: Db, source: WorkspaceRecoverySour sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,revision}' = ${String(previous.revision)}`)) .returning({ id: issueRecoveryActions.id }); return Boolean(updated); - }); + }, { companyId: source.companyId }); } export async function loadSystemRecovery(db: Db, source: WorkspaceRecoverySourceIdentity) { diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index e9f0bd4edbf..17075c34eb9 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -334,7 +334,7 @@ export async function requestTaskRetry(input: { details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy) } }); return { requestId: active?.id ?? id, accepted: true, stale: false, observedRunId: latest.id, chatActionId: intent.chatActionId }; - }); + }, { companyId: input.companyId }); } /** Chat owns admission and authorization. The task receipt follows that diff --git a/tests/runner-e2e/chat-feature-testops-api.mjs b/tests/runner-e2e/chat-feature-testops-api.mjs index 704dd5842a7..9b2b1ddfc3f 100644 --- a/tests/runner-e2e/chat-feature-testops-api.mjs +++ b/tests/runner-e2e/chat-feature-testops-api.mjs @@ -379,9 +379,13 @@ export async function verifyChatUpload(api, env, selected, exported, { wait = ms const result = await get(`/api/testresult/${row.id}`); if (result.id !== row.id || result.projectId !== 1 || result.launchId !== selected.launchId || result.testCaseId !== selected.caseId || result.fullName !== selected.fullName || result.jobRun?.id !== Number(selected.allureJobRunId) - || result.jobRun?.job?.id !== selected.jobId || result.jobRun?.launchId !== selected.launchId || result.jobRun?.url !== runUrl - || !Array.isArray(result.links) || !result.links.some(link => link.url === runUrl) - || !result.links.some(link => link.url === sourceUrl)) fail('chat_testops_upload_mismatch'); + || result.jobRun?.launchId !== selected.launchId || result.jobRun?.url !== runUrl + || result.jobRun.job != null && (result.jobRun.job.id !== run.job.id + || result.jobRun.job.type != null && result.jobRun.job.type !== run.job.type + || result.jobRun.job.url != null && !publicJobUrl(result.jobRun.job.url, env)) + || !Array.isArray(result.links) || result.links.some(link => typeof link.url === 'string' + && (link.url.startsWith('https://github.com/paperclipai/paperclip/actions/runs/') && link.url !== runUrl + || link.url.startsWith('https://github.com/paperclipai/paperclip/commit/') && link.url !== sourceUrl))) fail('chat_testops_upload_mismatch'); if (!['passed', 'failed', 'broken', 'skipped', 'unknown'].includes(result.status)) fail('chat_testops_upload_mismatch'); if (result.status !== 'unknown' && result.status !== exported.status) fail('chat_testops_upload_mismatch'); const attached = await get(`/api/testresult/attachment?${new URLSearchParams({ testResultId: String(row.id), page: '0', size: '100' })}`); diff --git a/tests/runner-e2e/chat-feature-testops-api.test.mjs b/tests/runner-e2e/chat-feature-testops-api.test.mjs index 4422a6e0d0b..9b2bc7c74f7 100644 --- a/tests/runner-e2e/chat-feature-testops-api.test.mjs +++ b/tests/runner-e2e/chat-feature-testops-api.test.mjs @@ -322,11 +322,11 @@ const uploadReceipt = { status: 'passed', cleanup: 'passed', caseId: 91, campaig { name: 'Execution provenance and checks', contentType: 'application/json', contentLength: 300 }, { name: 'Sanitized Product E2E record and accounting', contentType: 'application/json', contentLength: 800 }] }; const uploadRun = { id: 9, job: { id: 7, type: 'github' }, launchId: 10, url: 'https://github.com/paperclipai/paperclip/actions/runs/123', stage: 'finished' }; -function uploadApi(change = {}, absent = 0) { +function uploadApi(change = {}, absent = 0, runChange = {}) { let polls = 0; return async (route, method) => { assert.equal(method, 'GET'); - if (route === '/api/jobrun/9') return uploadRun; + if (route === '/api/jobrun/9') return { ...uploadRun, ...runChange }; if (route === '/api/launch/10') return { id: 10, projectId: 1 }; if (route.startsWith('/api/testresult?')) { polls++; return { content: polls <= absent ? [] : [{ id: 45, testCaseId: 91 }], totalElements: polls <= absent ? 0 : 1, totalPages: polls <= absent ? 0 : 1 }; } if (route === '/api/testresult/45') return { id: 45, projectId: 1, launchId: 10, testCaseId: 91, fullName, jobRun: uploadRun, status: 'passed', @@ -348,11 +348,45 @@ test('readback verifies exact current launch/case/source/status and safe attachm test('readback rejects wrong launch, case, current jobrun, source and contradictory result status', async () => { const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); - for (const change of [{ launchId: 11 }, { testCaseId: 92 }, { jobRun: { ...uploadRun, id: 11 } }, { links: [] }, { status: 'failed' }]) { + for (const change of [{ launchId: 11 }, { testCaseId: 92 }, { jobRun: { ...uploadRun, id: 11 } }, { links: [{ url: 'https://github.com/paperclipai/paperclip/commit/' + 'b'.repeat(40) }] }, { status: 'failed' }]) { await assert.rejects(verifyChatUpload(uploadApi(change), uploadContext, uploadSelection, uploadReceipt), /upload_mismatch/); } }); +test('readback joins an unexpanded result jobrun to the authoritative jobrun response', async () => { + const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); + const { job, ...embedded } = uploadRun; + const result = await verifyChatUpload(uploadApi({ jobRun: embedded }), uploadContext, uploadSelection, uploadReceipt); + assert.equal(result.resultId, 45); assert.equal(result.jobId, 7); assert.equal(result.verification, 'VERIFIED'); +}); + +test('readback accepts protected empty presentation links while retaining the source-bound current run', async () => { + const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); + const { job, ...embedded } = uploadRun; + const result = await verifyChatUpload(uploadApi({ jobRun: embedded, links: [] }), uploadContext, uploadSelection, uploadReceipt); + assert.equal(result.targetSha, uploadContext.GITHUB_SHA); assert.equal(result.runId, '123'); + assert.equal(result.resultId, 45); assert.equal(result.verification, 'VERIFIED'); +}); + +test('readback rejects authoritative or embedded identity contradictions and a foreign source selection', async () => { + const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); + const { job, ...embedded } = uploadRun; + for (const runChange of [{ job: { id: 8, type: 'github' } }, { job: { id: 7, type: 'gitlab' } }, + { url: 'https://github.com/paperclipai/paperclip/actions/runs/124' }, { launchId: 11 }]) { + await assert.rejects(verifyChatUpload(uploadApi({ jobRun: embedded, links: [] }, 0, runChange), uploadContext, uploadSelection, uploadReceipt), /jobrun_mismatch/); + } + for (const change of [{ jobRun: { ...embedded, id: 11 } }, { jobRun: { ...embedded, job: { id: 8 } } }, + { jobRun: { ...embedded, job: { id: 7, type: 'gitlab' } } }, { jobRun: { ...embedded, job: { id: 7, url: 'https://github.com/other/repo' } } }, + { jobRun: { ...embedded, launchId: 11 } }, { jobRun: { ...embedded, url: 'https://github.com/paperclipai/paperclip/actions/runs/124' } }, + { links: [{ url: 'https://github.com/paperclipai/paperclip/actions/runs/124' }] }, + { links: [{ url: 'https://github.com/paperclipai/paperclip/commit/' + 'b'.repeat(40) }] }]) { + await assert.rejects(verifyChatUpload(uploadApi(change), uploadContext, uploadSelection, uploadReceipt), /upload_mismatch/); + } + for (const selection of [{ ...uploadSelection, targetSha: 'b'.repeat(40) }, { ...uploadSelection, workflowSha: 'b'.repeat(40) }]) { + await assert.rejects(verifyChatUpload(uploadApi({ jobRun: embedded, links: [] }), uploadContext, selection, uploadReceipt), /upload_mismatch/); + } +}); + test('readback missing result or screenshot metadata is bounded unverified, never a fabricated pass', async () => { const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); let waits = 0; diff --git a/ui/src/components/task-chat/TaskChatComposerDock.test.tsx b/ui/src/components/task-chat/TaskChatComposerDock.test.tsx new file mode 100644 index 00000000000..68ba2e5623c --- /dev/null +++ b/ui/src/components/task-chat/TaskChatComposerDock.test.tsx @@ -0,0 +1,117 @@ +// @vitest-environment jsdom + +import { flushSync } from "react-dom"; +import { createRoot, type Root } from "react-dom/client"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { TaskChatComposerDock } from "./TaskChatComposerDock"; + +describe("TaskChatComposerDock", () => { + let container: HTMLDivElement; + let root: Root; + + beforeEach(() => { + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); + }); + + afterEach(() => { + flushSync(() => root.unmount()); + container.remove(); + vi.restoreAllMocks(); + }); + + function render(coordinator = true) { + const onTranscriptWheel = vi.fn(); + flushSync(() => root.render( +
+
+
+ +