test: prove every large-journal stimulus call completed

Require unique paired process receipts with the exact output byte count and digest before the first continuation. Reject repeated, unrelated, failed, unpaired, and mismatched execution receipts.

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

tests/runner-e2e/journal-evidence.test.ts

@@ -1,26 +1,61 @@
 import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises";
+import { createHash } from "node:crypto";
 import { tmpdir } from "node:os";
 import path from "node:path";
 import { describe, expect, it } from "vitest";
-import { largeJournalEvidence } from "./journal-evidence.js";
+import { completedJournalStimulusCalls, largeJournalEvidence } from "./journal-evidence.js";
+
+const output = `journal-continuity-${"x".repeat(65_000)}\n`;
+function call(index: number) {
+  const event = { schema: "paperclip.prp.event.v1", runId: "expected-run", sourceKind: "runner", sourceInstanceId: "runner", normalizedSessionId: "session" };
+  const payload = { schema: "paperclip.tool.execution.v1", transport: "process", operation: "execute", executionId: `exec-${index}`, name: "python3 fixture" };
+  return [
+    { ...event, eventType: "tool.execution.started", sourceSeq: index * 2, payload: { ...payload, status: "running" } },
+    { ...event, eventType: "tool.execution.completed", sourceSeq: index * 2 + 1, payload: { ...payload, status: "completed", exitCode: 0, outputBytes: Buffer.byteLength(output), outputDigest: `sha256:${createHash("sha256").update(output).digest("hex")}` } },
+  ];
+}
 
 describe("large journal boundary evidence", () => {
-  it("requires one correctly owned journal over the boundary and returns only sizes", async () => {
+  it("requires one correctly owned journal over the boundary and an independent completed-call count", async () => {
     const root = await mkdtemp(path.join(tmpdir(), "journal-evidence-"));
     const directory = path.join(root, "a".repeat(64), "control-plane");
     await mkdir(directory, { recursive: true });
     const filename = path.join(directory, "control-plane-state.json");
     const state = { schema: "paperclip.runner.durable.control-plane-state.v1", identity: { runId: "expected-run" }, commands: [] };
+    const input = { stateRoot: root, runId: "expected-run", minimumBytes: 1024, minimumCompletedStimulusCalls: 2 };
     try {
       await writeFile(filename, JSON.stringify(state));
-      await expect(largeJournalEvidence({ stateRoot: root, runId: "expected-run", minimumBytes: 1024 })).rejects.toThrow("boundary not reached");
+      await expect(largeJournalEvidence(input)).rejects.toThrow("boundary not reached");
       await writeFile(filename, JSON.stringify({ ...state, commands: [{ payload: "x".repeat(2048) }] }));
-      const evidence = await largeJournalEvidence({ stateRoot: root, runId: "expected-run", minimumBytes: 1024 });
-      expect(Object.keys(evidence)).toEqual(["journalBytes", "minimumBytes"]);
-      expect(evidence.journalBytes).toBeGreaterThan(2048);
-      await expect(largeJournalEvidence({ stateRoot: root, runId: "another-run", minimumBytes: 1024 })).rejects.toThrow("boundary not reached");
-    } finally {
-      await rm(root, { recursive: true, force: true });
+      await expect(largeJournalEvidence(input)).rejects.toThrow("stimulus incomplete");
+      await writeFile(filename, JSON.stringify({ ...state, committedEvents: [...call(1), ...call(2)].map(event => ({ envelope: { payload: event } })) }));
+      const evidence = await largeJournalEvidence(input);
+      expect(Object.keys(evidence)).toEqual(["journalBytes", "minimumBytes", "completedStimulusCalls", "minimumCompletedStimulusCalls"]);
+      expect(evidence.completedStimulusCalls).toBe(2);
+      await expect(largeJournalEvidence({ ...input, runId: "another-run" })).rejects.toThrow("boundary not reached");
+    } finally { await rm(root, { recursive: true, force: true }); }
+  });
+
+  it("does not count repeated receipts, unmatched completion, wrong runs, failed or unrelated output", () => {
+    const pair = call(1);
+    expect(completedJournalStimulusCalls([...pair, ...pair], "expected-run")).toBe(1);
+    expect(completedJournalStimulusCalls([pair[1]], "expected-run")).toBe(0);
+    expect(completedJournalStimulusCalls(pair, "another-run")).toBe(0);
+    expect(completedJournalStimulusCalls([pair[0], { ...pair[1], payload: { ...pair[1]!.payload, exitCode: 1 } }], "expected-run")).toBe(0);
+    expect(completedJournalStimulusCalls([pair[0], { ...pair[1], payload: { ...pair[1]!.payload, outputDigest: "sha256:unrelated" } }], "expected-run")).toBe(0);
+  });
+
+  it("requires the same runner, session, execution and command with ordered source sequence", () => {
+    const [start, complete] = call(1);
+    for (const changed of [
+      { ...complete, sourceInstanceId: "another-runner" },
+      { ...complete, normalizedSessionId: "another-session" },
+      { ...complete, sourceSeq: start!.sourceSeq },
+      { ...complete, payload: { ...complete!.payload, executionId: "another-exec" } },
+      { ...complete, payload: { ...complete!.payload, name: "another-command" } },
+      { ...complete, payload: { ...complete!.payload, outputBytes: 1 } },
+    ]) {
+      expect(completedJournalStimulusCalls([start, changed], "expected-run")).toBe(0);
     }
   });
 });

tests/runner-e2e/journal-evidence.ts

@@ -1,15 +1,51 @@
 import { lstat, readdir, readFile } from "node:fs/promises";
+import { createHash } from "node:crypto";
 import path from "node:path";
 // The fixture reads bounded private evidence independently of production admission.
 const MAX_JOURNAL_EVIDENCE_BYTES = 192 * 1024 * 1024;
+const STIMULUS_OUTPUT = `journal-continuity-${"x".repeat(65_000)}\n`;
+const STIMULUS_DIGEST = `sha256:${createHash("sha256").update(STIMULUS_OUTPUT).digest("hex")}`;
+const object = (value: unknown): Record<string, unknown> => value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
 
-/** Read-only size oracle; retained journal contents never enter public evidence. */
+/** Count independently completed process calls, never repeated receipts or prose. */
+export function completedJournalStimulusCalls(events: unknown[], runId: string): number {
+  const starts = new Map<string, { name: unknown; sourceSeq: number }>();
+  const completed = new Set<string>();
+  const executions = events.map(object).filter(event => {
+    const payload = object(event.payload);
+    return event.schema === "paperclip.prp.event.v1" && event.runId === runId && event.sourceKind === "runner" &&
+      typeof event.sourceInstanceId === "string" && typeof event.normalizedSessionId === "string" &&
+      Number.isSafeInteger(event.sourceSeq) && Number(event.sourceSeq) > 0 && payload.schema === "paperclip.tool.execution.v1" &&
+      payload.transport === "process" && payload.operation === "execute" && typeof payload.executionId === "string" &&
+      typeof payload.name === "string";
+  });
+  const key = (event: Record<string, unknown>) => JSON.stringify([event.sourceInstanceId, event.normalizedSessionId, object(event.payload).executionId]);
+  for (const event of executions) {
+    const payload = object(event.payload);
+    if (event.eventType === "tool.execution.started" && payload.status === "running") {
+      starts.set(key(event), { name: payload.name, sourceSeq: Number(event.sourceSeq) });
+    }
+  }
+  for (const event of executions) {
+    const payload = object(event.payload);
+    const start = starts.get(key(event));
+    if (event.eventType === "tool.execution.completed" && start && start.sourceSeq < Number(event.sourceSeq) &&
+      payload.name === start.name && payload.status === "completed" && payload.exitCode === 0 &&
+      payload.outputBytes === Buffer.byteLength(STIMULUS_OUTPUT) && payload.outputDigest === STIMULUS_DIGEST) {
+      completed.add(key(event));
+    }
+  }
+  return completed.size;
+}
+
+/** Read-only boundary and execution oracle; raw journal contents stay private. */
 export async function largeJournalEvidence(input: {
   stateRoot: string;
   runId: string;
   minimumBytes: number;
-}): Promise<{ journalBytes: number; minimumBytes: number }> {
-  const matches: number[] = [];
+  minimumCompletedStimulusCalls: number;
+}): Promise<{ journalBytes: number; minimumBytes: number; completedStimulusCalls: number; minimumCompletedStimulusCalls: number }> {
+  const matches: Array<{ bytes: number; calls: number }> = [];
   for (const entry of await readdir(input.stateRoot, { withFileTypes: true })) {
     if (!entry.isDirectory() || !/^[a-f0-9]{64}$/.test(entry.name)) continue;
     const filename = path.join(input.stateRoot, entry.name, "control-plane", "control-plane-state.json");
@@ -18,13 +54,18 @@ export async function largeJournalEvidence(input: {
     const state = JSON.parse(await readFile(filename, "utf8")) as {
       schema?: string;
       identity?: { runId?: string };
+      committedEvents?: unknown[];
     };
     if (state.schema === "paperclip.runner.durable.control-plane-state.v1" && state.identity?.runId === input.runId) {
-      matches.push(metadata.size);
+      const events = (state.committedEvents ?? []).map(entry => object(object(entry).envelope).payload);
+      matches.push({ bytes: metadata.size, calls: completedJournalStimulusCalls(events, input.runId) });
     }
   }
-  if (matches.length !== 1 || matches[0]! <= input.minimumBytes) {
+  if (matches.length !== 1 || matches[0]!.bytes <= input.minimumBytes) {
     throw new Error(`Large-journal boundary not reached: expected one matching journal above ${input.minimumBytes} bytes; observed ${JSON.stringify(matches)}`);
   }
-  return { journalBytes: matches[0]!, minimumBytes: input.minimumBytes };
+  if (matches[0]!.calls < input.minimumCompletedStimulusCalls) {
+    throw new Error(`Large-journal stimulus incomplete: expected at least ${input.minimumCompletedStimulusCalls} completed calls; observed ${matches[0]!.calls}`);
+  }
+  return { journalBytes: matches[0]!.bytes, minimumBytes: input.minimumBytes, completedStimulusCalls: matches[0]!.calls, minimumCompletedStimulusCalls: input.minimumCompletedStimulusCalls };
 }

tests/runner-e2e/runner.spec.ts

@@ -1551,6 +1551,7 @@ for (const execution of executions) {
                 stateRoot: path.join(process.env.PAPERCLIP_HOME!, "instances", process.env.PAPERCLIP_INSTANCE_ID!, "runtime", "paperclip-runner", "durable-sessions"),
                 runId: chronologicalRuns.at(-1)!.id,
                 minimumBytes: 2 * 1024 * 1024,
+                minimumCompletedStimulusCalls: 240,
               })
             : undefined;
           if (journalEvidence) {