diff --git a/server/src/__tests__/native-run-projection-work.test.ts b/server/src/__tests__/native-run-projection-work.test.ts index 678e66b0633..d4af716f0a2 100644 --- a/server/src/__tests__/native-run-projection-work.test.ts +++ b/server/src/__tests__/native-run-projection-work.test.ts @@ -11,7 +11,7 @@ import { join } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agentWakeupRequests, agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, agents, authUsers, companies, companyMemberships, completionContracts, createDb, environmentLeases, environments, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { nativeRunProjectionWork } from "../services/native-runtime/native-run-projection-work.js"; import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; @@ -22,6 +22,9 @@ import * as statusCommitter from "../services/native-runtime/status-decision-com import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; import { workspaceOperationService } from "../services/workspace-operations.js"; +import { environmentService } from "../services/environments.js"; +import { withQueuedCommentIdsInWakePayload } from "../services/issue-queued-comment-queue.js"; +import { requestTaskRetry } from "../services/task-retry.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -652,4 +655,147 @@ describe("committed native run projection lifecycle", () => { skipNativeResponsePresentation: true, skipNativeReviewResponse: true, onWorkspaceSettled: settled }); expect(settled).toHaveBeenCalledTimes(1); }); + + describe("a comment deferred behind a run whose projection is left to restart recovery", () => { + const commenter = "projection-comment-user"; + let busyRunId: string | undefined; + afterEach(async () => { + if (busyRunId) await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, busyRunId)); + await db.update(agents).set({ runtimeConfig: {} }).where(eq(agents.id, agentId)); + busyRunId = undefined; + }); + async function deferredComment(binding: Awaited>, bodies = ["Also include the FOLLOWUP line."]) { + await db.insert(authUsers).values({ id: commenter, name: "Commenter", email: `${commenter}@example.test`, + createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: commenter, + status: "active", membershipRole: "operator" }).onConflictDoNothing(); + // One provider slot is held by another task, so a promoted follow-up stays + // queued here instead of starting a provider. + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, agentId)); + [{ id: busyRunId }] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running", + contextSnapshot: { issueId: randomUUID() } }).returning({ id: heartbeatRuns.id }); + const comments = await db.insert(issueComments).values(bodies.map(body => ({ companyId, issueId: binding.issueId, + authorUserId: commenter, body }))).returning(); + const ids = comments.map(comment => comment.id), latest = ids[ids.length - 1]; + // The durable receipt the comment route writes while the run is active. + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", + triggerDetail: "system", reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: commenter, + payload: { issueId: binding.issueId, commentId: latest, _paperclipWakeContext: { + issueId: binding.issueId, taskId: binding.issueId, commentId: latest, wakeCommentId: latest, + wakeCommentIds: ids, wakeReason: "issue_commented", source: "issue.comment" } }, + }).returning(); + // The killed executor never reached its cleanup, so its local lease is still active. + const environmentId = (await db.select({ id: environments.id }).from(environments).where(eq(environments.driver, "local")))[0]?.id + ?? (await db.insert(environments).values({ name: "Local", driver: "local" }).returning())[0].id; + const lease = await environmentService(db).acquireLease({ companyId, environmentId, heartbeatRunId: binding.runId, + issueId: binding.issueId, leasePolicy: "ephemeral", provider: "local" }); + return { comment: comments[0], comments, wake, lease }; + } + // The durable writes of an acknowledged same-turn steer (the queued-comment + // steer route): the message leaves the saved queue and the live run records + // that its turn received it. + async function steer(binding: Awaited>, wakeId: string, commentId: string) { + const wake = await readWake(wakeId); + const remaining = (wake.payload as { _paperclipWakeContext: { wakeCommentIds: string[] } }) + ._paperclipWakeContext.wakeCommentIds.filter(id => id !== commentId); + await db.update(agentWakeupRequests).set({ payload: withQueuedCommentIdsInWakePayload(wake.payload, remaining) }) + .where(eq(agentWakeupRequests.id, wakeId)); + const run = await read(binding.runId); + await db.update(heartbeatRuns).set({ resultJson: { ...(run.resultJson ?? {}), queuedSteeringAcknowledgements: { + [commentId]: { status: "acknowledged", queueId: wakeId, turnId: "steered-turn", acknowledgedAt: new Date().toISOString() } } } }) + .where(eq(heartbeatRuns.id, binding.runId)); + } + async function startRecovered() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + const heartbeat = heartbeatService(db); + worker = heartbeat.registerNativeRunProjectionWork(coordinator, () => admitted); + const receipts = heartbeat.registerDeferredReceiptWork(coordinator, () => admitted); + const settlement = heartbeat.registerNativeEnvironmentSettlementWork(coordinator, () => admitted); + await Promise.all([worker.ready, receipts.ready, settlement.ready]); + } + const followUps = async (issueId: string, sourceRunId: string) => (await db.select().from(heartbeatRuns) + .where(sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)).filter(run => run.id !== sourceRunId); + + it("promotes the comment exactly once after the recovered run is projected terminal", async () => { + const binding = await seed(true); + const { comment, wake, lease } = await deferredComment(binding); + await complete(binding, "control_plane"); await barrier(binding); + // Startup recovery runs while the killed run is still `running`: the + // comment correctly stays deferred behind it. + await startRecovered(); await flush(); + expect(await readWake(wake.id)).toMatchObject({ status: "deferred_issue_execution", runId: null }); + // The accepted result commits, but its executor died before projecting the + // run. The background projection worker settles the run instead. + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await flush(); + expect((await read(binding.runId)).status).toBe("succeeded"); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0].status).not.toBe("active"); + const promoted = await readWake(wake.id); + expect(promoted).toMatchObject({ status: "queued", runId: expect.any(String) }); + const [followUp, ...extra] = await followUps(binding.issueId, binding.runId); + expect(extra).toEqual([]); + expect(followUp).toMatchObject({ id: promoted.runId, status: "queued" }); + expect(followUp.contextSnapshot?.wakeCommentIds).toEqual([comment.id]); + // Another restart re-derives nothing new: the comment is not consumed twice. + await coordinator!.stop(); await startRecovered(); await flush(); + await heartbeatService(db).resumeQueuedRuns(); + expect((await followUps(binding.issueId, binding.runId)).map(run => run.id)).toEqual([followUp.id]); + }); + + it("does not wake again for a comment the recovered run already received as a steer", async () => { + const binding = await seed(true); + const { comments: [steered, saved], wake } = await deferredComment(binding, + ["Steer this into the live turn.", "Also include the FOLLOWUP line."]); + await steer(binding, wake.id, steered.id); + await complete(binding, "control_plane"); await barrier(binding); + await startRecovered(); await flush(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await flush(); + const [followUp, ...extra] = await followUps(binding.issueId, binding.runId); + expect(extra).toEqual([]); + expect(followUp.contextSnapshot?.wakeCommentIds).toEqual([saved.id]); + expect((await read(binding.runId)).resultJson?.queuedSteeringAcknowledgements).toMatchObject({ + [steered.id]: { status: "acknowledged" } }); + }); + + it("cancels the saved queue and starts no follow-up when its only comment was steered", async () => { + const binding = await seed(true); + const { comment, wake } = await deferredComment(binding); + // The steer route cancels a queue whose final message was delivered. + await steer(binding, wake.id, comment.id); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }) + .where(eq(agentWakeupRequests.id, wake.id)); + await complete(binding, "control_plane"); await barrier(binding); + await startRecovered(); await flush(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await flush(); await heartbeatService(db).resumeQueuedRuns(); + expect((await read(binding.runId)).status).toBe("succeeded"); + expect(await followUps(binding.issueId, binding.runId)).toEqual([]); + expect(await readWake(wake.id)).toMatchObject({ status: "cancelled", runId: null }); + }); + + it("joins a Retry click to the promoted follow-up instead of consuming the comment twice", async () => { + const binding = await seed(true); + const { comment, wake } = await deferredComment(binding); + await complete(binding, "control_plane"); await barrier(binding); + await startRecovered(); await flush(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await flush(); + const [followUp] = await followUps(binding.issueId, binding.runId); + expect(followUp).toMatchObject({ id: (await readWake(wake.id)).runId, status: "queued" }); + // A click that observed the recovered source is stale; one that observed + // the follow-up joins it. Neither admits another run with the comment. + expect(await requestTaskRetry({ db, companyId, issueId: binding.issueId, actorId: commenter, + idempotencyKey: randomUUID(), observedRunId: binding.runId })).toMatchObject({ accepted: false, stale: true }); + expect(await requestTaskRetry({ db, companyId, issueId: binding.issueId, actorId: commenter, + idempotencyKey: randomUUID(), observedRunId: followUp.id })).toMatchObject({ accepted: true }); + await flush(); await heartbeatService(db).resumeQueuedRuns(); + const runs = await followUps(binding.issueId, binding.runId); + expect(runs.map(run => run.id)).toEqual([followUp.id]); + expect(runs.filter(run => (run.contextSnapshot?.wakeCommentIds as string[] | undefined)?.includes(comment.id))) + .toHaveLength(1); + }); + }); }); diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index e3dbfc135dd..986ada32bde 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -21,7 +21,7 @@ pnpm exec vitest run --config tests/restart-chaos/vitest.config.ts # credential node scripts/typecheck-test-projects.mjs tests/restart-chaos # typecheck ``` -Options: `--triggers time,comment,retry,queued,tool` (fault timing, below), +Options: `--triggers time,comment,retry,queued,tool,result` (fault timing, below), `--workloads A1,B1,B5,C1`, `--budget-cents` (company hard cap, default 3000), `--poll-ms` (2000–5000), `--keep` (keep the temporary instance). @@ -137,6 +137,7 @@ the observer, after a seeded delay: | Trigger | Fires | Plan | | --- | --- | --- | | `comment` | 0–5 s after the C1 follow-up was posted to a live run (C1 only) | C1 + K1, B8 | +| `result` | 0–3 s after the run that received the C1 follow-up proposed or accepted its result, only while that run is still `running` (C1 only) | C1 + K1, B8 | | `retry` | 0–2 s after an accepted Retry click, as an extra K1 after the planned fault | K8 | | `queued` | on the first poll whose run is still `queued` with no start | B1 | | `tool` | 15–60 s after `bench.sh` progress first appears (B5 only) | B6 | diff --git a/tests/restart-chaos/observer.ts b/tests/restart-chaos/observer.ts index a33b334429a..5f4069bc610 100644 --- a/tests/restart-chaos/observer.ts +++ b/tests/restart-chaos/observer.ts @@ -39,6 +39,8 @@ function stillQueued(runs: readonly RunSample[]): boolean { return runs.some(run => run.status === "queued" && !run.startedAt) && !runs.some(run => run.startedAt || run.status === "running"); } export const TRIGGER_LATE_MS = 10_000; +/** The run settles soon after acceptance; arming on the proposal keeps the window reachable. */ +const RESULT_TRIGGER_EVENTS = new Set(["run.result.proposed", "run.result.accepted"]); /** Rest must hold for this long before the observer acts on it. */ export const REST_CONFIRM_MS = 12_000; @@ -160,6 +162,7 @@ export class TaskObserver { private triggerWindowOpen(fault: ScheduledFault, now: number, dueAt: number): boolean { if (now - dueAt > TRIGGER_LATE_MS) return false; + if (fault.trigger === "result") return this.resultRunStillRunning(this.latestReachableRuns()); if (fault.trigger !== "queued") return true; const latest = [...this.record.samples].reverse().find(sample => sample.reachable); return stillQueued(latest?.runs ?? []); @@ -169,6 +172,11 @@ export class TaskObserver { if (!fault.trigger || fault.trigger === "time") return true; const armedAt = this.triggerArmedAt(fault); let open = typeof armedAt === "number" && now - (armedAt + (fault.delayMs ?? 0)) <= TRIGGER_LATE_MS; + if (open && fault.trigger === "result") { + const runs = await this.api.get(`/api/issues/${this.record.issueId}/runs`, 5_000) + .then(rows => (rows ?? []).map(toRunSample)).catch(() => null); + open = runs !== null && this.resultRunStillRunning(runs); + } if (open && fault.trigger === "queued") { const runs = await this.api.get(`/api/issues/${this.record.issueId}/runs`, 5_000) .then(rows => (rows ?? []).map(toRunSample)).catch(() => null); @@ -178,6 +186,15 @@ export class TaskObserver { return open; } + private latestReachableRuns(): RunSample[] { + return [...this.record.samples].reverse().find(sample => sample.reachable)?.runs ?? []; + } + + /** A `result` fault outside the accepted-but-unsettled window would test a different boundary. */ + private resultRunStillRunning(runs: readonly RunSample[]): boolean { + return runs.some(run => run.id === this.record.resultRunId && run.status === "running"); + } + private noteTriggerMissed(fault: ScheduledFault) { const trigger = fault.trigger ?? "time"; (this.record.triggerMissed ??= []).push(trigger); @@ -206,6 +223,14 @@ export class TaskObserver { return queued && !started ? first.at : "missed"; } case "tool": return this.record.benchFirstSeenAt ?? null; + case "result": { + // B8: the run that received the follow-up has accepted its result but + // has not settled yet. A follow-up posted at rest cannot cross it. + const followUp = this.record.followUp; + if (!followUp) return null; + if (!followUp.commentId || !followUp.duringRun) return "missed"; + return this.record.resultAcceptedAt ?? null; + } } } @@ -294,6 +319,10 @@ export class TaskObserver { if (!events?.length) break; for (const event of events) { if (typeof event.seq === "number" && event.seq > cursor.seq) cursor.seq = event.seq; + if (RESULT_TRIGGER_EVENTS.has(String(event.eventType)) && this.record.followUp?.duringRun && !this.record.resultAcceptedAt) { + this.record.resultAcceptedAt = Date.now(); + this.record.resultRunId = run.id; + } cursor.lastAt = latestIso([cursor.lastAt, event.createdAt]); } if (events.length < RUN_EVENT_PAGE) break; diff --git a/tests/restart-chaos/scheduler.ts b/tests/restart-chaos/scheduler.ts index 7a3f86a657f..fd6cb16f226 100644 --- a/tests/restart-chaos/scheduler.ts +++ b/tests/restart-chaos/scheduler.ts @@ -34,10 +34,11 @@ export const TRIGGER_DELAY_RANGE_MS: Record, readonly retry: [0, 2_000], queued: [0, 0], tool: [15_000, 60_000], + result: [0, 3_000], }; export function triggerFitsWorkload(trigger: TriggerId, workload: WorkloadId): boolean { - if (trigger === "comment") return workload === "C1"; + if (trigger === "comment" || trigger === "result") return workload === "C1"; if (trigger === "tool") return workload === "B5"; return true; } @@ -54,7 +55,7 @@ export function planTasks(options: Pick trigger !== "time")) throw new Error("A K13 storm fires on its own clock; use --triggers time"); for (const trigger of triggers) { if (!options.workloads.some(workload => triggerFitsWorkload(trigger, workload))) { - throw new Error(`Trigger "${trigger}" needs workload ${trigger === "comment" ? "C1" : "B5"} in --workloads`); + throw new Error(`Trigger "${trigger}" needs workload ${trigger === "comment" || trigger === "result" ? "C1" : "B5"} in --workloads`); } } const cells: Array<{ workload: WorkloadId; adapter: AdapterId; fault: TaskFaultKind; trigger: TriggerId }> = []; diff --git a/tests/restart-chaos/triggers.test.ts b/tests/restart-chaos/triggers.test.ts index b9a06db2517..3a0001ca73a 100644 --- a/tests/restart-chaos/triggers.test.ts +++ b/tests/restart-chaos/triggers.test.ts @@ -25,9 +25,15 @@ const options: WaveOptions = { env: "local", adapters: ["codex", "claude", "open triggers: ["time", "comment", "retry", "queued", "tool"] }; type Json = Record; -function scriptedApi(state: { runs: Json[]; issue: Json; retry?: Json }) { +function scriptedApi(state: { runs: Json[]; issue: Json; retry?: Json; events?: Json[] }) { const api = { - async get(route: string) { return route.endsWith("/runs") ? state.runs : route.endsWith("/comments") ? [] : state.issue; }, + async get(route: string) { + if (route.includes("/events")) { + const after = Number(new URL(route, "http://x").searchParams.get("afterSeq") ?? 0); + return (state.events ?? []).filter(event => Number(event.seq) > after); + } + return route.endsWith("/runs") ? state.runs : route.endsWith("/comments") ? [] : state.issue; + }, async post(route: string) { if (route.endsWith("/issues")) return { id: "i1", identifier: "CHA-1" }; if (route.endsWith("/retry")) return state.retry ?? { accepted: true }; @@ -198,6 +204,56 @@ describe("trigger firing", () => { expect(observer.faultDue(start + 50_000)).toMatchObject({ trigger: "tool" }); }); + it("result: arms from the target run's result event and fires only while that run is running", async () => { + const running = { runId: "r1", status: "running", startedAt: new Date().toISOString() }; + const state = { runs: [running] as Json[], issue: { id: "i1", status: "in_progress" }, + events: [{ seq: 1, eventType: "turn.started", createdAt: new Date().toISOString() }] as Json[] }; + const observer = new TaskObserver({ ...base, trigger: "result", triggerDelayMs: 2_000 }, scriptedApi(state), lab, new Timeline(null), ready); + await observer.launch(); + const start = observer.record.createdAt!; + observer.record.followUp = { at: start, postedAt: start, commentId: "c1", error: null, duringRun: true }; + await observer.poll(start + 1_000); + // An unrelated event does not arm the trigger. + expect(observer.record.resultAcceptedAt).toBeUndefined(); + state.events.push({ seq: 2, eventType: "run.result.accepted", createdAt: new Date().toISOString() }); + await observer.poll(start + 2_000); + const armedAt = observer.record.resultAcceptedAt!; + expect(observer.record.resultRunId).toBe("r1"); + expect(observer.faultDue(armedAt + 1_999)).toBeNull(); + const fault = observer.faultDue(armedAt + 2_000); + expect(fault).toMatchObject({ trigger: "result" }); + expect(await observer.confirmTrigger(fault!, armedAt + 2_000)).toBe(true); + // The run settled before dispatch: the fault would land outside the window. + state.runs = [{ ...running, status: "succeeded", finishedAt: new Date().toISOString() }]; + expect(await observer.confirmTrigger(fault!, armedAt + 2_000)).toBe(false); + expect(observer.record.triggerMissed).toEqual(["result"]); + }); + + it("result: records a miss when the target run settled during the delay", async () => { + const state = { runs: [{ runId: "r1", status: "running" }] as Json[], issue: { id: "i1", status: "in_progress" }, + events: [{ seq: 1, eventType: "run.result.proposed", createdAt: new Date().toISOString() }] as Json[] }; + const observer = new TaskObserver({ ...base, trigger: "result", triggerDelayMs: 3_000 }, scriptedApi(state), lab, new Timeline(null), ready); + await observer.launch(); + const start = observer.record.createdAt!; + observer.record.followUp = { at: start, postedAt: start, commentId: "c1", error: null, duringRun: true }; + await observer.poll(start + 1_000); + const armedAt = observer.record.resultAcceptedAt!; + state.runs = [{ runId: "r1", status: "succeeded" }]; + await observer.poll(armedAt + 1_000); + expect(observer.faultDue(armedAt + 3_000)).toBeNull(); + expect(observer.record.triggerMissed).toEqual(["result"]); + }); + + it("result: misses when the follow-up was posted at rest", async () => { + const observer = new TaskObserver({ ...base, trigger: "result", triggerDelayMs: 0 }, + scriptedApi({ runs: [], issue: { id: "i1", status: "done" } }), lab, new Timeline(null), ready); + await observer.launch(); + const start = observer.record.createdAt!; + observer.record.followUp = { at: start, postedAt: start, commentId: "c1", error: null, duringRun: false }; + expect(observer.faultDue(start + 1)).toBeNull(); + expect(observer.record.triggerMissed).toEqual(["result"]); + }); + it("records a miss instead of firing a triggered fault after its window has passed", async () => { const observer = new TaskObserver({ ...base, workload: "B5", trigger: "tool", triggerDelayMs: 1_000 }, scriptedApi({ runs: [], issue: { id: "i1", status: "in_progress" } }), lab, new Timeline(null), ready); diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index 43432730be9..19889a8658c 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -11,7 +11,7 @@ export type WorkloadId = (typeof WORKLOAD_IDS)[number]; export const ADAPTER_IDS = ["codex", "claude", "opencode"] as const; export type AdapterId = (typeof ADAPTER_IDS)[number]; -export const TRIGGER_IDS = ["time", "comment", "retry", "queued", "tool"] as const; +export const TRIGGER_IDS = ["time", "comment", "retry", "queued", "tool", "result"] as const; export type TriggerId = (typeof TRIGGER_IDS)[number]; export const ENVIRONMENT_KINDS = ["local", "daytona"] as const; @@ -227,6 +227,10 @@ export interface TaskRecord { /** B5: distinct bench progress values the observer read during polling. */ benchProgressSeen?: string[]; benchFirstSeenAt?: number | null; + /** C1: first time the observer saw the run that received the follow-up propose or accept its result. */ + resultAcceptedAt?: number | null; + /** The run whose result armed the `result` trigger; the fault fires only while it is still running. */ + resultRunId?: string | null; triggerMissed?: TriggerId[]; }