test: pin worker faults to process handles and verify handoff fixtures

Co-Authored-By: Paperclip <noreply@paperclip.ing>

tests/runner-e2e/README.md

@@ -900,9 +900,10 @@ requires cancellation with `issue_reassigned`, no overlap with the successor,
 one successor run, unchanged scope and plan, retained draft, and a completed
 successor-owned document. It budgets three provider runs.
 
-Worker recovery kills only the exact running native worker PID from the public
-run record, after verifying its command-line run ID and the isolated local
-workspace. The real UI must show one Retry button. Clicking it must produce one
+Worker recovery requires Linux with Python pidfd support (as on the CI workers).
+It kills only the exact running native worker PID from the public
+run record, after verifying its command-line run ID, process start identity, and isolated
+local workspace. The signal uses an owned pidfd so PID reuse cannot retarget it. The real UI must show one Retry button. Clicking it must produce one
 successful attempt, consume the original message once, preserve the saved plan,
 and return a reference that was supplied only after the crash. This qualifies
 **user-initiated Retry after worker process loss**, not automatic recovery of

tests/runner-e2e/SECURITY.md

@@ -241,10 +241,13 @@ artifact and S3 object versions for incident analysis; do not weaken scanning
 to make a campaign publish.
 
 `agent-chat-qualification.worker-crash-retry` is local-only and explicit-only.
+Linux with Python pidfd support is required; unsupported hosts fail before signalling.
 The fault target comes from the public run detail, must be native and running,
 and must have an exact `--run-id` command argument. PID 1, the harness PID,
 noninteger PIDs, remote profiles, and mismatched identities are refused. The
-fixture workspace must be within the disposable instance root. This is one
+fixture workspace must be within the disposable instance root. The helper pins an owned pidfd, verifies the exact run argument and recorded
+process start ticks again after evidence capture, and signals through that handle.
+PID reuse cannot redirect the signal. This is one
 specific process signal, never a name-based or machine-wide process kill.
 The bounded fixture command is released even on failure; normal instance cleanup
 still owns all disposable processes and files. No credentials enter the prompt,

tests/runner-e2e/catalog.ts

