Add owned native archive fault transport lifecycle

tests/runner-e2e/chat-archive-fault-control.test.ts

@@ -0,0 +1,69 @@
+import { createServer } from "node:http";
+import { randomUUID } from "node:crypto";
+import { mkdtemp, mkdir, readFile, rm, chmod, realpath, writeFile } from "node:fs/promises";
+import { execFile } from "node:child_process";
+import { promisify } from "node:util";
+import os from "node:os";
+import path from "node:path";
+import { fileURLToPath } from "node:url";
+import { expect, it } from "vitest";
+import { ARCHIVE_CONTROL_ENV, ARCHIVE_SCENARIO, archiveFaultClient, archiveServerEnvironment, startArchiveFaultControl } from "./chat-archive-fault-control.js";
+import { buildPaperclipServerEnvironment } from "./harness-env.js";
+import { reserveRunnerE2EServerPort } from "./ports.js";
+const id = `chat-features.runner-codex-subscription.local.${ARCHIVE_SCENARIO}`;
+it("starts before backend and gives only the test child a private authenticated control handoff", async () => {
+  const temporaryRoot = await realpath(await mkdtemp(path.join(os.tmpdir(), "paperclip-runner-e2e-"))); await mkdir(path.join(temporaryRoot, "selected-private-inputs"), { mode: 0o700 });
+  const port = await reserveRunnerE2EServerPort(), backendOrigin = `http://127.0.0.1:${port}`, secrets: string[] = [];
+  const control = await startArchiveFaultControl({ selected: [id], temporaryRoot, backendOrigin, privateSecrets: secrets });
+  const binding = { companyId: randomUUID(), agentId: randomUUID(), personId: randomUUID(), taskIds: [randomUUID(), randomUUID(), randomUUID()] };
+  let backendStarted = false; const runId = randomUUID();
+  const backend = createServer((req, res) => {
+    if (req.url === "/api/agents/me") { res.end(JSON.stringify({ id: binding.agentId, companyId: binding.companyId })); return; }
+    if (req.url === `/api/heartbeat-runs/${runId}`) { res.end(JSON.stringify({ id: runId, agentId: binding.agentId, companyId: binding.companyId, responsibleUserId: binding.personId })); return; }
+    res.end("owned offline backend");
+  });
+  try {
+    const env = { [ARCHIVE_CONTROL_ENV]: control.filename, PAPERCLIP_RUNNER_E2E_TEMP_ROOT: temporaryRoot };
+    const server = await archiveServerEnvironment({ selected: [id], temporaryRoot, backendOrigin, env });
+    expect(server.source[ARCHIVE_CONTROL_ENV]).toBeUndefined(); expect(server.overrides).toEqual({ PAPERCLIP_API_URL: control.proxyOrigin });
+    const privateReceipt = JSON.parse(await readFile(control.filename, "utf8"));
+    expect(JSON.stringify(server)).not.toContain(privateReceipt.token); expect(JSON.stringify(server)).not.toContain(privateReceipt.controlOrigin);
+    const actualProductEnv = buildPaperclipServerEnvironment(server.source, server.overrides);
+    expect(actualProductEnv.PAPERCLIP_API_URL).toBe(control.proxyOrigin);
+    expect(JSON.stringify(actualProductEnv)).not.toContain(control.filename); expect(JSON.stringify(actualProductEnv)).not.toContain(privateReceipt.token); expect(JSON.stringify(actualProductEnv)).not.toContain(privateReceipt.controlOrigin);
+    await new Promise<void>(resolve => backend.listen(port, "127.0.0.1", resolve)); backendStarted = true;
+    expect(await (await fetch(`${control.proxyOrigin}/api/health`)).text()).toBe("owned offline backend");
+    // A real separate Node test-host process reads the 0600 receipt; no bearer or
+    // URL is in argv, stdout or the product environment.
+    const script = path.join(temporaryRoot, "offline-client.mjs"), module = new URL("./chat-archive-fault-control.ts", import.meta.url).href;
+    await writeFile(script, `import{archiveFaultClient}from${JSON.stringify(module)};const c=await archiveFaultClient({selected:${JSON.stringify(id)},backendOrigin:${JSON.stringify(backendOrigin)}});await c.arm(${JSON.stringify(binding)});console.log(JSON.stringify({originBound:c.backendOrigin===${JSON.stringify(backendOrigin)},receipts:await c.receipts()}));`, { mode: 0o600 });
+    const child = await promisify(execFile)(process.execPath, ["--import", fileURLToPath(new URL("../../cli/node_modules/tsx/dist/loader.mjs", import.meta.url)), script], { env: { PATH: process.env.PATH, TMPDIR: os.tmpdir(), ...env }, timeout: 10000 });
+    expect(JSON.parse(child.stdout)).toEqual({ originBound: true, receipts: [] }); expect(child.stdout + child.stderr).not.toContain(privateReceipt.token); expect(child.stdout + child.stderr).not.toContain(privateReceipt.controlOrigin);
+    const client = await archiveFaultClient({ selected: id, backendOrigin, env }); await expect(client!.arm(binding)).rejects.toThrow("control failed");
+    await expect(client!.releaseForExplicitRetry()).rejects.toThrow("control failed");
+    const response = await fetch(`${control.proxyOrigin}/api/issues/${binding.taskIds[0]}/inbox-archive`, { method: "POST", headers: { authorization: "Bearer offline-marker", "x-paperclip-run-id": runId }, body: JSON.stringify({ userId: binding.personId }) }); await response.text();
+    expect(await client!.receipts()).toMatchObject([{ taskId: binding.taskIds[0], runId, forwarded: true, status: 200 }]);
+    expect(JSON.stringify(await client!.receipts())).not.toContain("offline-marker");
+    expect((await fetch(`${privateReceipt.controlOrigin}/receipts`, { method: "POST", body: "null" })).status).toBe(403);
+    await expect(archiveFaultClient({ selected: id, backendOrigin: "http://127.0.0.1:1", env })).rejects.toThrow("another backend");
+    await expect(archiveFaultClient({ selected: "unrelated", backendOrigin, env })).rejects.toThrow("exact selected");
+    await chmod(control.filename, 0o644); await expect(archiveFaultClient({ selected: id, backendOrigin, env })).rejects.toThrow("unsafe"); await chmod(control.filename, 0o600);
+    backend.closeAllConnections(); await new Promise<void>(resolve => backend.close(() => resolve())); backendStarted = false;
+    await control.close(); await expect(readFile(control.filename)).rejects.toThrow(); await expect(fetch(`${control.proxyOrigin}/api/health`)).rejects.toThrow();
+  } finally { if (backendStarted) { backend.closeAllConnections(); await new Promise<void>(resolve => backend.close(() => resolve())); } await control.close(); await rm(temporaryRoot, { recursive: true, force: true }); }
+});
+it("fails closed for a missing selected handoff and rejects unrelated executions without starting resources", async () => {
+  await expect(archiveFaultClient({ selected: id, backendOrigin: "http://127.0.0.1:12345", env: {} })).rejects.toThrow("missing");
+  await expect(startArchiveFaultControl({ selected: ["other"], temporaryRoot: "/unowned", backendOrigin: "http://127.0.0.1:12345", privateSecrets: [] })).rejects.toThrow("allowlisted");
+});
+
+it("preserves a preexisting private receipt when exclusive creation fails", async () => {
+  const temporaryRoot = await realpath(await mkdtemp(path.join(os.tmpdir(), "paperclip-runner-e2e-")));
+  await mkdir(path.join(temporaryRoot, "selected-private-inputs"), { mode: 0o700 });
+  const filename = path.join(temporaryRoot, "selected-private-inputs", "archive-fault.json");
+  await writeFile(filename, "existing-owned-receipt", { mode: 0o600 });
+  try {
+    await expect(startArchiveFaultControl({ selected: [id], temporaryRoot, backendOrigin: "http://127.0.0.1:12345", privateSecrets: [] })).rejects.toThrow();
+    expect(await readFile(filename, "utf8")).toBe("existing-owned-receipt");
+  } finally { await rm(temporaryRoot, { recursive: true, force: true }); }
+});

