diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index c4246ee559a..6238aedc3a4 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -52,7 +52,7 @@ new task or fault starts, the in-flight tasks drain to rest or their deadline, and only launched tasks are graded. Storm tasks show fault `K13` in the ledger; `storm_fault`, `storm_started` and `storm_ended` are in the timeline and the receipts are in `faults.json`. A replay of a storm task fires the faults that -crossed it, without a new storm. Only `--triggers time` applies. +crossed it, without a new storm. `--triggers time,result` may join a storm: a `result` task (C1) also fires its own K1 or K2 (from `--faults`) just after its run accepts a result, the next storm gap runs from its settle, and no triggered fault fires during the drain. Other triggers do not combine with a storm. ## Daytona (`--env daytona`) @@ -201,5 +201,7 @@ re-arms the task's own triggered faults instead of using their offsets. `n/a` and R3 decides; a task that rests unfinished with no Retry fails both. - Faults are serialized; each one hits every task running at that moment. - A `todo` or `in_progress` task with no live run and no named wait is not a rest. The observer keeps watching it until it resumes or passes the R5 stall limit. -- R5 also fails a run that stays `running` with no live evidence (no new run event, `lastUsefulActionAt` or `execution.lastConfirmedActivityAt`, and no start) for more than 3 minutes after the server is ready. A `running` status alone is not proof of life. +- R5 also fails a run that stays `running` with no live evidence (no new run event, `lastUsefulActionAt` or `execution.lastConfirmedActivityAt`, and no start) for more than 3 minutes after the server is ready. A `running` status alone is not proof of life. A quiet run is still live while a tool it started has not ended (for example a 4-minute `bench.sh`) or while its recorded local runner process still runs with the same PID and start time. +- The wave fingerprints the checkout and `packages/paperclip-runner/dist` at start and after every fault. A build, install or branch switch during a wave would make later restarts run different code, so it fails the wave as a harness error (`worktree_changed_mid_wave`). The baseline is taken after the harness's own setup (preflight rebuilds provider entrypoints) and covers the tracked diff, untracked file contents, the runner `dist` and the runnerd binary. Build before the wave, never during it. +- `faults.json` and `task-faults.json` (each task's launch time and the faults that crossed it) are written after every fault, so `--replay` of a task in an interrupted or wedged wave, or a storm task, re-fires the faults that crossed it at the same offsets. - A fault that fails to complete fails the wave as a harness error (exit 2), even when recovery brings the server back. diff --git a/tests/restart-chaos/grader.ts b/tests/restart-chaos/grader.ts index 343ebb250c5..4dd4d2cc109 100644 --- a/tests/restart-chaos/grader.ts +++ b/tests/restart-chaos/grader.ts @@ -159,6 +159,8 @@ export function deadRunningRun(sample: TaskSample, readySince: number): { id: st if (!sample.reachable || !sample.serverReady || sample.at - readySince <= STALL_LIMIT_MS) return null; for (const run of sample.runs ?? []) { if (run.status !== "running") continue; + // Quiet is not dead while a tool is still in flight or the run's process is verifiably alive. + if (run.toolInFlight || run.processAlive) continue; const evidence = Math.max(ms(run.lastEvidenceAt) || 0, ms(run.startedAt) || 0, readySince); if (sample.at - evidence > STALL_LIMIT_MS) return { id: run.id, idleMs: sample.at - Math.max(ms(run.lastEvidenceAt) || 0, ms(run.startedAt) || 0) }; } diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index eca608c0bf6..b8348803c6b 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -20,10 +20,11 @@ import { knownSecretValues, writeEvidenceBundle } from "./evidence.js"; import { gradeTask } from "./grader.js"; import { appendLedger, ledgerLine, writeSummary, type LedgerLine } from "./ledger.js"; import { TaskObserver } from "./observer.js"; -import { StormRun, observedFaultSchedule, planWave, replayWave } from "./scheduler.js"; +import { StormRun, TASK_FAULT_LOG, observedFaultSchedule, planWave, replayRecordFor, replayWave, type TaskFaultLog } from "./scheduler.js"; import { seedChaosLab } from "./seed.js"; import { ChaosSupervisor } from "./supervisor.js"; import { isBrokenStdoutError, markTimelineEchoBroken, Timeline } from "./timeline.js"; +import { guardedRunnerOutputs, worktreeFingerprint } from "./wave-guard.js"; import type { FaultKind, FaultReceipt, OracleArtifacts, TriggerId, WavePlan } from "./types.js"; const repositoryRoot = path.resolve(import.meta.dirname, "../.."); @@ -42,8 +43,11 @@ async function loadPlan(): Promise<{ plan: WavePlan; keep: boolean }> { // Replay the faults that actually crossed the task, including other tasks' faults. const graded = await readFile(path.join(sourceDir, `${command.ref.taskId}.grade.json`), "utf8") .then(text => JSON.parse(text) as { record: { createdAt: number | null; faultIds: string[] } }).catch(() => null); + // An interrupted or wedged wave has no grade files; its fault log was written after every fault. + const logged = await readFile(path.join(sourceDir, TASK_FAULT_LOG), "utf8").then(text => JSON.parse(text) as TaskFaultLog).catch(() => null); + const crossed = replayRecordFor(command.ref.taskId, graded?.record ?? null, logged); const faults = await readFile(path.join(sourceDir, "faults.json"), "utf8").then(text => JSON.parse(text) as FaultReceipt[]).catch(() => []); - const observed = graded ? observedFaultSchedule(graded.record.createdAt, graded.record.faultIds, faults, command.ref.taskId) : []; + const observed = crossed ? observedFaultSchedule(crossed.createdAt, crossed.faultIds, faults, command.ref.taskId) : []; if (!observed.length) console.warn("Replay: no recorded fault sequence for this task; using its planned fault only."); return { plan: replayWave(source, command.ref, Date.now(), command.overrides, observed), keep: command.keep }; } @@ -103,11 +107,22 @@ async function main(): Promise { const timeline = new Timeline(path.join(waveDir, "timeline.jsonl")); const gitSha = git(["rev-parse", "HEAD"]); const gitDirty = git(["status", "--porcelain"]).length > 0; + // Builds and branch switches must happen before the wave, never during it. + const guardedOutputs = guardedRunnerOutputs(repositoryRoot); + let startFingerprint: string | null = null; + let worktreeChanged = false; + const checkWorktree = (at: string) => { + if (worktreeChanged || startFingerprint === null || worktreeFingerprint(repositoryRoot, guardedOutputs) === startFingerprint) return; + worktreeChanged = true; + timeline.add("worktree_changed_mid_wave", { at, gitSha: git(["rev-parse", "HEAD"]) }); + }; timeline.add("wave_planned", { waveId: plan.waveId, seed: plan.seed, gitSha, gitDirty, replayOf: plan.replayOf ?? null, tasks: plan.tasks.map(task => `${task.taskId}:${task.adapter}/${task.workload}/${task.fault}${task.trigger ? `^${task.trigger}+${task.triggerDelayMs ?? 0}ms` : ""}@${task.fireAfterMs}ms${task.holdMs ? ` hold ${task.holdMs}ms` : ""}`), ...(plan.storm ? { storm: { durationMs: plan.storm.durationMs, faults: plan.storm.faults.map(fault => `${fault.kind}+${fault.gapMs}ms${fault.holdMs ? ` hold ${fault.holdMs}ms` : ""}`) } } : {}) }); preflight(plan); const serverEnvironment = plan.options.env === "daytona" ? await prepareDaytona(plan, timeline) : {}; + // Baseline only after the harness's own build steps (preflight rewrites the runner's dist). + startFingerprint = worktreeFingerprint(repositoryRoot, guardedOutputs); const supervisor = await ChaosSupervisor.create({ repositoryRoot, waveDir, timeline }); let sweeper: DaytonaWaveSweeper | null = null; @@ -116,6 +131,12 @@ async function main(): Promise { const onSignal = (signal: NodeJS.Signals) => { if (!interrupted) { interrupted = true; timeline.add("harness_interrupted", { signal }); } }; process.on("SIGINT", onSignal); process.on("SIGTERM", onSignal); const faults: FaultReceipt[] = []; + // Written after every fault, so an interrupted or wedged storm can still replay what crossed each task. + const taskFaultLog: TaskFaultLog = {}; + const persistFaultLog = async () => { + await writeFile(path.join(waveDir, "faults.json"), `${JSON.stringify(faults, null, 2)}\n`, { mode: 0o600 }).catch(() => undefined); + await writeFile(path.join(waveDir, TASK_FAULT_LOG), `${JSON.stringify(taskFaultLog, null, 2)}\n`, { mode: 0o600 }).catch(() => undefined); + }; const faultWork: Promise[] = []; const lines: LedgerLine[] = []; let anyFailure = false; @@ -171,9 +192,24 @@ async function main(): Promise { faultsQueued += 1; faultWork.push(supervisor.fault(kind, holdMs, requestedBy, trigger).finally(() => { faultsQueued -= 1; onSettled?.(); }).then(receipt => { faults.push(receipt); - for (const candidate of exposed) { candidate.noteFault(receipt.faultId, receipt.startedAt); candidate.extendDeadline((receipt.readyAt ?? Date.now()) - receipt.startedAt); } + checkWorktree(receipt.faultId); + for (const candidate of exposed) { + candidate.noteFault(receipt.faultId, receipt.startedAt); + candidate.extendDeadline((receipt.readyAt ?? Date.now()) - receipt.startedAt); + taskFaultLog[candidate.plan.taskId] = { createdAt: candidate.record.createdAt, faultIds: [...candidate.record.faultIds] }; + } + void persistFaultLog(); }, (error: Error & { receipt?: FaultReceipt }) => { - if (error.receipt) faults.push(error.receipt); + if (error.receipt) { + // A failed fault still crossed the tasks in flight; replay must fire it too. + const receipt = error.receipt; + faults.push(receipt); + for (const candidate of exposed) { + candidate.noteFault(receipt.faultId, receipt.startedAt); + taskFaultLog[candidate.plan.taskId] = { createdAt: candidate.record.createdAt, faultIds: [...candidate.record.faultIds] }; + } + void persistFaultLog(); + } faultFailures += 1; timeline.add("fault_error", { taskId: requestedBy, error: error.message }); })); @@ -203,15 +239,23 @@ async function main(): Promise { })); for (const observer of active) { if (faultsQueued || supervisor.faultInProgress || !supervisor.serverReady) break; + // A storm's drain gets no new fault; an unfired trigger is recorded as missed at rest. + if (storm && !storm.mayLaunch(Date.now())) break; const due = observer.faultDue(Date.now()); if (!due) continue; if (observer.record.restAt !== null) { timeline.add("fault_skipped_task_resting", { taskId: observer.plan.taskId, fault: due.kind }); continue; } const exposed = active.filter(candidate => !candidate.done); await Promise.all(exposed.map(candidate => candidate.snapshotBeforeFault())); if (!(await observer.confirmTrigger(due, Date.now()))) continue; + if (storm && !storm.mayLaunch(Date.now())) { + observer.markTriggerMissed(due); + timeline.add("fault_skipped_storm_ended", { taskId: observer.plan.taskId, fault: due.kind }); + break; + } const trigger = due.trigger && due.trigger !== "time" ? { trigger: due.trigger, delayMs: due.delayMs ?? 0 } : undefined; if (trigger) timeline.add("trigger_fired", { taskId: observer.plan.taskId, fault: due.kind, ...trigger }); - fireFault(due.kind, due.holdMs, observer.plan.taskId, exposed, trigger); + // Inside a storm a triggered fault is a restart too: the next storm gap runs from its settle. + fireFault(due.kind, due.holdMs, observer.plan.taskId, exposed, trigger, storm ? () => storm.settled(Date.now()) : undefined); } if (storm && !faultsQueued && !supervisor.faultInProgress && supervisor.serverReady) { const due = storm.due(Date.now()); @@ -231,6 +275,8 @@ async function main(): Promise { await delay(1_000); } await Promise.all(faultWork); + checkWorktree("wave_end"); + if (worktreeChanged) harnessError ??= new Error("The worktree changed during the wave (build, install or branch switch); later restarts may have run different code"); // A storm whose task pool ran dry early did not cover its storm; it is not a passing wave. if (storm && !interrupted && storm.exhaustedEarly(Date.now())) { timeline.add("storm_pool_exhausted", { faultsFired: storm.state.fired, launched: launched.length }); diff --git a/tests/restart-chaos/observer.ts b/tests/restart-chaos/observer.ts index 5f4069bc610..5e2901253cf 100644 --- a/tests/restart-chaos/observer.ts +++ b/tests/restart-chaos/observer.ts @@ -1,3 +1,4 @@ +import { spawnSync } from "node:child_process"; import { randomUUID } from "node:crypto"; import { readdir, readFile } from "node:fs/promises"; import os from "node:os"; @@ -9,6 +10,7 @@ import type { Timeline } from "./timeline.js"; import type { InterruptionSample, OracleArtifacts, RestSweep, RetryAttempt, RunSample, ScheduledFault, TaskPlan, TaskRecord, TaskSample, } from "./types.js"; +import { restartProcessStartMatches } from "../runner-e2e/process-tree-owner.js"; import { runOracleCommand } from "./oracle-sandbox.js"; import { taskFaultSchedule } from "./scheduler.js"; import { workloadDefinitions } from "./workloads.js"; @@ -60,6 +62,43 @@ export function toRunSample(run: Json): RunSample { }; } +interface RunEventCursor { + seq: number; + lastAt: string | null; + /** Tool starts minus tool ends. PRP item ids are not unique per tool call, so this counts. */ + openTools: number; + /** The run's latest recorded local runner process, until recovery reports it stopped. */ + process: { pid: number; startedAt: string } | null; +} + +/** R5 liveness: a tool in flight, or a recorded local runner process. */ +export function noteLivenessEvent(cursor: RunEventCursor, event: Json) { + const payload = (event.payload ?? {}) as Json; + switch (event.eventType) { + case "tool.execution.started": cursor.openTools += 1; break; + case "tool.execution.completed": cursor.openTools = Math.max(0, cursor.openTools - 1); break; + case "native.process_identity_recorded": + if (payload.targetKind === "local" && Number.isSafeInteger(payload.processPid) && typeof payload.processStartedAt === "string") { + // A replacement runner cannot finish a tool its predecessor started. + if (cursor.process && cursor.process.pid !== payload.processPid) cursor.openTools = 0; + cursor.process = { pid: payload.processPid, startedAt: payload.processStartedAt }; + } + break; + case "native.local_process_stopped": + // A stopped runner's in-flight tools can never send their completion. + if (cursor.process && payload.processPid === cursor.process.pid) { cursor.process = null; cursor.openTools = 0; } + break; + } +} + +/** Alive only when the PID still runs with the recorded start time, so a recycled PID never counts. */ +export function localProcessAlive(pid: number, startedAt: string): boolean { + if (process.platform === "win32") return false; + const result = spawnSync("ps", ["-o", "lstart=", "-p", String(pid)], { encoding: "utf8", env: { PATH: "/usr/bin:/bin", LANG: "C", LC_ALL: "C" }, timeout: 5_000 }); + const started = Date.parse(result.stdout?.trim() ?? ""); + return Number.isFinite(started) && restartProcessStartMatches(new Date(started).toISOString(), startedAt); +} + export function latestIso(values: ReadonlyArray): string | null { let best: number | null = null; for (const value of values) { @@ -104,6 +143,9 @@ export class TaskObserver { private readonly pendingFaults: ScheduledFault[]; done = false; + /** Replaceable in tests. A Daytona run's process is remote, so only local runs are checked. */ + processAlive = (pid: number, startedAt: string) => this.lab.environmentKind === "local" && localProcessAlive(pid, startedAt); + constructor(readonly plan: TaskPlan, private readonly api: ChaosApi, private readonly lab: ChaosLab, private readonly timeline: Timeline, private readonly supervisor: { serverReady: boolean; faultInProgress: boolean }, /** Paths oracle commands may not read: the operator's home, the instance root (secrets key, database) and results. */ @@ -195,6 +237,11 @@ export class TaskObserver { return runs.some(run => run.id === this.record.resultRunId && run.status === "running"); } + /** A due fault the wave decided not to fire (for example, the storm ended during its checks). */ + markTriggerMissed(fault: ScheduledFault) { + if (fault.trigger && fault.trigger !== "time") this.noteTriggerMissed(fault); + } + private noteTriggerMissed(fault: ScheduledFault) { const trigger = fault.trigger ?? "time"; (this.record.triggerMissed ??= []).push(trigger); @@ -309,11 +356,11 @@ export class TaskObserver { } } - private readonly runEventCursors = new Map(); + private readonly runEventCursors = new Map(); private async runEvidence(run: RunSample): Promise { if (run.status !== "running") return run; - const cursor = this.runEventCursors.get(run.id) ?? { seq: 0, lastAt: null }; + const cursor = this.runEventCursors.get(run.id) ?? { seq: 0, lastAt: null, openTools: 0, process: null }; for (let page = 0; page < 3; page += 1) { const events = await this.api.get(`/api/heartbeat-runs/${run.id}/events?afterSeq=${cursor.seq}&limit=${RUN_EVENT_PAGE}`, 5_000).catch(() => null); if (!events?.length) break; @@ -324,11 +371,14 @@ export class TaskObserver { this.record.resultRunId = run.id; } cursor.lastAt = latestIso([cursor.lastAt, event.createdAt]); + noteLivenessEvent(cursor, event); } if (events.length < RUN_EVENT_PAGE) break; } this.runEventCursors.set(run.id, cursor); - return { ...run, lastEvidenceAt: latestIso([run.lastEvidenceAt, cursor.lastAt]) }; + return { ...run, lastEvidenceAt: latestIso([run.lastEvidenceAt, cursor.lastAt]), + ...(cursor.openTools > 0 ? { toolInFlight: true } : {}), + ...(cursor.process && this.processAlive(cursor.process.pid, cursor.process.startedAt) ? { processAlive: true } : {}) }; } private async sample(now: number): Promise { diff --git a/tests/restart-chaos/scheduler.test.ts b/tests/restart-chaos/scheduler.test.ts index c926079c866..99f3a822fdf 100644 --- a/tests/restart-chaos/scheduler.test.ts +++ b/tests/restart-chaos/scheduler.test.ts @@ -163,6 +163,21 @@ describe("K13 restart storm", () => { expect(new Set(storm.faults.map(fault => fault.kind)).size).toBeGreaterThan(1); }); + it("lets result-trigger tasks join a storm with their own K1 or K2, while time cells stay K13", () => { + const tasks = planTasks({ ...stormOptions, tasks: 36, triggers: ["time", "result"] }); + const result = tasks.filter(task => task.trigger === "result"); + expect(result.length).toBeGreaterThan(0); + for (const task of result) { + expect(task.workload).toBe("C1"); + expect(["K1", "K2"]).toContain(task.fault); + expect(taskFaultSchedule(task)).toEqual([{ kind: task.fault, holdMs: 0, atMs: task.fireAfterMs, trigger: "result", delayMs: task.triggerDelayMs }]); + } + for (const task of tasks.filter(candidate => candidate.trigger === undefined)) expect(task.fault).toBe("K13"); + expect(() => planTasks({ ...stormOptions, faults: ["K4", "K6"], triggers: ["result"], workloads: ["C1"] })).toThrow(/needs K1 or K2/); + // A plain wave keeps its plan for a seed. + expect(planTasks({ ...options, triggers: ["time"] })).toEqual(planTasks({ ...options, triggers: undefined })); + }); + it("gives storm tasks no fault of their own", () => { const plan = planWave("w-20261106T101112-s42", stormOptions); expect(plan.storm?.faults.length).toBeGreaterThan(10); diff --git a/tests/restart-chaos/scheduler.ts b/tests/restart-chaos/scheduler.ts index fd6cb16f226..c331e740dc1 100644 --- a/tests/restart-chaos/scheduler.ts +++ b/tests/restart-chaos/scheduler.ts @@ -43,6 +43,23 @@ export function triggerFitsWorkload(trigger: TriggerId, workload: WorkloadId): b return true; } +/** Faults a `result` task may fire inside a storm: a fast restart that lands on the boundary. */ +const STORM_RESULT_FAULTS: readonly FaultKind[] = ["K1", "K2"]; + +/** + * A storm task has no fault of its own (`K13`): the wave's storm crosses it. + * A `result` task in a storm also fires its own K1 or K2 just after its run + * accepts a result, the window of RC-30, RC-33 and RC-41. + */ +function stormCellFaults(options: Pick, trigger: TriggerId): readonly TaskFaultKind[] { + if (!options.stormMinutes) return options.faults; + if (trigger === "time") return ["K13"]; + if (trigger !== "result") throw new Error(`A K13 storm fires on its own clock; --triggers ${trigger} does not combine with it (use time or result)`); + const faults = options.faults.filter(fault => STORM_RESULT_FAULTS.includes(fault)); + if (!faults.length) throw new Error("--triggers result in a storm needs K1 or K2 in --faults"); + return faults; +} + /** * Pairwise-friendly sampling: cycle a seeded shuffle of the full * workload × adapter × fault product so small waves still spread across cells. @@ -50,17 +67,19 @@ export function triggerFitsWorkload(trigger: TriggerId, workload: WorkloadId): b export function planTasks(options: Pick): TaskPlan[] { const rng = createRng(options.seed); const triggers = options.triggers?.length ? options.triggers : ["time" as const]; - // A storm task has no fault of its own: the wave's storm crosses it. - const faults: readonly TaskFaultKind[] = options.stormMinutes ? ["K13"] : options.faults; - if (options.stormMinutes && triggers.some(trigger => 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" || trigger === "result" ? "C1" : "B5"} in --workloads`); } } const cells: Array<{ workload: WorkloadId; adapter: AdapterId; fault: TaskFaultKind; trigger: TriggerId }> = []; + // Same cell order as before storms took triggers, so a seed keeps its plan. + const allowed = new Map(triggers.map(trigger => [trigger, stormCellFaults(options, trigger)])); + const faults = [...new Set([...allowed.values()].flat())]; for (const workload of options.workloads) for (const adapter of options.adapters) for (const fault of faults) { - for (const trigger of triggers) if (triggerFitsWorkload(trigger, workload)) cells.push({ workload, adapter, fault, trigger }); + for (const trigger of triggers) { + if (triggerFitsWorkload(trigger, workload) && allowed.get(trigger)!.includes(fault)) cells.push({ workload, adapter, fault, trigger }); + } } const tasks: TaskPlan[] = []; let order = rng.shuffle(cells); @@ -196,6 +215,17 @@ export function stormFaultDue(storm: StormPlan, state: StormState, now: number): return now >= (state.lastSettledAt ?? state.startedAt) + next.gapMs ? next : null; } +/** Written after every fault: each task's launch time and the faults that crossed it so far. */ +export const TASK_FAULT_LOG = "task-faults.json"; +export type TaskFaultLog = Record; + +/** The crossing record for a replay: the graded record, else the live fault log (interrupted or wedged waves). */ +export function replayRecordFor(taskId: string, graded: { createdAt: number | null; faultIds: string[] } | null, + logged: TaskFaultLog | null): { createdAt: number | null; faultIds: string[] } | null { + if (graded?.faultIds.length) return graded; + return logged?.[taskId] ?? graded; +} + /** Faults that actually crossed a task, as offsets from its launch. */ export function observedFaultSchedule(createdAt: number | null, faultIds: readonly string[], faults: ReadonlyArray<{ faultId: string; kind: FaultKind; holdMs: number; startedAt: number; requestedBy?: string; diff --git a/tests/restart-chaos/triggers.test.ts b/tests/restart-chaos/triggers.test.ts index 3a0001ca73a..a238bd28b0b 100644 --- a/tests/restart-chaos/triggers.test.ts +++ b/tests/restart-chaos/triggers.test.ts @@ -1,3 +1,4 @@ +import { spawnSync } from "node:child_process"; import { readFileSync } from "node:fs"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import os from "node:os"; @@ -6,7 +7,7 @@ import { describe, expect, it } from "vitest"; import { isSocketTypeOfServiceRace, type ChaosApi } from "./api.js"; import { parseCommand } from "./config.js"; import { gradeR5 } from "./grader.js"; -import { latestIso, QUEUED_TRIGGER_MAX_EMPTY_SAMPLES, TaskObserver, TRIGGER_LATE_MS } from "./observer.js"; +import { latestIso, localProcessAlive, noteLivenessEvent, QUEUED_TRIGGER_MAX_EMPTY_SAMPLES, TaskObserver, TRIGGER_LATE_MS } from "./observer.js"; import { observedFaultSchedule, planTasks, taskFaultSchedule, TRIGGER_DELAY_RANGE_MS } from "./scheduler.js"; import type { ChaosLab } from "./seed.js"; import { crashLoopShouldFire } from "./supervisor.js"; @@ -378,6 +379,74 @@ describe("R5 live evidence", () => { expect(gradeR5(record([sample(start, false), sample(start + 1_000), sample(start + 200_000)])).verdict).toBe("fail"); }); + const quietB5 = (live: { toolInFlight?: boolean; processAlive?: boolean }) => { + // A B5 run quiet for 4 minutes while bench.sh runs: no event, no activity, started long before. + const start = Date.parse("2026-11-06T10:00:00Z"); + const run = { id: "r1", status: "running", createdAt: null, startedAt: new Date(start - 600_000).toISOString(), finishedAt: null, + errorCode: null, retryOfRunId: null, wakeCommentIds: [], livenessState: null, lastEvidenceAt: new Date(start).toISOString(), ...live }; + return record([0, 60_000, 120_000, 180_000, 240_000].map(offset => ({ at: start + offset, reachable: true, serverReady: true, + issueStatus: "in_progress", runs: [run] }))); + }; + + it("does not flag a quiet run whose tool is still in flight or whose process is verifiably alive (B5 quiet 4 min)", () => { + expect(gradeR5(quietB5({ toolInFlight: true })).verdict).toBe("pass"); + expect(gradeR5(quietB5({ processAlive: true })).verdict).toBe("pass"); + expect(gradeR5(quietB5({}))).toMatchObject({ verdict: "fail", detail: expect.stringMatching(/^silent_dead_run r1/) }); + }); + + it("derives tool-in-flight and process liveness from run events", async () => { + const startedAt = new Date().toISOString(); + const identity = { processPid: 4242, processGroupId: 4242, processStartedAt: "2026-11-06T10:00:00.000Z", targetKind: "local" }; + const state = { runs: [{ runId: "r1", status: "running", startedAt }] as Json[], issue: { id: "i1", status: "in_progress" }, + events: [{ seq: 1, eventType: "native.process_identity_recorded", payload: identity, createdAt: startedAt }, + { seq: 2, eventType: "tool.execution.started", payload: {}, createdAt: startedAt }] as Json[] }; + const observer = new TaskObserver({ ...base, workload: "B5" }, scriptedApi(state), lab, new Timeline(null), ready); + const checked: Array<[number, string]> = []; + observer.processAlive = (pid, at) => { checked.push([pid, at]); return true; }; + await observer.launch(); + await observer.poll(Date.now()); + expect(observer.record.samples.at(-1)!.runs![0]).toMatchObject({ toolInFlight: true, processAlive: true }); + expect(checked).toEqual([[4242, "2026-11-06T10:00:00.000Z"]]); + state.events.push({ seq: 3, eventType: "tool.execution.completed", payload: {}, createdAt: startedAt }, + { seq: 4, eventType: "native.local_process_stopped", payload: { processPid: 4242 }, createdAt: startedAt }); + await observer.poll(Date.now() + 3_000); + const latest = observer.record.samples.at(-1)!.runs![0]!; + expect(latest.toolInFlight).toBeUndefined(); + expect(latest.processAlive).toBeUndefined(); + }); + + it("drops tool liveness when the runner that owned the tool stops or is replaced", () => { + const cursor = { seq: 0, lastAt: null, openTools: 0, process: null as { pid: number; startedAt: string } | null }; + const identity = (pid: number) => ({ eventType: "native.process_identity_recorded", + payload: { processPid: pid, processStartedAt: "2026-11-06T10:00:00.000Z", targetKind: "local" } }); + noteLivenessEvent(cursor, identity(10)); + noteLivenessEvent(cursor, { eventType: "tool.execution.started", payload: {} }); + expect(cursor.openTools).toBe(1); + // A fault killed the runner mid-tool: the tool can never report its end. + noteLivenessEvent(cursor, { eventType: "native.local_process_stopped", payload: { processPid: 10 } }); + expect(cursor).toMatchObject({ openTools: 0, process: null }); + noteLivenessEvent(cursor, identity(11)); + noteLivenessEvent(cursor, { eventType: "tool.execution.started", payload: {} }); + // A replacement runner recorded without a stop event still resets the count. + noteLivenessEvent(cursor, identity(12)); + expect(cursor).toMatchObject({ openTools: 0, process: { pid: 12 } }); + }); + + it("records a due trigger the wave skipped as missed", async () => { + const observer = new TaskObserver({ ...base, workload: "C1", trigger: "result", triggerDelayMs: 0 }, + scriptedApi({ runs: [], issue: { id: "i1", status: "in_progress" } }), lab, new Timeline(null), ready); + observer.markTriggerMissed({ kind: "K1", holdMs: 0, atMs: 0, trigger: "result", delayMs: 0 }); + observer.markTriggerMissed({ kind: "K2", holdMs: 0, atMs: 0 }); + expect(observer.record.triggerMissed).toEqual(["result"]); + }); + + it.skipIf(process.platform === "win32")("checks a local process by PID and start time, so a recycled PID is not alive", () => { + const lstart = spawnSync("ps", ["-o", "lstart=", "-p", String(process.pid)], { encoding: "utf8", env: { PATH: "/usr/bin:/bin", LANG: "C", LC_ALL: "C" } }).stdout.trim(); + const startedAt = new Date(Date.parse(lstart)).toISOString(); + expect(localProcessAlive(process.pid, startedAt)).toBe(true); + expect(localProcessAlive(process.pid, new Date(Date.parse(lstart) - 60_000).toISOString())).toBe(false); + }); + it("keeps the latest evidence timestamp", () => { expect(latestIso([null, "2026-11-06T10:00:00Z", "2026-11-06T09:00:00Z", "bad"])).toBe("2026-11-06T10:00:00.000Z"); expect(latestIso([undefined])).toBeNull(); diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index 19889a8658c..04951cd472b 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -118,6 +118,10 @@ export interface RunSample { wakeCommentIds: string[]; livenessState: string | null; lastEvidenceAt?: string | null; + /** A tool started with no matching end yet (a long `bench.sh` is quiet but live). */ + toolInFlight?: boolean; + /** The run's recorded local runner process is still running with the same start time. */ + processAlive?: boolean; } export interface InterruptionSample { diff --git a/tests/restart-chaos/wave-guard.test.ts b/tests/restart-chaos/wave-guard.test.ts new file mode 100644 index 00000000000..26a04a4357c --- /dev/null +++ b/tests/restart-chaos/wave-guard.test.ts @@ -0,0 +1,71 @@ +import { execFileSync } from "node:child_process"; +import { mkdir, mkdtemp, rm, utimes, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { describe, expect, it } from "vitest"; +import { observedFaultSchedule, replayRecordFor, type TaskFaultLog } from "./scheduler.js"; +import { worktreeFingerprint } from "./wave-guard.js"; + +describe("mid-wave worktree guard", () => { + it("changes on rebuilt runner output, a re-edited dirty file, an edited untracked file or a rebuilt runner binary", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "chaos-wave-guard-")); + const git = (...args: string[]) => execFileSync("git", args, { cwd: root, stdio: "ignore" }); + try { + git("init", "-q"); + git("config", "user.email", "chaos@example.invalid"); git("config", "user.name", "chaos"); + await writeFile(path.join(root, ".gitignore"), "dist/\nbin/\n"); + await writeFile(path.join(root, "server.ts"), "export const a = 1;\n"); + git("add", "."); git("commit", "-qm", "base"); + await mkdir(path.join(root, "dist/cli"), { recursive: true }); + await mkdir(path.join(root, "bin"), { recursive: true }); + await writeFile(path.join(root, "dist/cli/runner.js"), "one"); + await writeFile(path.join(root, "bin/runnerd"), "binary"); + const outputs = ["dist", "bin/runnerd"]; + const print = () => worktreeFingerprint(root, outputs); + + const clean = print(); + expect(print()).toBe(clean); + await utimes(path.join(root, "dist/cli/runner.js"), new Date(), new Date(Date.now() + 5_000)); + const rebuilt = print(); + expect(rebuilt).not.toBe(clean); + + // A wave may start dirty; re-editing the same dirty file must still be caught. + await writeFile(path.join(root, "server.ts"), "export const a = 2;\n"); + const dirty = print(); + await writeFile(path.join(root, "server.ts"), "export const a = 3;\n"); + expect(print()).not.toBe(dirty); + + await writeFile(path.join(root, "extra.ts"), "one"); + const untracked = print(); + await writeFile(path.join(root, "extra.ts"), "two"); + expect(print()).not.toBe(untracked); + + const beforeBinary = print(); + await utimes(path.join(root, "bin/runnerd"), new Date(), new Date(Date.now() + 9_000)); + expect(print()).not.toBe(beforeBinary); + } finally { await rm(root, { recursive: true, force: true }); } + }); +}); + +describe("storm replay from the live fault log", () => { + const faults = [ + { faultId: "f01", kind: "K1" as const, holdMs: 0, startedAt: 1_060_000, requestedBy: "storm" }, + { faultId: "f02", kind: "K6" as const, holdMs: 0, startedAt: 1_300_000, requestedBy: "storm" }, + { faultId: "f03", kind: "K4" as const, holdMs: 200_000, startedAt: 1_500_000, requestedBy: "storm" }, + ]; + const log: TaskFaultLog = { t07: { createdAt: 1_000_000, faultIds: ["f01", "f02"] } }; + + it("replays the storm faults that crossed an ungraded task at their offsets from its launch", () => { + const crossed = replayRecordFor("t07", null, log); + expect(crossed).toEqual(log.t07); + expect(observedFaultSchedule(crossed!.createdAt, crossed!.faultIds, faults, "t07")) + .toEqual([{ kind: "K1", holdMs: 0, atMs: 60_000 }, { kind: "K6", holdMs: 0, atMs: 300_000 }]); + }); + + it("prefers the graded record and falls back to the log only when the grade has no faults", () => { + const graded = { createdAt: 1_000_000, faultIds: ["f03"] }; + expect(replayRecordFor("t07", graded, log)).toBe(graded); + expect(replayRecordFor("t07", { createdAt: 1_000_000, faultIds: [] }, log)).toEqual(log.t07); + expect(replayRecordFor("t99", null, log)).toBeNull(); + }); +}); diff --git a/tests/restart-chaos/wave-guard.ts b/tests/restart-chaos/wave-guard.ts new file mode 100644 index 00000000000..92a81071487 --- /dev/null +++ b/tests/restart-chaos/wave-guard.ts @@ -0,0 +1,43 @@ +import { execFileSync } from "node:child_process"; +import { createHash } from "node:crypto"; +import { readdirSync, readFileSync, statSync } from "node:fs"; +import path from "node:path"; + +/** + * Every restart re-runs the control plane from this worktree. A build, an + * install or a branch switch during a wave would make later restarts run + * different code (RC-32 was a runner `dist` rebuilt mid-wave), so the wave + * fingerprints the checkout and the runner outputs after its own setup and + * re-checks after each fault. + */ +export function guardedRunnerOutputs(repositoryRoot: string): string[] { + const runner = process.env.PAPERCLIP_RUNNER_BINARY + ?? path.join(repositoryRoot, "packages/paperclip-runner/runner/target/debug/paperclip-runnerd"); + return ["packages/paperclip-runner/dist", path.relative(repositoryRoot, runner)]; +} + +export function worktreeFingerprint(repositoryRoot: string, outputs: readonly string[] = ["packages/paperclip-runner/dist"]): string { + const hash = createHash("sha256"); + const git = (args: string[]) => { + try { return execFileSync("git", args, { cwd: repositoryRoot, encoding: "buffer", maxBuffer: 256 * 1024 * 1024 }); } + catch { return Buffer.from("git-unavailable"); } + }; + hash.update(git(["rev-parse", "HEAD"])); + // Contents, not just paths: re-editing an already-dirty file must change the fingerprint. + hash.update(git(["diff", "--binary", "HEAD"])); + for (const file of git(["ls-files", "--others", "--exclude-standard", "-z"]).toString("utf8").split("\0").filter(Boolean).sort()) { + hash.update(`untracked:${file}:`); + try { hash.update(readFileSync(path.join(repositoryRoot, file))); } catch { hash.update("unreadable"); } + } + // Ignored build outputs (runner dist and the runnerd binary) are covered by size and mtime. + for (const output of outputs) { + const walk = (current: string) => { + const stat = statSync(current, { throwIfNoEntry: false }); + if (!stat) { hash.update(`missing:${path.relative(repositoryRoot, current)}\n`); return; } + if (!stat.isDirectory()) { hash.update(`${path.relative(repositoryRoot, current)}:${stat.size}:${stat.mtimeMs}\n`); return; } + for (const name of readdirSync(current).sort()) walk(path.join(current, name)); + }; + walk(path.join(repositoryRoot, output)); + } + return hash.digest("hex"); +}