@@ -975,7 +975,7 @@ export const runnerSuites: readonly RunnerSuiteFixture[] = [
     profiles: runnerProfiles.filter(profile => ["runner-codex", "runner-acpx-claude"].includes(profile.id))
       .map(profile => productionStoryProfile(defaultPermissionProfile(profile))),
     environments: [localEnvironment], tasks: chatQualificationTasks, expectedMatrixSize: 6,
-    definitionMetadata: { version: 1, permissions: "production-defaults", instructions: "production", crashBoundary: "verified-native-worker-pid-at-file-wait", recovery: "user-visible-retry", answerGrading: "exact-grounded-propositions-plus-separate-semantic-review", scheduling: "explicit-only" },
+    definitionMetadata: { version: 2, permissions: "production-defaults", instructions: "production", crashBoundary: "verified-native-worker-pid-at-file-wait", recovery: "user-visible-retry", answerGrading: "exact-grounded-propositions-plus-separate-semantic-review", scheduling: "explicit-only" },
   },
   ...(process.env.PAPERCLIP_RUNNER_E2E_CONNECTION_REVIEWS === "1" ? [connectionReviewSuite] : []),
   {

tests/runner-e2e/chat-qualification.test.ts

@@ -1,3 +1,5 @@
+import { execFileSync } from "node:child_process";
+import path from "node:path";
 import { describe, expect, it } from "vitest";
 import { assertActiveHandoff, assertAnswerFacts, assertCrashRecovered, assertWorkerIdentity } from "./chat-qualification.js";
 import { runnerMatrix } from "./catalog.js";
@@ -7,10 +9,13 @@ const run = { id: "old", agentId: "original", companyId: "company", runtimeMode:
 const successor = { ...run, id: "next", agentId: "successor", status: "succeeded", startedAt: "2026-09-21T00:01:01Z" };
 const before = { id: "task", assigneeAgentId: "original", description: "Preserve scope", projectId: null };
 const plan = { body: "Friday REFERENCE", latestRevisionId: "revision" };
-const handoff = { before, after: { ...before, status: "done", assigneeAgentId: "successor" }, oldRun: run, boundary: { ...run, status: "running" }, runs: [run, successor], successorId: "successor", planBefore: plan, planAfter: plan, draft: { body: "REFERENCE" }, output: { body: "REFERENCE", createdByAgentId: "successor" }, reference: "REFERENCE", audit: [{ action: "issue.reassigned", details: { source: "paperclip_runner_protocol" } }], taskIds: ["task"] };
-const recovery = { boundary: { ...run, status: "running" }, failed: { ...run, status: "failed" }, runs: [{ ...run, status: "failed" }, successor], issueId: "task", prompt: "Read my brief", comments: [{ body: "Read my brief" }, { body: "REFERENCE MARKER", authorAgentId: "original", createdByRunId: "next" }], reference: "REFERENCE", marker: "MARKER", planBefore: { body: "plan MARKER", latestRevisionId: "v1" }, planAfter: { body: "plan MARKER", latestRevisionId: "v1" } };
+const handoff = { before, after: { ...before, status: "done", assigneeAgentId: "successor" }, oldRun: run, boundary: { ...run, status: "running" }, runs: [run, successor], successorId: "successor", planBefore: plan, planAfter: plan, draft: { body: "REFERENCE" }, draftAfter: { body: "REFERENCE" }, output: { body: "REFERENCE", createdByAgentId: "successor" }, reference: "REFERENCE", audit: [{ action: "issue.reassigned", details: { source: "paperclip_runner_protocol" } }], taskIds: ["task"] };
+const recovery = { boundary: { ...run, status: "running" }, failed: { ...run, status: "failed" }, runs: [{ ...run, status: "failed" }, { ...successor, agentId: "original" }], issueId: "task", prompt: "Read my brief", comments: [{ body: "Read my brief" }, { body: "REFERENCE MARKER", authorAgentId: "original", createdByRunId: "next" }], reference: "REFERENCE", marker: "MARKER", planBefore: { body: "plan MARKER", latestRevisionId: "v1" }, planAfter: { body: "plan MARKER", latestRevisionId: "v1" } };
 
 describe("remaining native chat qualification", () => {
+  it("calibrates the pidfd helper against reuse and wrong-identity faults", () => {
+    execFileSync("python3", [path.join(import.meta.dirname, "worker-fault.test.py")], { stdio: "pipe" });
+  });
   it("requires a stopped original worker before exactly one successor executes", () => {
     expect(() => assertActiveHandoff(handoff)).not.toThrow();
     for (const change of [
@@ -23,6 +28,7 @@ describe("remaining native chat qualification", () => {
       { taskIds: ["replacement"] },
       { planAfter: { ...plan, body: "rewritten" } },
       { draft: { body: "no saved progress" } },
+      { draftAfter: { body: "deleted progress" } },
       { output: { body: "REFERENCE", createdByAgentId: "original" } },
       { audit: [] },
     ]) expect(() => assertActiveHandoff({ ...handoff, ...change })).toThrow();

tests/runner-e2e/chat-qualification.ts

@@ -27,7 +27,7 @@ export function assertWorkerIdentity(run: Row, command: string, environment: str
 
 export function assertActiveHandoff(e: {
   before: Row; after: Row; oldRun: ChatRun; boundary: ChatRun; runs: ChatRun[];
-  successorId: string; planBefore: Row; planAfter: Row; draft: Row; output: Row;
+  successorId: string; planBefore: Row; planAfter: Row; draft: Row; draftAfter: Row; output: Row;
   reference: string; audit: Row[]; taskIds: string[];
 }) {
   expect(e.boundary).toMatchObject({ id: e.oldRun.id, status: "running", agentId: e.before.assigneeAgentId });
@@ -43,6 +43,7 @@ export function assertActiveHandoff(e: {
   expect(Date.parse(successor.startedAt!)).toBeGreaterThanOrEqual(oldEnd);
   expect(e.planAfter).toEqual(e.planBefore);
   expect(e.draft.body).toContain(e.reference);
+  expect(e.draftAfter).toEqual(e.draft);
   expect(e.output.body).toContain(e.reference);
   expect(e.output.createdByAgentId).toBe(e.successorId);
   expect(e.audit.filter(a => a.action === "issue.reassigned")).toHaveLength(1);
@@ -56,9 +57,9 @@ export function assertCrashRecovered(e: {
   expect(e.boundary.status).toBe("running");
   expect(e.failed).toMatchObject({ id: e.boundary.id, status: "failed", runtimeMode: "native" });
   expect(e.runs).toHaveLength(2);
-  expect(e.runs.map(r => r.id)).toContain(e.failed.id);
+  expect(e.runs.find(r => r.id === e.failed.id)).toMatchObject({ status: "failed", contextSnapshot: { issueId: e.issueId } });
   const retry = e.runs.find(r => r.id !== e.failed.id)!;
-  expect(retry).toMatchObject({ status: "succeeded", runtimeMode: "native", contextSnapshot: { issueId: e.issueId } });
+  expect(retry).toMatchObject({ agentId: e.failed.agentId, status: "succeeded", runtimeMode: "native", contextSnapshot: { issueId: e.issueId } });
   expect(e.comments.filter(c => !c.authorAgentId && c.body === e.prompt)).toHaveLength(1);
   const replies = e.comments.filter(c => c.authorAgentId && c.body.includes(e.marker));
   expect(replies).toHaveLength(1);
@@ -93,7 +94,12 @@ export async function runActiveReassignment(context: Context) {
   const second = await api.post<Row>(`${company}/agents`, { ...config, name: "Morgan Successor", role: "engineer", reportsTo: f.agent.id });
   const wait = await brief(input, first.id);
   const reference = `REFERENCE${randomUUID().replaceAll("-", "")}`;
-  await api.patch(`/api/agents/${first.id}`, { instructionsBundle: { entryFile: "AGENTS.md", files: { "AGENTS.md": `First save a draft Paperclip document on the assigned task containing the reference from its plan. Then run node ${wait.scriptPath} and wait for the brief before finishing. Do not finish before the command returns.` } } });
+  const workerInstructions = `First save a draft Paperclip document on the assigned task containing the reference from its plan. Then run node ${wait.scriptPath} and wait for the brief before finishing. Do not finish before the command returns.`;
+  const savedInstructions = await api.request.put(`/api/agents/${first.id}/instructions-bundle/file`, {
+    data: { path: "AGENTS.md", content: workerInstructions },
+  });
+  expect(savedInstructions.ok()).toBe(true);
+  expect(await api.get(`/api/agents/${first.id}/instructions-bundle/file?path=AGENTS.md`)).toMatchObject({ content: workerInstructions });
   const task = await api.post<Row>(`${company}/issues`, { title: `Launch checklist ${input.nonce}`, status: "todo", assigneeAgentId: first.id,
     description: `Write a short launch checklist as a Paperclip document on this existing task. Use the saved plan and preserve any draft. Include its reference and ${marker} in the final checklist, then complete this task.`,
     initialPlan: `Welcome beginners on Friday at a free meetup. Reference: ${reference}` });
@@ -101,6 +107,7 @@ export async function runActiveReassignment(context: Context) {
     const boundary = await activeAtGate(context, wait.ready, first.id);
     const planBefore = await api.get<Row>(`/api/issues/${task.id}/documents/plan`);
     const draft = await readChatOutputDocument(api, task.id, reference);
+    await input.evidence("chat-active-reassignment-boundary.json", { task, boundary, planBefore, draft, successorId: second.id });
     // Only this positively observed run may be cancelled. Every other failure remains fatal.
     context.expectedStops.set(boundary.id, "cancelled");
     await sendChatMessage(input.page, `Reassign the currently running task ${task.identifier} from Riley Original to Morgan Successor now. Stop Riley's active execution as part of the handoff. Preserve the task, its description, saved plan, and draft; Morgan should complete the checklist on the same task. Do not create replacement work. Explain the handoff here.`);
@@ -108,6 +115,7 @@ export async function runActiveReassignment(context: Context) {
     const e = { before: task, after: await api.get<Row>(`/api/issues/${task.id}`), boundary,
       oldRun: await api.get<ChatRun>(`/api/heartbeat-runs/${boundary.id}`), runs: await context.allRuns(), successorId: second.id,
       planBefore, planAfter: await api.get<Row>(`/api/issues/${task.id}/documents/plan`), draft,
+      draftAfter: await api.get<Row>(`/api/issues/${task.id}/documents/${encodeURIComponent(draft.key)}`),
       output: await readChatOutputDocument(api, task.id, marker), reference,
       audit: await api.get<Row[]>(`/api/issues/${task.id}/activity`), taskIds: (await api.get<Row[]>(`${company}/issues`)).map(t => t.id) };
     await input.evidence("chat-active-reassignment.json", e);
@@ -128,10 +136,15 @@ export async function runWorkerCrash(context: Context) {
     const boundary = await activeAtGate(context, wait.ready, input.fixtures.agent.id) as ChatRun & Row;
     await context.refreshIssue();
     const planBefore = await input.api.get<Row>(`/api/issues/${context.issue().id}/documents/plan`);
+    if (process.platform !== "linux") throw new Error("Worker-crash qualification requires Linux pidfd support");
+    const faultHelper = path.join(import.meta.dirname, "worker-fault.py");
+    const processIdentity = JSON.parse(execFileSync("python3", [faultHelper, "inspect", String(boundary.processPid), boundary.id], { encoding: "utf8" }));
     const command = execFileSync("ps", ["-p", String(boundary.processPid), "-o", "command="], { encoding: "utf8" });
     assertWorkerIdentity(boundary, command, input.execution.environment.id);
-    await input.evidence("chat-worker-fault.json", { boundary, planBefore, fault: "SIGKILL exact public run processPid", recovery: "visible Retry button" });
-    process.kill(boundary.processPid, "SIGKILL");
+    await input.evidence("chat-worker-fault.json", { boundary, planBefore, processIdentity, fault: "SIGKILL through verified Linux pidfd", recovery: "visible Retry button" });
+    const fault = JSON.parse(execFileSync("python3", [faultHelper, "kill", String(boundary.processPid), boundary.id, processIdentity.startTicks], { encoding: "utf8" }));
+    expect(fault.signalled).toBe(true);
+    await input.evidence("chat-worker-fault-delivered.json", fault);
     context.expectedStops.set(boundary.id, "failed");
     await expect.poll(async () => (await input.api.get<ChatRun>(`/api/heartbeat-runs/${boundary.id}`)).status, { timeout: 120_000 }).toBe("failed");
     const failed = await input.api.get<Row>(`/api/heartbeat-runs/${boundary.id}`);
@@ -149,7 +162,7 @@ export async function runWorkerCrash(context: Context) {
     expect(await input.api.get(`/api/companies/${input.fixtures.company.id}/issues`)).toEqual([]);
   } finally {
     await writeFile(wait.gate, reference);
-    await input.evidence("chat-worker-final.json", { runs: await context.allRuns(), comments: await context.comments() });
+    await input.evidence("chat-worker-final.json", { runs: await context.allRuns(), comments: await context.comments().catch(error => ({ readError: String(error) })) });
   }
 }
 

tests/runner-e2e/worker-fault.py

@@ -0,0 +1,34 @@
+"""Linux-only E2E fault injection. pidfd pins the process across PID reuse."""
+import json
+import os
+import signal
+import sys
+
+mode, raw_pid, run_id, *expected = sys.argv[1:]
+pid = int(raw_pid)
+if sys.platform != "linux" or not hasattr(os, "pidfd_open") or not hasattr(signal, "pidfd_send_signal"):
+    raise RuntimeError("Worker-crash qualification requires Linux pidfd support")
+if pid <= 1 or pid == os.getpid() or mode not in ("inspect", "kill"):
+    raise RuntimeError("Invalid fault target")
+fd = os.pidfd_open(pid)
+try:
+    # Open the process handle before reading /proc. If the numeric PID is reused
+    # during these reads, fdinfo below no longer names it. Sending via the handle
+    # can only target the original process, never its replacement.
+    with open(f"/proc/{pid}/stat") as source:
+        start_ticks = source.read().rsplit(")", 1)[1].split()[19]
+    with open(f"/proc/{pid}/cmdline", "rb") as source:
+        args = source.read().decode().rstrip("\0").split("\0")
+    if not any(args[i:i + 2] == ["--run-id", run_id] for i in range(len(args) - 1)):
+        raise RuntimeError("Worker run ID does not match")
+    with open(f"/proc/self/fdinfo/{fd}") as source:
+        fields = dict(line.split(":", 1) for line in source if ":" in line)
+    if int(fields.get("Pid", "-1").strip()) != pid:
+        raise RuntimeError("Worker exited during identity check")
+    if mode == "kill":
+        if expected != [start_ticks]:
+            raise RuntimeError("Worker start identity changed")
+        signal.pidfd_send_signal(fd, signal.SIGKILL, None, 0)
+    print(json.dumps({"pid": pid, "startTicks": start_ticks, "runId": run_id, "signalled": mode == "kill"}))
+finally:
+    os.close(fd)

tests/runner-e2e/worker-fault.test.py

@@ -0,0 +1,79 @@
+import contextlib
+import io
+import json
+from pathlib import Path
+import runpy
+import signal
+import subprocess
+import sys
+import unittest
+from unittest.mock import patch
+
+HELPER = str(Path(__file__).with_name("worker-fault.py"))
+
+
+class FaultIdentityTests(unittest.TestCase):
+    def invoke(self, expected="1234", run="fixture-run", live_pid="43210", mode="kill"):
+        def fake_open(name, *args, **kwargs):
+            if name.endswith("/stat"):
+                return io.StringIO("43210 (worker) " + " ".join(["S"] + ["0"] * 18 + ["1234"]))
+            if name.endswith("/cmdline"):
+                return io.BytesIO(b"node\0runner\0--run-id\0fixture-run\0")
+            if "/fdinfo/" in name:
+                return io.StringIO("Pid:\t" + live_pid + "\n")
+            raise AssertionError(name)
+        with patch.object(sys, "platform", "linux"), patch.object(sys, "argv", [HELPER, mode, "43210", run, expected]), \
+             patch("os.pidfd_open", return_value=42, create=True) as opened, \
+             patch("signal.pidfd_send_signal", create=True) as sent, patch("os.close") as closed, \
+             patch("builtins.open", fake_open), contextlib.redirect_stdout(io.StringIO()):
+            try:
+                runpy.run_path(HELPER, run_name="__main__")
+            except RuntimeError:
+                sent.assert_not_called()
+                closed.assert_called_once_with(42)
+                raise
+            opened.assert_called_once_with(43210)
+            closed.assert_called_once_with(42)
+            if mode == "kill":
+                sent.assert_called_once_with(42, signal.SIGKILL, None, 0)
+            else:
+                sent.assert_not_called()
+
+    def test_signals_owned_handle_not_numeric_pid(self):
+        self.invoke()
+
+    def test_inspection_does_not_signal(self):
+        self.invoke(mode="inspect")
+
+    def test_refuses_changed_start_identity(self):
+        with self.assertRaisesRegex(RuntimeError, "start identity changed"):
+            self.invoke(expected="earlier-process")
+
+    def test_refuses_wrong_run(self):
+        with self.assertRaisesRegex(RuntimeError, "run ID"):
+            self.invoke(run="other-run")
+
+    def test_refuses_dead_original_handle_even_if_pid_was_reused(self):
+        with self.assertRaisesRegex(RuntimeError, "exited"):
+            self.invoke(live_pid="-1")
+
+    @unittest.skipUnless(sys.platform == "linux", "Real pidfd fault qualification runs on Linux CI")
+    def test_real_owned_child_with_wrong_then_correct_start_identity(self):
+        child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)", "--run-id", "fault-test"])
+        try:
+            args = [sys.executable, HELPER]
+            identity = json.loads(subprocess.check_output(args + ["inspect", str(child.pid), "fault-test"]))
+            wrong = subprocess.run(args + ["kill", str(child.pid), "fault-test", "wrong"], capture_output=True)
+            self.assertNotEqual(wrong.returncode, 0)
+            self.assertIsNone(child.poll())
+            result = json.loads(subprocess.check_output(args + ["kill", str(child.pid), "fault-test", identity["startTicks"]]))
+            self.assertTrue(result["signalled"])
+            self.assertEqual(child.wait(timeout=5), -signal.SIGKILL)
+        finally:
+            if child.poll() is None:
+                child.kill()
+            child.wait(timeout=5)
+
+
+if __name__ == "__main__":
+    unittest.main()