tests/runner-e2e/chat-archive-fault-control.ts

@@ -0,0 +1,96 @@
+import { createServer } from "node:http";
+import { randomBytes, timingSafeEqual } from "node:crypto";
+import { lstat, readFile, realpath, rm, writeFile } from "node:fs/promises";
+import path from "node:path";
+import { listenOnFetchAllowedPort } from "../e2e/fetch-allowed-port.js";
+import { startArchiveFaultProxy, validArchiveBinding, type ArchiveBinding, type ArchiveFaultTransport, type ArchiveTransportReceipt } from "./chat-catalog-domains/internal-archive-transport.js";
+export const ARCHIVE_CONTROL_ENV = "PAPERCLIP_RUNNER_E2E_ARCHIVE_CONTROL";
+export const ARCHIVE_SCENARIO = "inbox-search.archive-partial-failure";
+const executionId = `chat-features.runner-codex-subscription.local.${ARCHIVE_SCENARIO}`;
+interface Binding { version: 1; executionId: string; temporaryRoot: string; backendOrigin: string; origin: string; controlOrigin: string; token: string }
+const origin = (value: unknown): value is string => { try { const u = new URL(String(value)); return u.protocol === "http:" && u.hostname === "127.0.0.1" && !!u.port && u.origin === value; } catch { return false; } };
+async function privateDirectory(root: string) {
+  try {
+    const directory = path.join(root, "selected-private-inputs"), metadata = await lstat(root), dir = await lstat(directory);
+    if (!path.basename(root).startsWith("paperclip-runner-e2e-") || await realpath(root) !== root || await realpath(directory) !== directory || !metadata.isDirectory() || !dir.isDirectory() || metadata.mode & 0o077 || dir.mode & 0o077
+      || typeof process.getuid === "function" && (metadata.uid !== process.getuid() || dir.uid !== process.getuid())) throw new Error();
+  } catch { throw new Error("Private archive root is unavailable or unsafe"); }
+}
+async function privateBinding(filename: string, root: string, selected: readonly string[]) {
+  if (selected.length !== 1 || selected[0] !== executionId || filename !== path.join(root, "selected-private-inputs", "archive-fault.json")) throw new Error("Archive transport requires its exact selected owned case");
+  try {
+    await privateDirectory(root);
+    const directory = path.dirname(filename), dir = await lstat(directory), file = await lstat(filename);
+    if (await realpath(root) !== root || await realpath(directory) !== directory || !dir.isDirectory() || dir.mode & 0o077 || file.isSymbolicLink() || !file.isFile() || (file.mode & 0o777) !== 0o600 || file.size > 4096
+      || typeof process.getuid === "function" && (dir.uid !== process.getuid() || file.uid !== process.getuid())) throw new Error();
+    const value = JSON.parse(await readFile(filename, "utf8")) as Binding;
+    if (Object.keys(value).sort().join() !== "backendOrigin,controlOrigin,executionId,origin,temporaryRoot,token,version" || value.version !== 1 || value.executionId !== executionId || value.temporaryRoot !== root
+      || !origin(value.backendOrigin) || !origin(value.origin) || !origin(value.controlOrigin) || new Set([value.backendOrigin, value.origin, value.controlOrigin]).size !== 3 || !/^[a-f0-9]{64}$/.test(value.token)) throw new Error();
+    return value;
+  } catch { throw new Error("Private archive transport binding is unavailable or unsafe"); }
+}
+function validReceipt(value: unknown): value is ArchiveTransportReceipt {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return false;
+  const row = value as Record<string, unknown>, uuid = (v: unknown) => typeof v === "string" && /^[a-f0-9]{8}-(?:[a-f0-9]{4}-){3}[a-f0-9]{12}$/i.test(v);
+  return Object.keys(row).every(key => ["requestId", "sequence", "taskId", "phase", "requestSha256", "runId", "explicitUserId", "forwarded", "status", "completedAt"].includes(key))
+    && uuid(row.requestId) && uuid(row.taskId) && uuid(row.runId) && Number.isSafeInteger(row.sequence) && Number(row.sequence) > 0
+    && ["initial", "retry"].includes(String(row.phase)) && typeof row.requestSha256 === "string" && /^[a-f0-9]{64}$/.test(row.requestSha256)
+    && typeof row.forwarded === "boolean" && typeof row.completedAt === "string" && Number.isFinite(Date.parse(row.completedAt))
+    && (row.explicitUserId === undefined || typeof row.explicitUserId === "string" && /^[A-Za-z0-9_-]{8,128}$/.test(row.explicitUserId))
+    && (row.status === undefined || Number.isInteger(row.status) && Number(row.status) >= 100 && Number(row.status) <= 599);
+}
+export async function startArchiveFaultControl(input: { selected: readonly string[]; temporaryRoot: string; backendOrigin: string; privateSecrets: string[] }) {
+  if (input.selected.length !== 1 || input.selected[0] !== executionId || !origin(input.backendOrigin)) throw new Error("Archive fault is allowlisted only for its owned local execution");
+  await privateDirectory(input.temporaryRoot);
+  if (/(?:^|[\s,])(?:http|net|tls|undici|\*)(?:$|[\s,])/i.test(process.env.NODE_DEBUG ?? "")) throw new Error("Archive private control refuses raw transport debug output");
+  const proxy = await startArchiveFaultProxy(input.backendOrigin), token = randomBytes(32).toString("hex");
+  const control = createServer((req, res) => { void (async () => {
+    const supplied = req.headers.authorization;
+    if (typeof supplied !== "string" || supplied.length !== token.length + 7 || !timingSafeEqual(Buffer.from(supplied.slice(7)), Buffer.from(token)) || !supplied.startsWith("Bearer ")) { res.writeHead(403); res.end(); return; }
+    if (req.method !== "POST" || !["/arm", "/receipts", "/release"].includes(req.url ?? "")) { res.writeHead(404); res.end(); return; }
+    const chunks: Buffer[] = []; let size = 0; for await (const chunk of req) { size += chunk.length; if (size > 4096) { res.writeHead(413); res.end(); return; } chunks.push(chunk); }
+    const value: unknown = JSON.parse(Buffer.concat(chunks).toString() || "null"); let result: unknown = null;
+    if (req.url === "/arm") { if (!validArchiveBinding(value as ArchiveBinding)) throw new Error(); await proxy.arm(value as ArchiveBinding); }
+    else { if (value !== null) throw new Error(); if (req.url === "/release") await proxy.releaseForExplicitRetry(); else result = await proxy.receipts(); }
+    res.writeHead(200, { "content-type": "application/json" }); res.end(JSON.stringify({ result }));
+  })().catch(() => { if (!res.headersSent) res.writeHead(409); res.end(JSON.stringify({ error: "Archive control operation rejected" })); }); });
+  let filename: string | undefined, created = false;
+  try {
+    const port = await listenOnFetchAllowedPort(control), controlOrigin = `http://127.0.0.1:${port}`;
+    filename = path.join(input.temporaryRoot, "selected-private-inputs", "archive-fault.json");
+    input.privateSecrets.push(token, controlOrigin, filename);
+    const binding: Binding = { version: 1, executionId, temporaryRoot: input.temporaryRoot, backendOrigin: input.backendOrigin, origin: proxy.origin, controlOrigin, token };
+    await writeFile(filename, `${JSON.stringify(binding)}\n`, { mode: 0o600, flag: "wx" }); created = true;
+    await privateBinding(filename, input.temporaryRoot, input.selected);
+    let closed = false;
+    return { filename, proxyOrigin: proxy.origin, async close() { if (closed) return; closed = true; control.closeAllConnections(); await new Promise<void>(resolve => control.close(() => resolve())); await proxy.close(); await rm(filename!, { force: true }); } };
+  } catch (error) { control.closeAllConnections(); control.close(); await proxy.close(); if (created && filename) await rm(filename, { force: true }); throw error; }
+}
+/** Only the test supervisor sees this receipt. The product receives solely the
+ * normal API-origin override; never a control filename, URL or bearer. */
+export async function archiveServerEnvironment(input: { selected: readonly string[]; temporaryRoot: string; backendOrigin: string; env: NodeJS.ProcessEnv }) {
+  const source = { ...input.env }, filename = source[ARCHIVE_CONTROL_ENV]; delete source[ARCHIVE_CONTROL_ENV];
+  if (!filename) { if (input.selected.includes(executionId)) throw new Error("Selected archive case is missing pre-server fault transport"); return { source, overrides: {} }; }
+  const binding = await privateBinding(filename, input.temporaryRoot, input.selected);
+  if (binding.backendOrigin !== input.backendOrigin) throw new Error("Archive backend origin differs from owned server");
+  return { source, overrides: { PAPERCLIP_API_URL: binding.origin } };
+}
+export async function archiveFaultClient(input: { selected: string; backendOrigin: string; privateSecrets?: string[]; env?: NodeJS.ProcessEnv }): Promise<ArchiveFaultTransport | undefined> {
+  const env = input.env ?? process.env, filename = env[ARCHIVE_CONTROL_ENV];
+  if (!filename) { if (input.selected === executionId) throw new Error("Selected archive case is missing private transport handoff"); return; }
+  const binding = await privateBinding(filename, env.PAPERCLIP_RUNNER_E2E_TEMP_ROOT ?? "", [input.selected]);
+  if (binding.backendOrigin !== input.backendOrigin) throw new Error("Archive client belongs to another backend");
+  input.privateSecrets?.push(binding.token, binding.controlOrigin, filename);
+  const request = async (operation: "/arm" | "/receipts" | "/release", value: unknown = null) => {
+    try { const response = await fetch(`${binding.controlOrigin}${operation}`, { method: "POST", redirect: "error", headers: { authorization: `Bearer ${binding.token}`, "content-type": "application/json" }, body: JSON.stringify(value), signal: AbortSignal.timeout(10000) }); if (!response.ok) throw new Error(); const envelope: unknown = await response.json();
+      if (!envelope || typeof envelope !== "object" || Object.keys(envelope).join() !== "result") throw new Error();
+      const result = (envelope as { result: unknown }).result;
+      if (operation === "/receipts") { if (!Array.isArray(result) || !result.every(validReceipt)) throw new Error(); }
+      else if (result !== null) throw new Error();
+      return result; }
+    catch { throw new Error("Private archive transport control failed"); }
+  };
+  return { origin: binding.origin, backendOrigin: binding.backendOrigin,
+    async arm(value) { await request("/arm", value); }, async receipts() { return await request("/receipts") as ArchiveTransportReceipt[]; }, async releaseForExplicitRetry() { await request("/release"); },
+    async close() { /* Launcher owns teardown after the backend has stopped. */ } };
+}

