diff --git a/doc/plans/2026-10-30-run-dispatch-sleep.md b/doc/plans/2026-10-30-run-dispatch-sleep.md index 82dd8692043..e5cd0a9e014 100644 --- a/doc/plans/2026-10-30-run-dispatch-sleep.md +++ b/doc/plans/2026-10-30-run-dispatch-sleep.md @@ -12,7 +12,7 @@ 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. -An idle queue has no recurring read or timer-driven dispatch attempt. +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 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 index 3d6566e543d..b6fbb7353de 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts @@ -1,3 +1,4 @@ +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"; @@ -28,7 +29,7 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean if (state.due !== null && state.due <= Date.now()) state.due = null; }); const next = vi.fn(async () => state.due); - const dispatch = vi.fn(async () => {}); + 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, { @@ -37,8 +38,8 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean }, () => !state.draining && !state.standby); cleanup.push(() => coordinator.stop()); return { db, state, promote, next, dispatch, coordinator, worker, - notify: () => db.transaction(tx => notifyRunDispatch(tx)), - settle: () => vi.advanceTimersByTimeAsync(0) }; + notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)), + settle: () => vi.advanceTimersByTimeAsync(1_000) }; } describe("run dispatch obligations", () => { @@ -67,20 +68,39 @@ describe("run dispatch obligations", () => { 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++) { + await f.notify({ companyId: "company-a" }); + await vi.advanceTimersByTimeAsync(10); + } + expect(f.promote.mock.calls.length).toBeLessThanOrEqual(2); + expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2); + await f.settle(); + expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" }); + expect(f.dispatch.mock.calls.slice(1).every(([scope]) => scope?.companyId === "company-a")).toBe(true); + 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.notify(); 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("retains failed work across drain and settles after recovery", async () => { - const f = fixture(); f.state.fail = true; await f.worker.ready; + 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.notify(); await f.settle(); + expect(f.promote).toHaveBeenCalledTimes(2); f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000); - expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).not.toHaveBeenCalled(); + 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(2); expect(f.coordinator.nextWakeAt()).toBeNull(); + 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; diff --git a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts index 4a616887479..bcc8ed23972 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts @@ -1,3 +1,4 @@ +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"; @@ -76,9 +77,11 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]); } async function completed(id: string) { - await vi.waitFor(async () => expect((await status(id))?.status).toBe("succeeded"), { timeout: 10_000 }); + await vi.waitFor(async () => expect(["running", "succeeded"]).toContain((await status(id))?.status), { timeout: 10_000 }); await heartbeat.drainActiveRunExecutions(); - expect((await status(id)).stdoutExcerpt).toContain("dispatch-proof"); + 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(); @@ -87,10 +90,10 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled }); 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: 2_000, random: () => 0 }); + 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())); + 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()); @@ -110,7 +113,7 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled 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())); + 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"); }); @@ -118,7 +121,14 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled 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 new Promise(resolve => setTimeout(resolve, 400)); + 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(); @@ -151,13 +161,39 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled await blocker.select().from(agents).where(eq(agents.id, f.agentId)).for("update"); writing = agentService(db).clearError(f.agentId); void writing.catch(() => {}); - await vi.waitFor(() => expect(databaseWorkPending(db, DELIVERY_QUEUES.queuedRunDispatch)).toBe(true)); + 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 }); diff --git a/server/src/__tests__/managed-skill-content.test.ts b/server/src/__tests__/managed-skill-content.test.ts index cd4ddb8135f..c59f4a1277a 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"; @@ -79,7 +81,96 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => { const companyId = await company(); - await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + 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("reports a provider mismatch for local versions under persisted Cloud configuration", async () => { + const companyId = await company(); + const skill = await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" }); + const local = await skillInstallationMaintenance(db).verifyVersions({ companyId, limit: 50 }); + expect(local.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "local_storage" })]); const key = "cloud-runtime-config:v1"; const run = (phase: string) => promisify(execFile)(process.execPath, [ "--import", fileURLToPath(import.meta.resolve("tsx")), @@ -87,16 +178,18 @@ describe.skipIf(!support.supported)("managed skills use committed versions", () `--phase=${phase}`, `--company=${companyId}`, "--limit=50", ], { env: { ...process.env, DATABASE_URL: database.connectionString, PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture", - PAPERCLIP_STORAGE_PROVIDER: "local_disk", PAPERCLIP_STORAGE_LOCAL_DIR: path.join(root, "objects") }, + PAPERCLIP_STORAGE_PROVIDER: "s3", PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-cli-fixture", + PAPERCLIP_STORAGE_S3_REGION: "us-east-1", PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64) }, 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 [persisted] = await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } }).returning(); try { const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!); - expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]); + expect(eligibility.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "version_backed" })]); const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!); - expect(verification.results).toEqual([expect.objectContaining({ status: "local_storage" })]); + expect(verification.results).toEqual([expect.objectContaining({ versionId: skill.currentVersionId, status: "provider_mismatch" })]); + expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toEqual([persisted]); } finally { await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key)); } diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index ec7f09bc3ac..e00c8519df7 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1341,7 +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).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 879ac991182..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", ); } diff --git a/server/src/modules/agent-lifecycle/adapters/records.ts b/server/src/modules/agent-lifecycle/adapters/records.ts index db842941836..96892987b01 100644 --- a/server/src/modules/agent-lifecycle/adapters/records.ts +++ b/server/src/modules/agent-lifecycle/adapters/records.ts @@ -81,22 +81,19 @@ export function agentRecords(db: Db, effects: LifecycleEffects) { } 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"); - return 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() - .then((rows) => rows[0] ?? null); + 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 63559e7701d..b05f94e1a4b 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -514,6 +514,7 @@ export function createPostgresRunDispatchAdapter( 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; @@ -526,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, ), 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/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 6fef650fe42..1839176769f 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -2,8 +2,10 @@ 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) { - await notifyRunDispatch(transaction); +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); @@ -14,9 +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); + 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 8c73880ce57..42dc9928ad5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1336,8 +1336,8 @@ export function heartbeatService( return heartbeatQueue.dispatchPendingNativeStatusWakeups(...args); } - function resumeQueuedRunsAfterDrain() { - const promise = heartbeatQueue.resumeQueuedRunsAfterDrain(); + function resumeQueuedRunsAfterDrain(...args: Parameters) { + const promise = heartbeatQueue.resumeQueuedRunsAfterDrain(...args); activeWakeupPromises.add(promise); void promise.then( () => { activeWakeupPromises.delete(promise); }, @@ -3721,7 +3721,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 || diff --git a/server/src/services/heartbeat/dispatch-work.ts b/server/src/services/heartbeat/dispatch-work.ts index 3291008dd9a..5b1f728688a 100644 --- a/server/src/services/heartbeat/dispatch-work.ts +++ b/server/src/services/heartbeat/dispatch-work.ts @@ -1,3 +1,4 @@ +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"; @@ -7,7 +8,7 @@ export function registerRunDispatchWork( owner: { promoteDueScheduledRetries(): Promise; nextScheduledRetryAt(): Promise; - resumeQueuedRunsAfterDrain(): Promise; + resumeQueuedRunsAfterDrain(scope?: RunDispatchScope): Promise; schedulingSuppressed(): Promise; }, canRun: () => boolean, @@ -16,6 +17,7 @@ export function registerRunDispatchWork( const retries = registerDeadlineWork(coordinator, DELIVERY_QUEUES.scheduledRetryDispatch, { retryMs: 30_000, continuationMs: 1_000, + notificationCoalesceMs: 1_000, canRun, run: async signal => { suppressed = await owner.schedulingSuppressed(); @@ -23,17 +25,34 @@ export function registerRunDispatchWork( }, 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 => { - if (!signal.aborted && canRun()) await owner.resumeQueuedRunsAfterDrain(); + 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() { retries.wake(); queue.wake(); }, + 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 9b08eece201..64ca0b84ffd 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1973,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 @@ -1983,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, ), diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index e44906e72fb..9d3b30367d2 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1045,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); diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index b5b64abc948..bdbbfa07752 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -966,7 +966,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) { @@ -1042,7 +1042,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/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/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/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>'{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-FEATURES.md b/tests/runner-e2e/CHAT-FEATURES.md index 3d237a790c1..864e25564a2 100644 --- a/tests/runner-e2e/CHAT-FEATURES.md +++ b/tests/runner-e2e/CHAT-FEATURES.md @@ -223,8 +223,12 @@ actors before any TestOps mutation. A later dispatch or replay cannot reuse that approved number. Missing or uncertain metadata stops; an uncertain request needs reconciliation rather than a new send. The nonce and launch search provide additional evidence. Normal TestOps **Run** uses `evaluate` and is independent of -this one-time administrator fence. The operation parameter identifier and launch -nonce are also protected configuration, not dispatch inputs. +this one-time administrator fence. The launch nonce is protected configuration, +not a dispatch input. `CHAT_TESTOPS_OPERATION_PARAMETER_ID` is required only when +the registered job defines `operation`; the launcher sets that parameter to +`evaluate`. An official job with only `Branch` keeps its exact +`refs/heads/master` value and uses the fixed workflow's `evaluate` default. +Unknown parameters and other branch values are rejected. `CHAT_TESTOPS_INTEGRATION_ID` names the reviewed existing project GitHub integration. The launcher verifies its project, type and enabled state, then binds diff --git a/tests/runner-e2e/chat-feature-testops-api.mjs b/tests/runner-e2e/chat-feature-testops-api.mjs index e6c73b97941..0a0a16cbcfc 100644 --- a/tests/runner-e2e/chat-feature-testops-api.mjs +++ b/tests/runner-e2e/chat-feature-testops-api.mjs @@ -17,7 +17,8 @@ const diagnosticCodes = new Set(['untrusted_chat_testops_context', 'testops_toke 'testops_created_case_identity_invalid', 'testops_created_plan_identity_invalid', 'testops_plan_scope_mismatch', 'testops_operation_invalid', 'testops_output_invalid', 'untrusted_public_testops_context', 'chat_testops_identity_missing', 'chat_testops_case_mismatch', 'chat_testops_plan_mismatch', 'chat_testops_plan_job_mismatch', 'chat_testops_job_mismatch', - 'chat_testops_jobrun_mismatch', 'chat_testops_launch_mismatch', 'chat_testops_selection_mismatch', 'chat_testops_parameter_mismatch', 'chat_testops_nonce_invalid', 'chat_testops_nonce_conflict', 'chat_testops_catalog_incomplete', 'chat_testops_launch_admission', 'chat_testops_github_http', 'chat_testops_integration_mismatch', 'chat_testops_upload_mismatch', 'chat_testops_upload_unverified']); + 'chat_testops_job_id_mismatch', 'chat_testops_job_type_mismatch', 'chat_testops_job_parameters_shape_mismatch', + 'chat_testops_job_parameters_limit', 'chat_testops_job_url_mismatch', 'chat_testops_jobrun_mismatch', 'chat_testops_launch_mismatch', 'chat_testops_selection_mismatch', 'chat_testops_parameter_mismatch', 'chat_testops_nonce_invalid', 'chat_testops_nonce_conflict', 'chat_testops_catalog_incomplete', 'chat_testops_launch_admission', 'chat_testops_github_http', 'chat_testops_integration_mismatch', 'chat_testops_upload_mismatch', 'chat_testops_upload_unverified']); export function failureProjection(error, stage) { const stages = ['context', 'output_admission', 'authentication', 'metadata_inspection', 'plan_bootstrap', 'selection_validation', 'launch_admission', 'launch_preflight', 'plan_launch', 'bootstrap_job_metadata', 'result_export', 'upload_verification', 'bootstrap_context', 'receipt']; return { stage: stages.includes(stage) ? stage : 'unknown', @@ -146,9 +147,7 @@ export async function validateChatSelection(api, env) { const assignments = await api(`/api/testplan/${planId}/job`); if (!Array.isArray(assignments) || assignments.length !== 1 || assignments[0].job?.id !== jobId || assignments[0].testCasesCount !== 1) fail('chat_testops_plan_job_mismatch'); const job = await api(`/api/job/${jobId}`); - const jobUrls = ['https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml']; - if (positive(env.CHAT_TESTOPS_GITHUB_WORKFLOW_ID)) jobUrls.push(`https://github.com/paperclipai/paperclip/actions/workflows/${env.CHAT_TESTOPS_GITHUB_WORKFLOW_ID}`); - if (job.id !== jobId || job.type !== 'github' || !jobUrls.includes(job.url)) fail('chat_testops_job_mismatch'); + if (job.id !== jobId || job.type !== 'github' || !publicJobUrl(job.url, env)) fail('chat_testops_job_mismatch'); const run = await api(`/api/jobrun/${jobRunId}`); if (run.id !== jobRunId || run.job?.id !== jobId || run.job?.type !== 'github' || !['scheduled', 'in_progress'].includes(run.stage) || !id(run.launchId) @@ -162,12 +161,16 @@ export async function validateChatSelection(api, env) { } function configuredIds(env) { - const names = ['CHAT_TESTOPS_CASE_ID', 'CHAT_TESTOPS_PLAN_ID', 'CHAT_TESTOPS_JOB_ID', 'CHAT_TESTOPS_OPERATION_PARAMETER_ID', 'CHAT_TESTOPS_INTEGRATION_ID']; + const names = ['CHAT_TESTOPS_CASE_ID', 'CHAT_TESTOPS_PLAN_ID', 'CHAT_TESTOPS_JOB_ID', 'CHAT_TESTOPS_INTEGRATION_ID']; if (names.some(name => !positive(env[name]))) fail('chat_testops_identity_missing'); - return Object.fromEntries(names.map((name, index) => [['caseId', 'planId', 'jobId', 'parameterId', 'integrationId'][index], Number(env[name])])); + const operation = env.CHAT_TESTOPS_OPERATION_PARAMETER_ID; + if (operation !== undefined && operation !== '' && !positive(operation)) fail('chat_testops_identity_missing'); + return { ...Object.fromEntries(names.map((name, index) => [['caseId', 'planId', 'jobId', 'integrationId'][index], Number(env[name])])), + parameterId: positive(operation) ? Number(operation) : null }; } function publicJobUrl(url, env) { return url === 'https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml' + || url === 'https://github.com/paperclipai/paperclip/actions?query=workflow%3A%22Chat+feature+TestOps%22' || positive(env.CHAT_TESTOPS_GITHUB_WORKFLOW_ID) && url === `https://github.com/paperclipai/paperclip/actions/workflows/${env.CHAT_TESTOPS_GITHUB_WORKFLOW_ID}`; } async function registeredPlan(api, env) { @@ -182,10 +185,24 @@ async function registeredPlan(api, env) { if (job.id !== ids.jobId || job.type !== 'github' || !publicJobUrl(job.url, env)) fail('chat_testops_job_mismatch'); const integration = await api(`/api/integration/${ids.integrationId}/project/1`); if (integration.id !== ids.integrationId || integration.projectId !== 1 || integration.info?.type !== 'github' || integration.disabled !== false) fail('chat_testops_integration_mismatch'); - const definition = await api(`/api/environment/${ids.parameterId}`); - if (definition.id !== ids.parameterId || definition.name !== 'operation' || !Array.isArray(job.parameters) || job.parameters.length !== 1 - || job.parameters[0].id !== ids.parameterId || !['bootstrap', 'evaluate'].includes(job.parameters[0].value)) fail('chat_testops_parameter_mismatch'); - return { ...ids, selected, plan, job }; + if (!Array.isArray(job.parameters) || job.parameters.length < 1 || job.parameters.length > 2) fail('chat_testops_parameter_mismatch'); + const seenIds = new Set(), seenNames = new Set(), parameters = []; + for (const parameter of job.parameters) { + if (!id(parameter.id) || seenIds.has(parameter.id)) fail('chat_testops_parameter_mismatch'); + seenIds.add(parameter.id); + const definition = await api(`/api/environment/${parameter.id}`); + if (definition.id !== parameter.id || !['Branch', 'operation'].includes(definition.name) || seenNames.has(definition.name)) fail('chat_testops_parameter_mismatch'); + seenNames.add(definition.name); + if (definition.name === 'Branch') { + if (parameter.value !== 'refs/heads/master') fail('chat_testops_parameter_mismatch'); + parameters.push({ id: parameter.id, value: parameter.value }); + } else { + if (ids.parameterId !== parameter.id || !['bootstrap', 'evaluate'].includes(parameter.value)) fail('chat_testops_parameter_mismatch'); + parameters.push({ id: parameter.id, value: 'evaluate' }); + } + } + if (ids.parameterId !== null && !seenNames.has('operation')) fail('chat_testops_parameter_mismatch'); + return { ...ids, selected, plan, job, parameters }; } export async function bootstrapJobMetadata(api, env, expectedLaunchId) { @@ -205,8 +222,11 @@ export async function bootstrapJobMetadata(api, env, expectedLaunchId) { if (launch.id !== expectedLaunchId || launch.projectId !== 1) fail('chat_testops_launch_mismatch'); } const job = await api(`/api/job/${run.job.id}`); - if (job.id !== run.job.id || job.type !== 'github' || !Array.isArray(job.parameters) || job.parameters.length > 4) fail('chat_testops_job_mismatch'); - if (strict && !publicJobUrl(job.url, env)) fail('chat_testops_job_mismatch'); + if (job.id !== run.job.id) fail('chat_testops_job_id_mismatch'); + if (job.type !== 'github') fail('chat_testops_job_type_mismatch'); + if (!Array.isArray(job.parameters)) fail('chat_testops_job_parameters_shape_mismatch'); + if (job.parameters.length > 4) fail('chat_testops_job_parameters_limit'); + if (strict && !publicJobUrl(job.url, env)) fail('chat_testops_job_url_mismatch'); const parameters = []; const seen = new Set(); for (const parameter of job.parameters) { @@ -214,9 +234,10 @@ export async function bootstrapJobMetadata(api, env, expectedLaunchId) { seen.add(parameter.id); const definition = await api(`/api/environment/${parameter.id}`); if (strict && definition.id !== parameter.id) fail('chat_testops_parameter_mismatch'); - const known = ['operation', 'ALLURE_JOB_RUN_ID', 'ALLURE_USERNAME'].includes(definition.name); + const known = ['operation', 'ALLURE_JOB_RUN_ID', 'ALLURE_USERNAME', 'Branch'].includes(definition.name); parameters.push({ id: parameter.id, name: known ? definition.name : 'UNKNOWN', - value: definition.name === 'operation' && ['bootstrap', 'evaluate', 'launch-plan'].includes(parameter.value) ? parameter.value : 'NOT_PROJECTED' }); + value: definition.name === 'operation' && ['bootstrap', 'evaluate', 'launch-plan'].includes(parameter.value) + || definition.name === 'Branch' && parameter.value === 'refs/heads/master' ? parameter.value : 'NOT_PROJECTED' }); } return { ...(strict ? { schema: 'paperclip.chat-feature-testops-bootstrap-registration/v1', ...context, capturedAfterWatch: true, runnableBindingQualified: false } : {}), status: 'observed', projectId: 1, jobId: job.id, allureJobRunId: env.ALLURE_JOB_RUN_ID, @@ -274,7 +295,7 @@ export async function launchChatPlan(api, env, markOnce, admission) { jobId: known.jobId, caseId: known.caseId, launchId: existing[0].id, status: 'EXISTING_NO_RESEND', actualResult: null }; } await markOnce({ ...source, launchAdmission: admission, nonce, planId: known.planId, jobId: known.jobId, caseId: known.caseId }); - const parameters = [{ id: known.parameterId, value: 'evaluate' }]; + const parameters = known.parameters; await api(`/api/job/${known.jobId}`, 'PATCH', { integrationId: known.integrationId, canRun: true, parameters }); const patched = await api(`/api/job/${known.jobId}`); if (patched.id !== known.jobId || patched.type !== 'github' || !publicJobUrl(patched.url, env) || !same(patched.parameters, parameters)) fail('chat_testops_parameter_mismatch'); diff --git a/tests/runner-e2e/chat-feature-testops-api.test.mjs b/tests/runner-e2e/chat-feature-testops-api.test.mjs index 40ea857bf6d..6a7c5ce549a 100644 --- a/tests/runner-e2e/chat-feature-testops-api.test.mjs +++ b/tests/runner-e2e/chat-feature-testops-api.test.mjs @@ -168,21 +168,24 @@ async function fence(env = launchEnv, changes = {}, workflowChanges = {}, repoCh : url.includes('/actions/runs/') ? { ...ownRun, ...changes } : { id: 99, path: '.github/workflows/chat-feature-testops.yml', state: 'active', ...workflowChanges }) })); } -function launchServer({ wrongParameter = false, wrongAssignment = false, uncertainRun = false, integrationChange = {} } = {}) { - const calls = [], parameters = [{ id: 17, value: 'bootstrap' }]; let assigned = false, launched = false, requests = 0; +function launchServer({ wrongParameter = false, wrongAssignment = false, uncertainRun = false, integrationChange = {}, initialParameters = [{ id: 17, value: 'bootstrap' }], expectedParameters = [{ id: 17, value: 'evaluate' }], definitions = {}, jobUrl = 'https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml' } = {}) { + const calls = [], parameters = structuredClone(initialParameters); let assigned = false, launched = false, requests = 0; const api = async (route, method, body) => { calls.push({ route, method, body }); if (route.startsWith('/api/launch?')) return { totalPages: 1, content: launched ? [{ id: 10, name: `Chat feature evals ยท ${'1'.repeat(32)}` }] : [] }; if (route === '/api/integration/2/project/1') return { id: 2, projectId: 1, info: { type: 'github' }, disabled: false, ...integrationChange }; - if (route === '/api/environment/17') return { id: 17, name: wrongParameter ? 'OTHER_PRIVATE_PARAMETER' : 'operation' }; + if (route.startsWith('/api/environment/')) { + const id = Number(route.split('/').at(-1)); + return definitions[id] ?? { id, name: id === 17 ? wrongParameter ? 'OTHER_PRIVATE_PARAMETER' : 'operation' : 'Branch' }; + } if (route === '/api/job/7') { - if (method === 'PATCH') { assert.deepEqual(body, { integrationId: 2, canRun: true, parameters: [{ id: 17, value: 'evaluate' }] }); parameters[0].value = 'evaluate'; } - return { id: 7, type: 'github', url: 'https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml', parameters }; + if (method === 'PATCH') { assert.deepEqual(body, { integrationId: 2, canRun: true, parameters: expectedParameters }); parameters.splice(0, parameters.length, ...structuredClone(body.parameters)); } + return { id: 7, type: 'github', url: jobUrl, parameters }; } if (route === '/api/testplan/8/job') return wrongAssignment ? [{ job: { id: 70 }, testCasesCount: 1 }] : assigned ? [{ job: { id: 7, parameters }, testCasesCount: 1 }] : []; if (route === '/api/testplan/8/assign') { assert.equal(method, 'POST'); assert.deepEqual(body, { jobId: 7, selection: { inverted: false, manual: false, leafsInclude: [91] } }); assigned = true; return null; } - if (route === '/api/testplan/8/jobparameter') { assert.deepEqual(body, { jobId: 7, parameters: [{ id: 17, value: 'evaluate' }] }); return null; } + if (route === '/api/testplan/8/jobparameter') { assert.deepEqual(body, { jobId: 7, parameters: expectedParameters }); return null; } if (route === '/api/testplan/8/run') { requests++; assert.equal(method, 'POST'); if (uncertainRun) throw new Error('synthetic uncertain transport'); launched = true; return { id: 10, projectId: 1, name: body.launchName }; } return selectedApi()(route); }; @@ -432,3 +435,106 @@ test('post-watch reader caps parameter definitions and refuses duplicate IDs', a await assert.rejects(bootstrapJobMetadata(strictBootstrapApi({ job: { parameters } }), publicEnv, 10)); } }); + + +const observedLegacyJobUrl = 'https://github.com/paperclipai/paperclip/actions?query=workflow%3A%22Chat+feature+TestOps%22'; + +test('official observed legacy and existing workflow URLs pass all three actual job consumers', async () => { + for (const url of [observedLegacyJobUrl, 'https://github.com/paperclipai/paperclip/actions/workflows/chat-feature-testops.yml', + 'https://github.com/paperclipai/paperclip/actions/workflows/99']) { + const env = { ...publicEnv, CHAT_TESTOPS_GITHUB_WORKFLOW_ID: '99' }; + const selected = await validateChatSelection(selectedApi({ '/api/job/7': { id: 7, type: 'github', url } }), env); + assert.equal(selected.caseId, 91); + const registered = await bootstrapJobMetadata(strictBootstrapApi({ job: { url } }), env, 10); + assert.equal(registered.publicWorkflowMatches, true); + const server = launchServer({ jobUrl: url }); + const launched = await launchChatPlan(server.api, launchEnv, async () => {}, await fence()); + assert.equal(launched.status, 'REQUESTED'); + assert.equal(server.requests(), 1); + } +}); + +test('official legacy URL admission rejects foreign workflow repo query credentials and fragment in all consumers', async () => { + for (const url of [observedLegacyJobUrl.replace('/paperclip/actions', '/paperclip-cloud/actions'), + observedLegacyJobUrl.replace('Chat+feature+TestOps', 'Other+workflow'), `${observedLegacyJobUrl}&extra=1`, + observedLegacyJobUrl.replace('https://github.com', 'https://user:password@github.com'), `${observedLegacyJobUrl}#extra`]) { + await assert.rejects(validateChatSelection(selectedApi({ '/api/job/7': { id: 7, type: 'github', url } }), publicEnv), /chat_testops_job_mismatch/); + await assert.rejects(bootstrapJobMetadata(strictBootstrapApi({ job: { url } }), publicEnv, 10), error => { + assert.deepEqual(failureProjection(error, 'bootstrap_context'), { stage: 'bootstrap_context', code: 'chat_testops_job_url_mismatch', httpStatus: null }); + return true; + }); + const server = launchServer({ jobUrl: url }); + await assert.rejects(launchChatPlan(server.api, launchEnv, async () => assert.fail('no mutation marker'), await fence()), /chat_testops_job_mismatch/); + assert.equal(server.calls.some(call => call.method), false); + } +}); + +test('bootstrap job predicate failures retain only distinct fixed codes and no observed values', async () => { + for (const [job, code] of [ + [{ id: 99 }, 'chat_testops_job_id_mismatch'], + [{ type: 'SYNTHETIC_PRIVATE_TYPE' }, 'chat_testops_job_type_mismatch'], + [{ parameters: undefined }, 'chat_testops_job_parameters_shape_mismatch'], + [{ parameters: null }, 'chat_testops_job_parameters_shape_mismatch'], + [{ parameters: 'SYNTHETIC_PRIVATE_PARAMETERS' }, 'chat_testops_job_parameters_shape_mismatch'], + [{ parameters: Array.from({ length: 5 }, (_, i) => ({ id: i + 1 })) }, 'chat_testops_job_parameters_limit'], + [{ url: 'https://SYNTHETIC_PRIVATE_URL.invalid/' }, 'chat_testops_job_url_mismatch'], + ]) { + await assert.rejects(bootstrapJobMetadata(strictBootstrapApi({ job }), publicEnv, 10), error => { + const projected = failureProjection(error, 'bootstrap_context'); + assert.deepEqual(projected, { stage: 'bootstrap_context', code, httpStatus: null }); + assert.ok(!JSON.stringify(projected).includes('SYNTHETIC_PRIVATE')); + return true; + }); + } +}); + + +test('launch preserves exact Branch and defaults evaluate only when operation is genuinely absent', async () => { + for (const [parameters, opId, expectedParameters] of [ + [[{ id: 19, value: 'refs/heads/master' }], undefined, [{ id: 19, value: 'refs/heads/master' }]], + [[{ id: 19, value: 'refs/heads/master' }], '', [{ id: 19, value: 'refs/heads/master' }]], + [[{ id: 17, value: 'bootstrap' }], '17', [{ id: 17, value: 'evaluate' }]], + [[{ id: 19, value: 'refs/heads/master' }, { id: 17, value: 'bootstrap' }], '17', [{ id: 19, value: 'refs/heads/master' }, { id: 17, value: 'evaluate' }]], + ]) { + const env = { ...launchEnv, CHAT_TESTOPS_OPERATION_PARAMETER_ID: opId }; + const server = launchServer({ initialParameters: parameters, expectedParameters }); + let marked = 0; + const result = await launchChatPlan(server.api, env, async () => { marked++; assert.equal(server.calls.some(call => call.method), false); }, await fence(env)); + assert.equal(result.status, 'REQUESTED'); assert.equal(server.requests(), 1); assert.equal(marked, 1); + assert.deepEqual(server.calls.filter(call => call.route === '/api/job/7' && call.method === 'PATCH')[0].body.parameters, expectedParameters); + assert.deepEqual(server.calls.filter(call => call.route === '/api/testplan/8/jobparameter')[0].body.parameters, expectedParameters); + } +}); + +test('launch refuses empty unknown duplicate wrong Branch or conflicting operation before any mutation', async () => { + const branch = { id: 19, value: 'refs/heads/master' }; + for (const [parameters, opId, definitions] of [ + [[], undefined, {}], + [[{ id: 19, value: 'refs/heads/other' }], undefined, {}], + [[branch, branch], undefined, {}], + [[branch, { id: 20, value: 'refs/heads/master' }], undefined, {}], + [[{ id: 19, value: 'refs/heads/master' }], undefined, { 19: { id: 99, name: 'Branch' } }], + [[{ id: 19, value: 'PRIVATE_PARAMETER_VALUE' }], undefined, { 19: { id: 19, name: 'PRIVATE_PARAMETER_NAME' } }], + [[branch], '17', {}], + [[branch], 'bad-id', {}], + [[{ id: 17, value: 'bootstrap' }], undefined, {}], + [[{ id: 17, value: 'bootstrap' }], '18', {}], + [[{ id: 17, value: 'launch-plan' }], '17', {}], + [[branch, { id: 17, value: 'bootstrap' }, { id: 20, value: 'refs/heads/master' }], '17', {}], + ]) { + const env = { ...launchEnv, CHAT_TESTOPS_OPERATION_PARAMETER_ID: opId }; + const server = launchServer({ initialParameters: parameters, definitions }); + await assert.rejects(launchChatPlan(server.api, env, async () => assert.fail('no mutation marker'), await fence(env)), /chat_testops_(?:identity_missing|parameter_mismatch)/); + assert.equal(server.calls.some(call => call.method), false); assert.equal(server.requests(), 0); + } +}); + +test('bootstrap projects only fixed Branch name and exact master value alongside genuine numeric ID', async () => { + const base = strictBootstrapApi({ job: { parameters: [{ id: 19, value: 'refs/heads/master' }] } }); + const result = await bootstrapJobMetadata(route => route === '/api/environment/19' ? { id: 19, name: 'Branch' } : base(route), publicEnv, 10); + assert.deepEqual(result.parameters, [{ id: 19, name: 'Branch', value: 'refs/heads/master' }]); + const hidden = strictBootstrapApi({ job: { parameters: [{ id: 19, value: 'PRIVATE_BRANCH_VALUE' }] } }); + const unknown = await bootstrapJobMetadata(route => route === '/api/environment/19' ? { id: 19, name: 'Branch' } : hidden(route), publicEnv, 10); + assert.deepEqual(unknown.parameters, [{ id: 19, name: 'Branch', value: 'NOT_PROJECTED' }]); + assert.ok(!JSON.stringify(unknown).includes('PRIVATE_BRANCH')); +});