tests/runner-e2e/chat-catalog-domains/internal-archive-partial.ts

@@ -2,12 +2,12 @@ import { expect } from "@playwright/test";
 import type { ChatCatalogCase, ChatCatalogContext } from "../chat-catalog-contract.js";
 import { pollUntil } from "../api.js";
 import { makeDefinition } from "./internal-definition.js";
-import type { ArchiveFaultTransport, ArchiveTransportReceipt } from "./internal-archive-transport.js";
+import type { ArchiveTransportReceipt } from "./internal-archive-transport.js";
 import { checks, entity, observation, observeInternal, project, sameTask, task, type Entity, type InternalObservation } from "./internal-support.js";
 interface Item extends Entity { archivedByActorType?: string; archivedByAgentId?: string; archivedByRunId?: string }
 interface PartialArchive extends InternalObservation { ownerItems: Item[]; transport: ArchiveTransportReceipt[]; phaseReply: string }
 /** Optional private launcher handle: no environment/JSON credential or fake response fallback. */
-function transport(ctx: ChatCatalogContext) { return (ctx.input as ChatCatalogContext["input"] & { archiveFaultTransport?: ArchiveFaultTransport }).archiveFaultTransport; }
+function transport(ctx: ChatCatalogContext) { return ctx.input.archiveFaultTransport; }
 function firstIsValid(o: PartialArchive) {
   const originals = o.state.originalTasks as Entity[]; const forwarded = o.transport.filter(row => row.phase === "initial" && row.forwarded && row.status === 200); const dropped = o.transport.filter(row => row.phase === "initial" && !row.forwarded);
   const archived = o.ownerItems.filter(row => originals.some(old => old.id === row.id) && !!row.archivedAt);
@@ -46,26 +46,27 @@ function archivePartial(): ChatCatalogCase {
   const base = makeDefinition("inbox-search.archive-partial-failure", async ctx => {
     const proxy = transport(ctx);
     if (!proxy || proxy.backendOrigin !== ctx.input.api.baseURL || proxy.origin === ctx.input.api.baseURL) {
-      await ctx.input.evidence("chat-catalog-archive-native-transport-prerequisite.json", { qualified: false, required: "Start owned transparent archive proxy before server bootstrap; set that server's real PAPERCLIP_API_URL to its proxy origin, retain browser/fixture API on actual backend, inject in-process archiveFaultTransport handle. Native per-item POST traffic must traverse it. Browser route interception and manufactured HTTP errors do not qualify." });
+      await ctx.input.evidence("chat-catalog-archive-native-transport-prerequisite.json", { qualified: false, required: "Start owned transparent archive proxy before server bootstrap; set that server's real PAPERCLIP_API_URL to its proxy origin, retain browser/fixture API on actual backend, inject authenticated owned cross-process archiveFaultTransport control. Native per-item POST traffic must traverse it. Browser route interception and manufactured HTTP errors do not qualify." });
       throw new Error("Missing real native archive transport binding: the single-case launcher must start the owned proxy before server bootstrap and route native PAPERCLIP_API_URL through it. No user query admitted.");
     }
     const session = await ctx.input.api.get<{ user?: { id: string } }>("/api/auth/get-session"); if (!session.user?.id) throw new Error("Archive requires actual responsible human");
     const p = await project(ctx); const originals: Entity[] = [];
     for (const title of ["Delivered launch design", "Delivered launch copy", "Delivered launch checklist"]) { const t = await task(ctx, title, p.id, { status: "done" }); await ctx.input.api.post(`/api/issues/${t.id}/comments`, { body: "Following this completed delivery in my personal inbox." }); originals.push(t); }
     await ctx.input.api.putSensitive(`/api/companies/${ctx.input.fixtures.company.id}/users/me/inbox-agent-policy`, { mode: "allowlist", allowedAgentIds: [ctx.input.fixtures.agent.id] });
-    proxy.arm({ companyId: ctx.input.fixtures.company.id, agentId: ctx.input.fixtures.agent.id, personId: session.user.id, taskIds: originals.map(row => row.id) });
+    await proxy.arm({ companyId: ctx.input.fixtures.company.id, agentId: ctx.input.fixtures.agent.id, personId: session.user.id, taskIds: originals.map(row => row.id) });
     return { project: p, task: originals[0], originalTasks: originals, personId: session.user.id };
   }, gradeArchivePartial);
   const observe = async (ctx: ChatCatalogContext, state: Record<string, unknown>): Promise<PartialArchive> => {
     const o = await observeInternal(ctx, state); const first = observation(state.firstPartial); const ids = new Set(first?.runs.map(row => row.id) ?? []);
     const comments = ctx.issue() ? await ctx.input.api.get<Entity[]>(`/api/issues/${ctx.issue()!.id}/comments?order=asc`) : [];
-    return { ...o, ownerItems: await ctx.input.api.get(`/api/companies/${ctx.input.fixtures.company.id}/issues?touchedByUserId=me&includeAgentCreatedTasks=true&limit=1000`), transport: transport(ctx)?.receipts() ?? [], phaseReply: comments.filter(row => row.authorAgentId === o.leadId && !ids.has(row.createdByRunId ?? "") && o.runs.some(run => run.id === row.createdByRunId)).map(row => row.body ?? "").join("\n") };
+    return { ...o, ownerItems: await ctx.input.api.get(`/api/companies/${ctx.input.fixtures.company.id}/issues?touchedByUserId=me&includeAgentCreatedTasks=true&limit=1000`), transport: await transport(ctx)?.receipts() ?? [], phaseReply: comments.filter(row => row.authorAgentId === o.leadId && !ids.has(row.createdByRunId ?? "") && o.runs.some(run => run.id === row.createdByRunId)).map(row => row.body ?? "").join("\n") };
   };
-  return { ...base, coverageGaps: ["native-archive-transport-launcher-wiring"], authentication: "human-session", expectedRunCount: 2, minimumRunCount: 2, expectedHumanMessageCount: 2, attemptTimeoutMs: 480000,
-    sourceFiles: [...base.sourceFiles, "chat-catalog-domains/internal-archive-partial.ts", "chat-catalog-domains/internal-archive-transport.ts", "../e2e/fetch-allowed-port.ts"],
+  return { ...base, coverageGaps: [], authentication: "human-session", expectedRunCount: 2, minimumRunCount: 2, expectedHumanMessageCount: 2, attemptTimeoutMs: 480000,
+    sourceFiles: [...base.sourceFiles, "chat-catalog-domains/internal-archive-partial.ts", "chat-catalog-domains/internal-archive-transport.ts", "../e2e/fetch-allowed-port.ts", "chat-archive-fault-control.ts"],
     async act(ctx, state) { const first = await pollUntil({ label: "Real initial partial archive turn settles", deadlineAt: Date.now() + 150000, load: () => observe(ctx, state), accept: o => o.runs.length === 1 && o.runs[0]?.status === "succeeded" && ctx.issue()?.conversationState === "waiting" });
       state.firstPartial = { ...first, state: structuredClone(state) }; await ctx.input.evidence("chat-catalog-before-partial-archive-retry.json", state.firstPartial);
-      transport(ctx)!.releaseForExplicitRetry(); const { sendChatMessage } = await import("../chat-flow.js"); await sendChatMessage(ctx.input.page, ctx.scenario.query); },
+      if (!firstIsValid(first)) throw new Error("Initial partial archive evidence failed before declared retry");
+      await transport(ctx)!.releaseForExplicitRetry(); const { sendChatMessage } = await import("../chat-flow.js"); await sendChatMessage(ctx.input.page, ctx.scenario.query); },
     observe,
     async verifyResult(ctx, state) { const source = ctx.input.page.url(); try { await ctx.input.page.goto(new URL(`/${ctx.input.fixtures.company.issuePrefix}/issues/${entity(state, "task").identifier}`, ctx.input.api.baseURL).href); await expect(ctx.input.page.getByRole("heading", { name: entity(state, "task").title!, exact: true }).first()).toBeVisible(); } finally { await ctx.input.page.goto(source); await expect(ctx.input.page.getByTestId("task-chat-composer-input")).toBeVisible(); } },
     // Launcher owns the pre-bootstrap transport lifetime; it closes only after server shutdown.

tests/runner-e2e/chat-catalog-domains/internal-archive-transport.test.ts

@@ -1,23 +1,89 @@
-import { createServer } from "node:http";
+import { createServer, request } from "node:http";
+import { randomUUID } from "node:crypto";
 import { expect, it } from "vitest";
 import { listenOnFetchAllowedPort } from "../../e2e/fetch-allowed-port.js";
 import { startArchiveFaultProxy } from "./internal-archive-transport.js";
-it("forwards actual request bytes/responses, drops third before upstream admission and releases only for retry", async () => {
-  const admitted: Array<{ path: string; body: string; actor?: string }> = [];
-  // Credential-free transport unit backend; this is not a Product E2E run or fake eval outcome.
-  const backend = createServer(async (req, res) => { const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(chunk); admitted.push({ path: req.url!, body: Buffer.concat(chunks).toString(), actor: req.headers.authorization }); res.writeHead(200, { "content-type": "application/json", "x-original-response": "preserved" }); res.end(JSON.stringify({ realBackendReceipt: admitted.length })); });
-  const port = await listenOnFetchAllowedPort(backend); const proxy = await startArchiveFaultProxy(`http://127.0.0.1:${port}`);
+const ids = () => ({ companyId: randomUUID(), agentId: randomUUID(), personId: randomUUID(), taskIds: [randomUUID(), randomUUID(), randomUUID()], runId: randomUUID() });
+it("preserves real bytes/responses, independently checks actor/run/person and drops only the third before admission", async () => {
+  const binding = ids(), admitted: Array<{ path: string; body: string; actor?: string }> = [];
+  const backend = createServer(async (req, res) => {
+    res.setHeader("content-type", "application/json");
+    if (req.url === "/api/agents/me") { res.end(JSON.stringify({ id: binding.agentId, companyId: binding.companyId })); return; }
+    if (req.url === `/api/heartbeat-runs/${binding.runId}`) { res.end(JSON.stringify({ id: binding.runId, agentId: binding.agentId, companyId: binding.companyId, responsibleUserId: binding.personId })); return; }
+    if (req.url?.startsWith("/api/heartbeat-runs/")) { res.writeHead(404); res.end("{}"); return; }
+    const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(chunk);
+    admitted.push({ path: req.url!, body: Buffer.concat(chunks).toString(), actor: req.headers.authorization }); res.writeHead(200, { "x-original-response": "preserved" }); res.end(JSON.stringify({ backendReceipt: admitted.length }));
+  });
+  const port = await listenOnFetchAllowedPort(backend), proxy = await startArchiveFaultProxy(`http://127.0.0.1:${port}`);
   try {
-    proxy.arm({ companyId: "company", agentId: "ceo", personId: "person", taskIds: ["one", "two", "three"] });
-    expect(() => proxy.releaseForExplicitRetry()).toThrow();
-    const send = (id: string) => fetch(`${proxy.origin}/api/issues/${id}/inbox-archive`, { method: "POST", headers: { authorization: "Bearer unit-only", "content-type": "application/json", "x-paperclip-run-id": "source" }, body: '{ "userId": "person" }', signal: AbortSignal.timeout(5000) });
-    const first = await send("one"); expect(first.headers.get("x-original-response")).toBe("preserved"); expect(await first.json()).toEqual({ realBackendReceipt: 1 });
-    expect(await (await send("two")).json()).toEqual({ realBackendReceipt: 2 });
-    await expect(send("three")).rejects.toThrow(); await expect(send("three")).rejects.toThrow();
-    expect(admitted).toHaveLength(2); expect(admitted[0]).toEqual({ path: "/api/issues/one/inbox-archive", body: '{ "userId": "person" }', actor: "Bearer unit-only" });
-    expect(proxy.receipts().map(row => [row.taskId, row.forwarded, row.phase])).toEqual([["one", true, "initial"], ["two", true, "initial"], ["three", false, "initial"], ["three", false, "initial"]]);
-    proxy.releaseForExplicitRetry(); expect(await (await send("three")).json()).toEqual({ realBackendReceipt: 3 });
-    expect(proxy.receipts().at(-1)).toMatchObject({ taskId: "three", phase: "retry", forwarded: true, status: 200, explicitUserId: "person", runId: "source" });
-    const ordinary = await fetch(`${proxy.origin}/api/health`); expect(ordinary.status).toBe(200); await ordinary.arrayBuffer(); expect(admitted.at(-1)?.path).toBe("/api/health");
-  } finally { await proxy.close(); backend.closeAllConnections(); await new Promise<void>(resolve => backend.close(() => resolve())); }
+    await proxy.arm({ companyId: binding.companyId, agentId: binding.agentId, personId: binding.personId, taskIds: binding.taskIds });
+    await expect(proxy.releaseForExplicitRetry()).rejects.toThrow();
+    const body = `{ "userId": "${binding.personId}" }`;
+    const send = (id: string, requestBody = body, run = binding.runId) => fetch(`${proxy.origin}/api/issues/${id}/inbox-archive`, { method: "POST", headers: { authorization: "Bearer unit-only", "content-type": "application/json", "x-paperclip-run-id": run }, body: requestBody, signal: AbortSignal.timeout(5000) });
+    const first = await send(binding.taskIds[0]!); expect(first.headers.get("x-original-response")).toBe("preserved"); expect(await first.json()).toEqual({ backendReceipt: 1 });
+    expect(await (await send(binding.taskIds[1]!)).json()).toEqual({ backendReceipt: 2 });
+    // Wrong person/run and unrelated task still reach the backend unchanged.
+    await (await send(binding.taskIds[2]!, '{ "userId": "another-person" }')).arrayBuffer();
+    await (await send(binding.taskIds[2]!, body, randomUUID())).arrayBuffer();
+    await (await send(randomUUID())).arrayBuffer();
+    expect(await proxy.receipts()).toHaveLength(2);
+    await expect(send(binding.taskIds[2]!)).rejects.toThrow(); await expect(send(binding.taskIds[2]!)).rejects.toThrow();
+    expect(admitted).toHaveLength(5); expect(admitted[0]).toEqual({ path: `/api/issues/${binding.taskIds[0]}/inbox-archive`, body, actor: "Bearer unit-only" });
+    expect((await proxy.receipts()).map(row => [row.taskId, row.forwarded, row.phase])).toEqual([[binding.taskIds[0], true, "initial"], [binding.taskIds[1], true, "initial"], [binding.taskIds[2], false, "initial"], [binding.taskIds[2], false, "initial"]]);
+    await proxy.releaseForExplicitRetry(); expect(await (await send(binding.taskIds[2]!)).json()).toEqual({ backendReceipt: 6 });
+    expect((await proxy.receipts()).at(-1)).toMatchObject({ taskId: binding.taskIds[2], phase: "retry", forwarded: true, status: 200, explicitUserId: binding.personId, runId: binding.runId });
+    await expect(proxy.releaseForExplicitRetry()).rejects.toThrow();
+  } finally { backend.closeAllConnections(); await new Promise<void>(resolve => backend.close(() => resolve())); await proxy.close(); }
+});
+it("rejects external backends and malformed/duplicate binding without admitting a fault", async () => {
+  await expect(startArchiveFaultProxy("https://example.com")).rejects.toThrow();
+  const backend = createServer((_req, res) => res.end()), port = await listenOnFetchAllowedPort(backend), proxy = await startArchiveFaultProxy(`http://127.0.0.1:${port}`);
+  try { await expect(proxy.arm({ companyId: "wrong", agentId: "wrong", personId: "person", taskIds: ["one", "one", "three"] })).rejects.toThrow(); }
+  finally { backend.closeAllConnections(); await new Promise<void>(resolve => backend.close(() => resolve())); await proxy.close(); }
+});
+
+it("forwards malformed and oversized bound requests unchanged without fault receipts", async () => {
+  const binding = ids(), admitted: string[] = [];
+  const backend = createServer(async (req, res) => {
+    const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(chunk);
+    admitted.push(Buffer.concat(chunks).toString()); res.end("backend-original");
+  });
+  const port = await listenOnFetchAllowedPort(backend), proxy = await startArchiveFaultProxy(`http://127.0.0.1:${port}`);
+  try {
+    await proxy.arm({ companyId: binding.companyId, agentId: binding.agentId, personId: binding.personId, taskIds: binding.taskIds });
+    const endpoint = `${proxy.origin}/api/issues/${binding.taskIds[0]}/inbox-archive`, headers = { "x-paperclip-run-id": binding.runId };
+    for (const body of ["not-json", "null", "[]", '{"unexpected":true}', "x".repeat(70000)]) {
+      expect(await (await fetch(endpoint, { method: "POST", headers, body })).text()).toBe("backend-original"); expect(admitted.at(-1)).toBe(body);
+    }
+    // Unknown-length chunked input crosses the inspection bound without losing
+    // the already-buffered prefix or applying a synthetic transport failure.
+    const body = "y".repeat(90000);
+    await new Promise<void>((resolve, reject) => {
+      const outgoing = request(endpoint, { method: "POST", headers }, response => { response.resume(); response.on("end", resolve); });
+      outgoing.on("error", reject); outgoing.write(body.slice(0, 50000)); setTimeout(() => outgoing.end(body.slice(50000)), 5);
+    });
+    expect(admitted.at(-1)).toBe(body); expect(await proxy.receipts()).toEqual([]);
+    const escaped = await new Promise<number>((resolve, reject) => {
+      const outgoing = request(proxy.origin, { path: "http://example.invalid/not-owned" }, response => { response.resume(); response.on("end", () => resolve(response.statusCode!)); }); outgoing.on("error", reject); outgoing.end();
+    });
+    expect(escaped).toBe(400); expect(admitted).toHaveLength(6);
+  } finally { backend.closeAllConnections(); await new Promise<void>(resolve => backend.close(() => resolve())); await proxy.close(); }
+});
+
+it("passes the owned backend upgrade stream unchanged and closes its owned sockets", async () => {
+  const backend = createServer(); let seenHeader: string | undefined;
+  backend.on("upgrade", (req, socket, head) => {
+    seenHeader = req.headers["x-offline-identity"] as string;
+    socket.write("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: offline-echo\r\n\r\n");
+    if (head.length) socket.write(head); socket.on("data", bytes => socket.write(bytes)); socket.on("end", () => socket.destroy());
+  });
+  const port = await listenOnFetchAllowedPort(backend), proxy = await startArchiveFaultProxy(`http://127.0.0.1:${port}`);
+  try {
+    expect(await new Promise<string>((resolve, reject) => {
+      const outgoing = request(`${proxy.origin}/actual-native-control`, { headers: { connection: "Upgrade", upgrade: "offline-echo", "x-offline-identity": "opaque-preserved" } });
+      outgoing.setTimeout(5000, () => { outgoing.destroy(); reject(new Error("Offline upgrade timed out")); }); outgoing.on("error", reject);
+      outgoing.on("upgrade", (_response, socket) => { socket.on("error", reject); socket.once("data", bytes => { resolve(bytes.toString()); socket.destroy(); }); socket.write("unchanged-native-frame"); }); outgoing.end();
+    })).toBe("unchanged-native-frame");
+    expect(seenHeader).toBe("opaque-preserved"); expect(await proxy.receipts()).toEqual([]);
+  } finally { await proxy.close(); await new Promise<void>(resolve => backend.close(() => resolve())); }
 });

tests/runner-e2e/chat-catalog-domains/internal-archive-transport.ts

@@ -1,40 +1,86 @@
-import { createServer, request as httpRequest } from "node:http";
+import { createServer, request as httpRequest, type IncomingMessage, type ServerResponse } from "node:http";
 import { createHash, randomUUID } from "node:crypto";
+import { connect, type Socket } from "node:net";
 import { listenOnFetchAllowedPort } from "../../e2e/fetch-allowed-port.js";
 export interface ArchiveTransportReceipt { requestId: string; sequence: number; taskId: string; phase: "initial" | "retry"; requestSha256: string; runId?: string; explicitUserId?: string; forwarded: boolean; status?: number; completedAt: string }
-export interface ArchiveFaultTransport { origin: string; backendOrigin: string; arm(input: { companyId: string; agentId: string; personId: string; taskIds: string[] }): void; receipts(): ArchiveTransportReceipt[]; releaseForExplicitRetry(): void; close(): Promise<void> }
-/** Started by the single-case launcher BEFORE server bootstrap; the actual server API origin must point here. */
+export interface ArchiveBinding { companyId: string; agentId: string; personId: string; taskIds: string[] }
+export interface ArchiveFaultTransport { origin: string; backendOrigin: string; arm(input: ArchiveBinding): Promise<void>; receipts(): Promise<ArchiveTransportReceipt[]>; releaseForExplicitRetry(): Promise<void>; close(): Promise<void> }
+const uuid = (value: unknown): value is string => typeof value === "string" && /^[a-f0-9]{8}-(?:[a-f0-9]{4}-){3}[a-f0-9]{12}$/i.test(value);
+export function validArchiveBinding(value: ArchiveBinding) { return !!value && Object.keys(value).sort().join() === "agentId,companyId,personId,taskIds" && uuid(value.companyId) && uuid(value.agentId) && typeof value.personId === "string" && /^[A-Za-z0-9_-]{8,128}$/.test(value.personId) && Array.isArray(value.taskIds) && value.taskIds.length === 3 && value.taskIds.every(uuid) && new Set(value.taskIds).size === 3; }
+/** Started by the single-case launcher before backend bootstrap. No public action
+ * or actor is invented: identity probes use the original authenticated request. */
 export async function startArchiveFaultProxy(backendOrigin: string): Promise<ArchiveFaultTransport> {
-  const upstream = new URL(backendOrigin); if (upstream.protocol !== "http:" || !["127.0.0.1", "localhost"].includes(upstream.hostname) || upstream.pathname !== "/") throw new Error("Archive proxy requires the actual owned local HTTP backend origin");
-  const receipts: ArchiveTransportReceipt[] = []; let binding: { companyId: string; agentId: string; personId: string; taskIds: string[] } | undefined; let released = false; let failedTaskId: string | undefined;
-  let serial = Promise.resolve(); const successes = new Set<string>();
-  const server = createServer(async (req, res) => {
+  const upstream = new URL(backendOrigin);
+  if (upstream.protocol !== "http:" || upstream.hostname !== "127.0.0.1" || !upstream.port || upstream.pathname !== "/" || upstream.search || upstream.hash || upstream.username || upstream.password) throw new Error("Archive proxy requires the exact owned loopback backend origin");
+  const receipts: ArchiveTransportReceipt[] = []; let binding: ArchiveBinding | undefined; let released = false; let failedTaskId: string | undefined;
+  let serial = Promise.resolve(); const shutdown = new AbortController(); let closing = false; const successes = new Set<string>(); const sockets = new Set<Socket>();
+  const target = (url: string | undefined) => url?.startsWith("/") && !url.startsWith("//") && new URL(url, upstream).origin === upstream.origin ? new URL(url, upstream) : null;
+  const forward = (req: IncomingMessage, res: ServerResponse, bytes?: Buffer, receipt?: ArchiveTransportReceipt, continueBody = false) => new Promise<void>(resolve => {
+    if (closing) { req.socket.destroy(); resolve(); return; }
+    const url = target(req.url); if (!url) { res.writeHead(400); res.end(); resolve(); return; }
+    const outgoing = httpRequest(url, { method: req.method, headers: { ...req.headers, host: upstream.host } }, incoming => {
+      if (receipt) receipt.status = incoming.statusCode;
+      res.writeHead(incoming.statusCode!, incoming.headers); incoming.pipe(res);
+      const done = () => { if (receipt && !receipt.completedAt) { receipt.completedAt = new Date().toISOString(); receipts.push(receipt); if (incoming.statusCode! >= 200 && incoming.statusCode! < 300) successes.add(receipt.taskId); } resolve(); };
+      incoming.on("end", done); incoming.on("error", () => { req.socket.destroy(); resolve(); });
+    }); outgoing.on("socket", socket => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); }); outgoing.on("error", () => { req.socket.destroy(); resolve(); });
+    if (continueBody) { if (bytes) outgoing.write(bytes); req.pipe(outgoing); } else if (bytes) outgoing.end(bytes); else req.pipe(outgoing);
+  });
+  const identity = async (req: IncomingMessage, runId: string, personId: string) => {
+    try {
+      const headers: Record<string, string> = {}; for (const name of ["authorization", "x-paperclip-run-id"]) if (typeof req.headers[name] === "string") headers[name] = req.headers[name]!;
+      const get = async (route: string) => { const response = await fetch(new URL(route, upstream), { headers, redirect: "error", signal: AbortSignal.any([shutdown.signal, AbortSignal.timeout(5000)]) }); return response.ok ? await response.json() as Record<string, unknown> : null; };
+      const agent = await get("/api/agents/me"), run = await get(`/api/heartbeat-runs/${runId}`);
+      return agent?.id === binding?.agentId && agent?.companyId === binding?.companyId && run?.id === runId && run?.agentId === binding?.agentId && run?.companyId === binding?.companyId && run?.responsibleUserId === personId;
+    } catch { return false; }
+  };
+  const server = createServer((req, res) => { void (async () => {
+    if (!target(req.url)) { res.writeHead(400); res.end(); return; }
     const match = req.method === "POST" ? /^\/api\/issues\/([^/?]+)\/inbox-archive(?:\?.*)?$/.exec(req.url ?? "") : null;
     const taskId = match?.[1];
-    if (!taskId || !binding?.taskIds.includes(taskId)) {
-      const outgoing = httpRequest(new URL(req.url ?? "/", upstream), { method: req.method, headers: { ...req.headers, host: upstream.host } }, incoming => { res.writeHead(incoming.statusCode!, incoming.headers); incoming.pipe(res); }); outgoing.on("error", () => req.socket.destroy()); req.pipe(outgoing); return;
+    if (!taskId || !binding?.taskIds.includes(taskId)) { await forward(req, res); return; }
+    // Large/noncanonical requests are ordinary backend traffic, never fault
+    // candidates. Keep their original stream and bytes rather than rejecting.
+    if (Number(req.headers["content-length"] ?? 0) > 65536) { await forward(req, res); return; }
+    const chunks: Buffer[] = []; let size = 0, oversized = false;
+    for await (const chunk of req.iterator({ destroyOnReturn: false })) {
+      size += chunk.length; chunks.push(chunk);
+      if (size > 65536) { oversized = true; break; }
     }
-    const chunks: Buffer[] = []; let size = 0; for await (const chunk of req) { size += chunk.length; if (size > 65536) { req.socket.destroy(); return; } chunks.push(chunk); } const bytes = Buffer.concat(chunks);
-    // Queue unchanged requests; do not split a batch or manufacture a response. A third distinct item is dropped only after two real successful responses.
+    const bytes = Buffer.concat(chunks);
+    if (oversized) { await forward(req, res, bytes, undefined, true); return; }
     const prior = serial; let finish!: () => void; serial = new Promise<void>(resolve => { finish = resolve; }); await prior;
     try {
-      let explicitUserId: string | undefined; try { const value = JSON.parse(bytes.toString()); if (typeof value.userId === "string") explicitUserId = value.userId; } catch { /* Forward validation errors unchanged. */ }
-      const receipt: ArchiveTransportReceipt = { requestId: randomUUID(), sequence: receipts.length + 1, taskId, phase: released ? "retry" : "initial", requestSha256: createHash("sha256").update(bytes).digest("hex"), runId: typeof req.headers["x-paperclip-run-id"] === "string" ? req.headers["x-paperclip-run-id"] : undefined, explicitUserId, forwarded: false, completedAt: "" };
+      let explicitUserId: string | undefined, validBody = false;
+      try {
+        const value: unknown = JSON.parse(bytes.toString());
+        if (value && typeof value === "object" && !Array.isArray(value)) {
+          const record = value as Record<string, unknown>;
+          validBody = Object.keys(record).every(key => key === "userId") && (record.userId === undefined || typeof record.userId === "string");
+          if (typeof record.userId === "string") explicitUserId = record.userId;
+        }
+      } catch { /* Backend validation remains authoritative. */ }
+      const runId = req.headers["x-paperclip-run-id"];
+      if (!validBody || typeof runId !== "string" || !uuid(runId) || explicitUserId !== undefined && explicitUserId !== binding.personId || !await identity(req, runId, binding.personId)) { await forward(req, res, bytes); return; }
+      const receipt: ArchiveTransportReceipt = { requestId: randomUUID(), sequence: receipts.length + 1, taskId, phase: released ? "retry" : "initial", requestSha256: createHash("sha256").update(bytes).digest("hex"), runId, explicitUserId, forwarded: false, completedAt: "" };
       if (!released && (failedTaskId === taskId || !successes.has(taskId) && successes.size === 2)) { failedTaskId = taskId; receipt.completedAt = new Date().toISOString(); receipts.push(receipt); req.socket.destroy(); return; }
-      receipt.forwarded = true;
-      await new Promise<void>(resolve => {
-        const outgoing = httpRequest(new URL(req.url!, upstream), { method: req.method, headers: { ...req.headers, host: upstream.host } }, incoming => {
-          receipt.status = incoming.statusCode; res.writeHead(incoming.statusCode!, incoming.headers); incoming.pipe(res);
-          incoming.on("end", () => { receipt.completedAt = new Date().toISOString(); receipts.push(receipt); if (incoming.statusCode! >= 200 && incoming.statusCode! < 300) successes.add(taskId); resolve(); });
-          incoming.on("error", () => { req.socket.destroy(); resolve(); });
-        }); outgoing.on("error", () => { req.socket.destroy(); resolve(); }); outgoing.end(bytes);
-      });
+      receipt.forwarded = true; await forward(req, res, bytes, receipt);
     } finally { finish(); }
+  })().catch(() => { req.socket.destroy(); }); });
+  server.on("connection", socket => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); });
+  // Native control transport may upgrade on the same API origin. Forward only
+  // to this exact backend; retain no raw headers, credentials or frame bytes.
+  server.on("upgrade", (req, socket, head) => {
+    if (!target(req.url)) { socket.destroy(); return; }
+    const outgoing = connect(Number(upstream.port), "127.0.0.1", () => {
+      outgoing.write(`${req.method} ${req.url} HTTP/${req.httpVersion}\r\n${Object.entries(req.headers).map(([name, value]) => `${name}: ${name === "host" ? upstream.host : Array.isArray(value) ? value.join(", ") : value}`).join("\r\n")}\r\n\r\n`);
+      if (head.length) outgoing.write(head); socket.pipe(outgoing); outgoing.pipe(socket);
+    }); sockets.add(outgoing); outgoing.on("close", () => sockets.delete(outgoing)); outgoing.on("error", () => socket.destroy()); socket.on("error", () => outgoing.destroy()); socket.on("close", () => outgoing.destroy()); outgoing.on("close", () => socket.destroy());
   });
   const port = await listenOnFetchAllowedPort(server);
   return { origin: `http://127.0.0.1:${port}`, backendOrigin: upstream.origin,
-    arm(input) { if (binding || new Set(input.taskIds).size !== 3) throw new Error("Transport fault must be armed once for exactly three actual task IDs"); binding = structuredClone(input); },
-    receipts: () => structuredClone(receipts), releaseForExplicitRetry() { if (!failedTaskId || successes.size !== 2) throw new Error("Cannot release fault before the actual two-success/one-drop boundary"); released = true; },
-    async close() { server.closeAllConnections(); await new Promise<void>(resolve => server.close(() => resolve())); },
+    async arm(input) { if (binding || !validArchiveBinding(input)) throw new Error("Transport fault must be armed once for exact owned actor/person and three UUID tasks"); binding = structuredClone(input); },
+    async receipts() { return structuredClone(receipts); }, async releaseForExplicitRetry() { if (released || !failedTaskId || successes.size !== 2) throw new Error("Cannot release fault before actual two-success/one-drop boundary"); released = true; },
+    async close() { closing = true; shutdown.abort(); for (const socket of sockets) socket.destroy(); server.closeAllConnections(); await new Promise<void>(resolve => server.close(() => resolve())); await serial; },
   };
 }

tests/runner-e2e/chat-flow.ts

@@ -1,3 +1,5 @@
+import { archiveFaultClient } from "./chat-archive-fault-control.js";
+import type { ArchiveFaultTransport } from "./chat-catalog-domains/internal-archive-transport.js";
 import { prepareChatCatalogRespondent } from "./chat-catalog-respondent.js";
 import type { ChatPrivateInputs } from "./chat-private-inputs.js";
 import { chatCatalogAllowedRunStatuses } from "./chat-catalog-run-policy.js";
@@ -359,6 +361,8 @@ export function assertChatReassignment(input: {
 }
 
 export interface ChatFlowInput {
+  /** Test-host-only owned transparent fault transport; never model state. */
+  archiveFaultTransport?: ArchiveFaultTransport;
   page: Page;
   api: RunnerApi;
   fixtures: LiveFixtureValues;
@@ -375,6 +379,7 @@ export interface ChatFlowInput {
   check?: (id: string, passed: boolean, detail: string) => void;
 }
 export async function runChatFlow(input: ChatFlowInput) {
+  input.archiveFaultTransport ??= await archiveFaultClient({ selected: input.execution.id, backendOrigin: input.api.baseURL, privateSecrets: input.privateSecrets });
   if (input.execution.suite.id === CHAT_FEATURE_SUITE) await prepareChatCatalogRespondent(input, findChatCatalogCase(input.execution.task.id));
   const { page, api, fixtures: f, execution, nonce } = input;
   const chatPath = `/api/companies/${f.company.id}/chats/${f.agent.id}`;

tests/runner-e2e/launch.ts

@@ -1,3 +1,4 @@
+import { ARCHIVE_CONTROL_ENV, ARCHIVE_SCENARIO, startArchiveFaultControl } from "./chat-archive-fault-control.js";
 import { abnormalFinalizationResult } from "./abnormal-finalization.js";
 import { CHAT_ORACLE_KEY_ENV } from "./chat-private-oracle.js";
 import { stagePublishedChatInstall, ChatInstalledCleanupError } from "./chat-installed-lane.js";
@@ -511,6 +512,7 @@ async function runAttempt(input: {
   let attemptSecrets: string[] = [];
   let processCleanupFailed = false;
   let startupReceiptPath: string | undefined;
+  let archiveControl: Awaited<ReturnType<typeof startArchiveFaultControl>> | undefined;
   let gateway: Awaited<ReturnType<EvalOperations["start"]>> | undefined;
   const rawCleanupResults: Array<{ cleanup: string; synthetic: boolean; status: string }> = [];
   try {
@@ -563,6 +565,7 @@ async function runAttempt(input: {
     if (needsPrivateInputs && executions.length !== 1) throw new Error("Private case inputs require one explicitly selected execution");
     await mkdir(selectedPrivateInputsDirectory, { mode: 0o700 });
     const privateInputs = needsPrivateInputs ? stageChatPrivateInputs({ execution, descriptor: selectedDescriptor, privateDir: selectedPrivateInputsDirectory, env: process.env, secrets: credentials }) : undefined;
+    if (selectedDescriptor?.scenarioId === ARCHIVE_SCENARIO) archiveControl = await startArchiveFaultControl({ selected: executions.map(row => row.id), temporaryRoot, backendOrigin: `http://127.0.0.1:${port}`, privateSecrets: credentials });
     const diagnosticRegistry = stagePrivateDiagnostics(temporaryRoot, executions.map(row => row.id), credentials);
     if (installedCli) await writeFile(path.join(privateDir, "installed-cli-admission.json"), `${JSON.stringify(installedCli, null, 2)}\n`, { mode: 0o600 });
     if (installedPlugin) await writeFile(path.join(privateDir, "installed-daytona-plugin-admission.json"), `${JSON.stringify(installedPlugin, null, 2)}\n`, { mode: 0o600 });
@@ -573,6 +576,7 @@ async function runAttempt(input: {
     const childEnv: NodeJS.ProcessEnv = {
       ...buildRunnerE2EProcessEnvironment(sourceEnvironment, executions),
       [CHAT_PRIVATE_INPUTS_ENV]: privateInputs?.filename,
+      [ARCHIVE_CONTROL_ENV]: archiveControl?.filename,
       [PRIVATE_DIAGNOSTICS_ENV]: diagnosticRegistry,
       PATH: providerPath,
       PAPERCLIP_RUNNER_E2E_EXECUTION_IDS: JSON.stringify(
@@ -926,6 +930,9 @@ async function runAttempt(input: {
     }
     return [...publishedResults];
   } finally {
+    // runProcess has completed its owned Playwright/server process shutdown.
+    let archiveCleanupError: unknown;
+    try { await archiveControl?.close(); } catch (error) { archiveCleanupError = error; processCleanupFailed = true; }
     let gatewayCleanupError: unknown;
     try { await gateway?.close(); } catch (error) { gatewayCleanupError = error; }
     if (!processCleanupFailed) reapNewDetachedDarwinSharedMemory(sharedMemoryBaseline);
@@ -972,7 +979,7 @@ async function runAttempt(input: {
         `Refusing to remove unexpected temporary path ${temporaryRoot}`,
       );
     }
-    if (gatewayCleanupError) cleanupError = new AggregateError([cleanupError, gatewayCleanupError].filter(Boolean), "Eval gateway cleanup failed");
+    if (gatewayCleanupError || archiveCleanupError) cleanupError = new AggregateError([cleanupError, gatewayCleanupError, archiveCleanupError].filter(Boolean), "Eval owned gateway/transport cleanup failed");
     if (startupReceiptPath) {
       const receipt = JSON.parse(await readFile(startupReceiptPath, "utf8"));
       receipt.status = cleanupError ? "installed_launch_chain_cleanup_failed" : "installed_launch_chain_passed";

tests/runner-e2e/server.ts

@@ -1,3 +1,4 @@
+import { archiveServerEnvironment } from "./chat-archive-fault-control.js";
 import { CHAT_INSTALL_PLAN, refreshInstalledChatPins, publishInstalledChatReceipt } from "./chat-installed-lane.js";
 import { parseFinalizationCrashBinding, readFinalizationCrashBoundary, type FinalizationCrashBinding } from "./finalization-crash-boundary.js";
 import { parseRestartRunnerIdentity, type RestartRunnerIdentity } from "./process-tree-owner.js";
@@ -51,9 +52,11 @@ const restartTimeoutMs = 180_000;
 const gracefulStopTimeoutMs = 30_000;
 const installedRelease = process.env.PAPERCLIP_RUNNER_E2E_INSTALLED_CLI
   ? installedReleaseLaunch(process.env.PAPERCLIP_RUNNER_E2E_INSTALLED_CLI) : null;
+const archiveEnvironment = await archiveServerEnvironment({ selected: executionIds, temporaryRoot, backendOrigin: `http://127.0.0.1:${port}`, env: process.env });
 const serverEnvironment = buildPaperclipServerEnvironment(installedRelease
-  ? installedReleaseEnvironment(process.env, repositoryRoot, path.join(temporaryRoot, "provider-bin"))
-  : process.env, {
+  ? installedReleaseEnvironment(archiveEnvironment.source, repositoryRoot, path.join(temporaryRoot, "provider-bin"))
+  : archiveEnvironment.source, {
+  ...archiveEnvironment.overrides,
   NODE_ENV: "test",
   PORT: port,
   // Keep provider caches attempt-private without changing Playwright's browser