Merge landed external master carry into PR 18395

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

* commit '5b137b5cdaf968b6545a7ac7a3137849240fd196':
  perf(external-objects): schedule scoped refresh obligations (#18384)
  test: wait for native work and accounting before Stop redirect (#18463)

tests/runner-e2e/PRODUCT-JOURNEYS.md

@@ -34,10 +34,37 @@ never start a server or depend on a tenant filesystem.
 | `build-revise` | Native agent; two work runs; qualified Docker artifact oracle | Browser-download both ZIP versions, independently exercise source/CLI behavior, verify different hashes and preserve the original download |
 | `protected-actions` | Native agent; four work runs; authenticated reachable review-service fixture | Approve one action and decline another on separate owned tasks; verify zero calls before either decision, one after approval, zero after decline and a truthful saved explanation |
 | `create-skill-studio` | Native agent with skill-write permission; one work run | Create a uniquely named company skill, open its activity card, edit in Studio, reload and verify task-card readback |
-| `stop-redirect` | Native agent; three work runs; qualified Docker oracle | Download a verified checkpoint, Stop an active revision, require native cancellation acknowledgement, submit a new direction once and verify completion without original-run activity |
+| `stop-redirect` | Native agent; three work runs; qualified Docker oracle | Download a verified checkpoint, observe an active native command with current-turn usage, Stop it, verify cancellation and settled accounting, then send one new direction and complete without original-run activity |
 | `followup-while-running` | Native agent; a maximum of two work runs | Save a follow-up while an independently observed native shell command is still active, then require one attributed final reply with the late requirement after reload; in-turn steering and one normal successor are accepted |
 | `cloud-onboarding` | Signed-in admitted Cloud account; fresh workspace; qualified bounded provider route | Create workspace in the portal, connect API key and native Codex agent in the wizard, complete a task, download and independently check its artifact |
 
+The deployment-targeted Stop journey asks the revision to finish one harmless
+`printf` readiness command, then run `sleep 90` once in a separate shell call
+before changing files. The readiness call lets a provider report its first
+current-turn usage before the long hold starts. Running the hold first can leave
+no complete usage until that command has already ended. Startup alone cannot trigger Stop. An independent observer
+requires both a still-running bound native command and a complete, nonzero usage
+report for its current turn. After Stop, native cancellation and exact lease
+cleanup remain required. Before the redirect, the original run must have a
+complete accounting receipt, no pending accounting and a saved accounting
+timestamp. Receipt sources and sequences remain recorded, but controller changes
+and clock order cannot substitute for the run's actual settlement state. A receipt
+that was already ready before Stop is accepted only when the exact cancelled run
+is actually accounted and no longer pending. Canonical usage receipts, including
+raw, partial and complete forms, do not count as continued work when their API row
+and envelope match the observed run, company and agent, with consistent source
+event IDs, sequence numbers and instance IDs. An accounting source can change
+after Stop; its receipt does not need the pre-Stop Runner instance or accounting
+source and sequence. Pre-Stop work still requires the current Runner and turn.
+Missing context or mismatched source event IDs, sequence numbers, instances or
+envelope versions keep the usage row in the strict work evidence. Tools, messages
+and other item events still count, including work that arrives between
+cancellation and accounting settlement.
+These waits share the original journey deadline.
+The three-run limit, budget and original download checks remain unchanged.
+Missing or partial accounting never counts as free work. Local evidence controls
+do not establish a deployed pass.
+
 The question-resumption journey uses the same natural-language scenario, browser
 answer helper and behavioral grader as the classic `question-resume` suite. Native
 reference cards may expose a text field or an explicit canonical written-answer

tests/runner-e2e/native-event-envelope.ts

@@ -8,3 +8,22 @@ export function isValidNativePrpEnvelope(
     (envelope.schema === "paperclip.prp.event.v2" && version === 2);
   return validSchema && protocolSchemaVersion === version;
 }
+
+export interface NativePrpRunBinding {
+  runId: string;
+  companyId: string;
+  agentId: string;
+}
+
+export function isBoundNativePrpEventRow(row: Record<string, any>, binding?: NativePrpRunBinding, sourceInstanceId?: string): boolean {
+  const event = row.payload?.prpEvent;
+  return Boolean(binding && [binding.runId, binding.companyId, binding.agentId]
+    .every(value => typeof value === "string" && value.trim())
+    && event && isValidNativePrpEnvelope(event, row.protocolSchemaVersion)
+    && row.runId === binding.runId && row.companyId === binding.companyId && row.agentId === binding.agentId
+    && event.runId === binding.runId && event.eventType === row.eventType && event.sourceKind === "runner"
+    && typeof event.sourceEventId === "string" && event.sourceEventId.trim() && event.sourceEventId === row.sourceEventId
+    && typeof event.sourceInstanceId === "string" && event.sourceInstanceId.trim() && event.sourceInstanceId === row.sourceInstanceId
+    && (sourceInstanceId === undefined || event.sourceInstanceId === sourceInstanceId)
+    && Number.isSafeInteger(event.sourceSeq) && event.sourceSeq > 0 && event.sourceSeq === row.sourceSeq);
+}

tests/runner-e2e/product-journeys/agent-flows.ts

@@ -2,7 +2,8 @@ import { createHash } from "node:crypto";
 import { mkdir, writeFile } from "node:fs/promises";
 import path from "node:path";
 import { expect } from "@playwright/test";
-import { activeNativeTurnStarted, nativeCancellationHasNoLaterWork, nativeStopAcknowledged, nativeWorkEvents, stopAndRedirectThroughUi } from "../stop-redirect-flow.js";
+import { nativeCancellationHasNoLaterWork, nativeStopAcknowledged, nativeWorkEvents, stopAndRedirectThroughUi } from "../stop-redirect-flow.js";
+import { activeStopWorkBoundary, stoppedRunAccountingFinalized, STOP_REVISION_WAIT, type StopWorkBoundary } from "./stop-redirect-evidence.js";
 import { runnerTasks } from "../catalog.js";
 import { everydayTasks, SLUGIFY_REQUIREMENTS, SLUGIFY_REVISION } from "../everyday-cases.js";
 import { latestStoryDelivery, type StoryDelivery } from "../everyday-delivery.js";
@@ -214,25 +215,30 @@ export async function runAgentJourney(id: AgentJourney, context: JourneyContext,
     await runArtifactOracle(null, "base", options.deadlineAt);
     const issue = await create("", SLUGIFY_REQUIREMENTS);
     const initial = await download(issue, await settled(issue, 1), "checkpoint", "base");
-    await reply(issue, SLUGIFY_REVISION);
+    await reply(issue, `${SLUGIFY_REVISION}\n${STOP_REVISION_WAIT}`);
     const state = await waitFor(options.deadlineAt, () => read(issue), value => value.taskRuns.some(run => run.status === "running"), "stop_boundary_not_active");
     const original = state.taskRuns.find(run => run.status === "running")!;
     await open(issue);
     const originalEvents = () => collectRunEvents<Row>((afterSeq, limit) =>
       context.api.get(`/api/heartbeat-runs/${original.id}/events?afterSeq=${afterSeq}&limit=${limit}`));
-    const boundary = await waitFor(options.deadlineAt, async () => {
-      const events = await originalEvents();
-      await retain("stop-startup-events.json", events);
-      const current = await read(issue);
-      const run = current.taskRuns.find(row => row.id === original.id);
-      if (!run || !active(run)) throw new JourneyError("failed", "stop_boundary_already_settled");
-      return { state: current, run, events };
-    }, value => activeNativeTurnStarted(value.run, value.events), "stop_native_turn_not_active");
-    await retain("stop-before.json", boundary.state);
     cancelled.push(original.id);
-    const runActivity = async () => nativeWorkEvents(await originalEvents());
     let beforeEvents: Row[];
+    let boundary: StopWorkBoundary;
+    const runActivity = async () => nativeWorkEvents(await originalEvents(), boundary);
     const redirected = await stopAndRedirectThroughUi({ page: context.page, nonce,
+      beforeStop: async () => {
+        const observed = await waitFor(options.deadlineAt, async () => {
+          const events = await originalEvents();
+          await retain("stop-startup-events.json", events);
+          const current = await read(issue);
+          const run = current.taskRuns.find(row => row.id === original.id);
+          if (!run || !active(run)) throw new JourneyError("failed", "stop_boundary_already_settled");
+          return { state: current, boundary: activeStopWorkBoundary(run, events) };
+        }, value => Boolean(value.boundary), "stop_native_command_and_usage_not_observed");
+        boundary = observed.boundary!;
+        await retain("stop-boundary.json", boundary);
+        await retain("stop-before.json", observed.state);
+      },
       awaitRetirement: async () => {
         const stopped = await waitFor(options.deadlineAt, () => read(issue), value => {
           const run = value.taskRuns.find(row => row.id === original.id);
@@ -245,7 +251,7 @@ export async function runAgentJourney(id: AgentJourney, context: JourneyContext,
           return events;
         }, events => events.some(row => row.payload?.prpEvent?.eventType === "turn.cancelled"), "stop_native_terminal_not_retained");
         check("stop-native-terminal-retained", beforeEvents.some(row => row.payload?.prpEvent?.eventType === "turn.cancelled"));
-        if (!nativeCancellationHasNoLaterWork(beforeEvents)) check("stopped-run-no-later-events", false);
+        if (!nativeCancellationHasNoLaterWork(beforeEvents, boundary)) check("stopped-run-no-later-events", false);
       },
       beforeRedirect: async () => {
         const leases = await waitFor(options.deadlineAt, () => context.api.get<Row[]>(`/api/environments/${context.environmentId}/leases`), rows => {
@@ -253,6 +259,13 @@ export async function runAgentJourney(id: AgentJourney, context: JourneyContext,
           return owned.length > 0 && owned.every(row => ["released", "retained", "expired"].includes(row.status) && row.cleanupStatus === "success");
         }, "stopped_run_resources_not_released");
         await retain("stop-leases.json", leases.filter(row => row.issueId === issue.id && row.heartbeatRunId === original.id));
+        const accounted = await waitFor(options.deadlineAt, async () =>
+          (await read(issue)).taskRuns.find(row => row.id === original.id),
+        run => Boolean(run && stoppedRunAccountingFinalized(run, boundary)), "stopped_run_accounting_not_finalized");
+        await retain("stop-accounting.json", { runId: accounted!.id, costAccountingPending: accounted!.costAccountingPending,
+          costAccountedAt: accounted!.costAccountedAt, receiptId: accounted!.usageJson.accountingReceiptId,
+          sourceId: accounted!.usageJson.accountingReceiptSourceId, sequence: accounted!.usageJson.accountingReceiptSequence,
+          ready: accounted!.usageJson.accountingReceiptReady });
         check("active-work-stopped");
       },
       reply: message => reply(issue, message),
@@ -264,7 +277,7 @@ export async function runAgentJourney(id: AgentJourney, context: JourneyContext,
     check("new-direction-completed", completion.runId !== original.id);
     check("stopped-run-remained-terminal", final.taskRuns.find(run => run.id === original.id)?.status === "cancelled");
     const afterEvents = await runActivity();
-    check("stopped-run-no-later-events", nativeCancellationHasNoLaterWork(afterEvents)
+    check("stopped-run-no-later-events", nativeCancellationHasNoLaterWork(afterEvents, boundary!)
       && JSON.stringify(beforeEvents!) === JSON.stringify(afterEvents));
     check("stop-preserved-download", createHash("sha256").update(await context.api.bytes(`/api/attachments/${initial.attachmentId}/content`)).digest("hex") === initial.sha256);
     const queue = await context.api.get<Row>(`/api/issues/${issue.id}/queued-comments`);

tests/runner-e2e/product-journeys/launch.ts

@@ -108,7 +108,7 @@ export async function launchProductJourneys(args: string[], repositoryRoot: stri
   try { sha = execFileSync("git", ["rev-parse", "HEAD"], { cwd: repositoryRoot, encoding: "utf8", stdio: ["ignore", "pipe", "ignore"] }).trim(); } catch { /* Packaged harnesses may not include Git metadata. */ }
   const source = resolveRunnerE2ESource({ sha, ref: null, workflowRunUrl: null });
   const definitionHash = createHash("sha256").update(JSON.stringify(productJourneys));
-  for (const name of ["catalog.ts", "task-properties.ts", "task-search-filter.ts", "task-search-evidence.ts", "basic-flows.ts", "basic-evidence.ts", "task-document-flow.ts", "task-document-evidence.ts", "project-context-flow.ts", "project-context-readback.ts", "project-context-evidence.ts", "agent-flows.ts", "active-followup-flow.ts", "active-followup-evidence.ts", "clarification-flow.ts", "clarification-evidence.ts", "../question-resume-ui.ts", "../question-resume-scoring.ts", "../runtime-question-readiness.ts", "../chat-flow.ts", "../continuation-readiness.ts", "../continuation-cases.ts", "success-contract.ts", "product-screenshot.ts", "cloud-onboarding.ts", "onboarding-connect.ts", "onboarding-first-task.ts", "workspace-deferral.ts", "onboarding-pre-acceptance.ts", "artifact-evidence.ts", "fixtures.ts", "launch.ts", "types.ts", "collection.ts", "local-target.ts"]) definitionHash.update(await readFile(path.join(import.meta.dirname, name)));
+  for (const name of ["catalog.ts", "task-properties.ts", "task-search-filter.ts", "task-search-evidence.ts", "basic-flows.ts", "basic-evidence.ts", "task-document-flow.ts", "task-document-evidence.ts", "project-context-flow.ts", "project-context-readback.ts", "project-context-evidence.ts", "agent-flows.ts", "active-followup-flow.ts", "active-followup-evidence.ts", "stop-redirect-evidence.ts", "clarification-flow.ts", "clarification-evidence.ts", "../question-resume-ui.ts", "../question-resume-scoring.ts", "../runtime-question-readiness.ts", "../chat-flow.ts", "../continuation-readiness.ts", "../continuation-cases.ts", "success-contract.ts", "product-screenshot.ts", "cloud-onboarding.ts", "onboarding-connect.ts", "onboarding-first-task.ts", "workspace-deferral.ts", "onboarding-pre-acceptance.ts", "artifact-evidence.ts", "fixtures.ts", "launch.ts", "types.ts", "collection.ts", "local-target.ts"]) definitionHash.update(await readFile(path.join(import.meta.dirname, name)));
   for (const file of ["catalog.ts", "api.ts", "run-observations.ts", "native-task-reply.ts", "native-event-envelope.ts", "user-actions.ts", "plan-revision-flow.ts", "skill-studio-flow.ts", "protected-action-flow.ts", "story-artifact.ts", "stop-redirect-flow.ts", "everyday-cases.ts", "everyday-delivery.ts", "everyday-decisions.ts", "connection-guidance-evidence.ts", "everyday-artifact.py"]) definitionHash.update(await readFile(path.join(import.meta.dirname, "..", file)));
   definitionHash.update(runnerSuites.flatMap(suite => suite.tasks).find(task => task.id === "ask-question")!.buildPrompt("definition"));
   const suiteDefinitionHash = definitionHash.digest("hex");

tests/runner-e2e/product-journeys/stop-redirect-evidence.test.ts

@@ -0,0 +1,198 @@
+import { describe, expect, it } from "vitest";
+import { activeNativeTurnStarted, nativeCancellationHasNoLaterWork, nativeWorkEvents } from "../stop-redirect-flow.js";
+import { activeStopWorkBoundary, stoppedRunAccountingFinalized, STOP_REVISION_WAIT } from "./stop-redirect-evidence.js";
+import { agentJourneyRunCountValid } from "./success-contract.js";
+
+type Row = Record<string, any>;
+const run = { id: "run", companyId: "company", agentId: "agent", nativeIssueId: "issue", runtimeMode: "native", status: "running",
+  usageJson: { accountingReceiptSourceId: "receipt-source", accountingReceiptSequence: 5, accountingReceiptReady: false } };
+function event(seq: number, eventType: string, payload = {}, turnId = "turn"): Row {
+  return { seq, eventType, runId: run.id, companyId: run.companyId, agentId: run.agentId,
+    protocolSchemaVersion: 1, sourceEventId: `event-${seq}`, sourceInstanceId: "runner", sourceSeq: seq,
+    payload: { prpEvent: { schema: "paperclip.prp.event.v1", schemaVersion: 1, eventType, runId: run.id, turnId,
+      sourceKind: "runner", sourceEventId: `event-${seq}`, sourceInstanceId: "runner", sourceSeq: seq, payload: structuredClone(payload) } } };
+}
+const command = { schema: "paperclip.tool.execution.v1", executionId: "sleep", transport: "builtin", operation: "unknown",
+  name: "bash", namespace: null, readOnly: false, status: "running" };
+const usage = { kind: "usage", usage: { runDeltaComplete: true, runDelta: { inputTokens: 120, outputTokens: 10 } } };
+const events = () => [event(1, "turn.started"), event(2, "item.completed", usage), event(3, "tool.execution.started", command)];
+const before = () => activeStopWorkBoundary(run, events())!;
+const settled = () => ({ ...structuredClone(run), status: "cancelled", costAccountingPending: false,
+  finishedAt: "2026-01-01T00:00:01Z", costAccountedAt: "2026-01-01T00:00:02Z",
+  usageJson: { ...run.usageJson, accountingReceiptReady: true, accountingReceiptId: "receipt", accountingReceiptSequence: 6 } });
+
+describe("Stop work and redirect admission evidence", () => {
+  it("rejects the observed startup-only incident before and after cancellation", () => {
+    // Closed event metadata only. Synthetic identities replace the private run.
+    // The observed stream had93 contiguous rows and no usage or command event.
+    const rows: Row[] = Array.from({ length: 93 }, (_, i) => ({ seq: i + 1, eventType: "adapter.message" }));
+    rows[73] = event(74, "turn.started");
+    rows[77] = event(78, "turn.accepted");
+    rows[79] = event(80, "item.completed", { kind: "interrupt_acknowledgement" });
+    rows[81] = event(82, "turn.cancelled");
+    expect(activeNativeTurnStarted(run, rows.slice(0, 79))).toBe(true);
+    expect(activeStopWorkBoundary(run, rows.slice(0, 79))).toBeNull();
+    expect(activeStopWorkBoundary({ ...run, status: "cancelled" }, rows)).toBeNull();
+  });
+  it("requires the same current native turn's genuine usage and still-active command", () => {
+    expect(before()).toMatchObject({ runId: "run", executionId: "sleep", turnId: "turn", usageSeq: 2,
+      accountingSourceId: "receipt-source", accountingSequence: 5 });
+    const missing = [events().slice(0, 1), [event(1, "turn.started"), event(2, "tool.execution.started", command)],
+      [event(1, "turn.started"), event(2, "item.completed", usage)],
+      [...events(), event(4, "tool.execution.completed", { ...command, status: "completed" })],
+      [...events(), event(4, "turn.started", {}, "new-turn")], [...events(), event(4, "turn.cancelled")]];
+    for (const rows of missing) expect(activeStopWorkBoundary(run, rows)).toBeNull();
+  });
+  it("does not Stop a hold whose first usage report arrives only after the command ends", () => {
+    const rows = [event(1, "turn.started"), event(2, "tool.execution.started", command),
+      event(3, "tool.execution.progressed", command), event(4, "tool.execution.completed", { ...command, status: "completed" }),
+      event(5, "item.completed", { kind: "usage" }), event(6, "item.completed", usage), event(7, "turn.completed")];
+    for (let end = 1; end <= rows.length; end++) expect(activeStopWorkBoundary(run, rows.slice(0, end))).toBeNull();
+    expect(activeStopWorkBoundary({ ...run, status: "succeeded" }, rows)).toBeNull();
+  });
+  it("observes complete readiness usage and a separate still-active hold in the same turn", () => {
+    const ready = { ...command, executionId: "readiness" };
+    const rows = [event(1, "turn.started"), event(2, "tool.execution.started", ready),
+      event(3, "tool.execution.completed", { ...ready, status: "completed" }), event(4, "item.completed", { kind: "usage" }),
+      event(5, "item.completed", usage), event(6, "tool.execution.started", command)];
+    for (let end = 1; end < rows.length; end++) expect(activeStopWorkBoundary(run, rows.slice(0, end))).toBeNull();
+    expect(activeStopWorkBoundary(run, rows)).toMatchObject({ executionId: "sleep", startedSeq: 6, usageSeq: 5, turnId: "turn" });
+    expect(activeStopWorkBoundary(run, [...rows, event(7, "tool.execution.completed", { ...command, status: "completed" })])).toBeNull();
+  });
+  it.each(["foreign-company", "foreign-agent", "foreign-run", "foreign-turn", "foreign-runner", "bad-source", "bad-envelope",
+    "partial", "missing-count", "negative-count", "fractional-count", "placeholder-zero", "session-total", "semantic-tool", "read-only"])("rejects %s proof", kind => {
+    const rows = events(), row = rows[1]!, envelope = row.payload.prpEvent;
+    if (kind === "foreign-company") row.companyId = "foreign";
+    if (kind === "foreign-agent") row.agentId = "foreign";
+    if (kind === "foreign-run") { row.runId = "foreign"; envelope.runId = "foreign"; }
+    if (kind === "foreign-turn") envelope.turnId = "old-turn";
+    if (kind === "foreign-runner") { row.sourceInstanceId = "foreign"; envelope.sourceInstanceId = "foreign"; }
+    if (kind === "bad-source") row.sourceSeq++;
+    if (kind === "bad-envelope") row.protocolSchemaVersion = 2;
+    if (kind === "partial") envelope.payload.usage.runDeltaComplete = false;
+    if (kind === "missing-count") delete envelope.payload.usage.runDelta.outputTokens;
+    if (kind === "negative-count") envelope.payload.usage.runDelta.outputTokens = -1;
+    if (kind === "fractional-count") envelope.payload.usage.runDelta.outputTokens = 1.5;
+    if (kind === "placeholder-zero") envelope.payload.usage.runDelta = { inputTokens: 0, outputTokens: 0 };
+    if (kind === "session-total") envelope.payload.usage.usageBasis = "session_cumulative";
+    if (kind === "semantic-tool") rows[2]!.payload.prpEvent.payload = { ...command, name: "paperclip_finish" };
+    if (kind === "read-only") rows[2]!.payload.prpEvent.payload = { ...command, readOnly: true };
+    expect(activeStopWorkBoundary(run, rows)).toBeNull();
+  });
+  it("rejects stale usage from before the current turn, even with a live command", () => {
+    const rows = [event(1, "item.completed", usage, "old-turn"), event(2, "turn.started"), event(3, "tool.execution.started", command)];
+    expect(activeStopWorkBoundary(run, rows)).toBeNull();
+    expect(activeStopWorkBoundary(run, [event(1, "turn.started", {}, "old-turn"), event(2, "item.completed", usage, "old-turn"),
+      event(3, "turn.started"), event(4, "tool.execution.started", command, "old-turn")])).toBeNull();
+    expect(activeStopWorkBoundary(run, [...events(), event(4, "item.completed", { kind: "usage", usage: {
+      ...usage.usage, runDeltaComplete: false,
+    } })])).toBeNull();
+  });
+  it("requires the stopped run's settled API receipt without assuming controller or clock order", () => {
+    expect(stoppedRunAccountingFinalized(settled(), before())).toBe(true);
+    const reconciled = settled();
+    Object.assign(reconciled.usageJson, { accountingReceiptSourceId: "reconciled-source", accountingReceiptSequence: 5 });
+    reconciled.costAccountedAt = "2026-01-01T00:00:00Z";
+    expect(stoppedRunAccountingFinalized(reconciled, before())).toBe(true);
+    const changes: Array<(r: Row) => void> = [r => { r.id = "foreign"; }, r => { r.companyId = "foreign"; },
+      r => { r.agentId = "foreign"; }, r => { r.nativeIssueId = "foreign"; }, r => { r.runtimeMode = "legacy"; },
+      r => { r.status = "succeeded"; }, r => { r.costAccountingPending = true; }, r => { delete r.costAccountingPending; },
+      r => { r.costAccountedAt = null; }, r => { r.costAccountedAt = "invalid"; },
+      r => { r.usageJson.accountingReceiptReady = false; }, r => { delete r.usageJson.accountingReceiptId; },
+      r => { delete r.usageJson.accountingReceiptSourceId; }, r => { r.usageJson.accountingReceiptSequence = 0; }];
+    for (const change of changes) { const r = settled(); change(r); expect(stoppedRunAccountingFinalized(r, before())).toBe(false); }
+  });
+  it("accepts a previously ready receipt only when the exact stopped run is actually settled", () => {
+    const ready = { ...structuredClone(run), usageJson: { ...run.usageJson,
+      accountingReceiptReady: true, accountingReceiptId: "existing-receipt" } };
+    const boundary = activeStopWorkBoundary(ready, events())!;
+    expect(boundary).not.toBeNull();
+    const cancelled = { ...ready, status: "cancelled", costAccountingPending: false,
+      costAccountedAt: "2026-01-01T00:00:00Z", finishedAt: "2026-01-01T00:00:01Z" };
+    expect(stoppedRunAccountingFinalized(cancelled, boundary)).toBe(true);
+    for (const patch of [{ status: "running" }, { costAccountingPending: true },
+      { costAccountedAt: null }, { usageJson: { ...ready.usageJson, accountingReceiptReady: false } }]) {
+      expect(stoppedRunAccountingFinalized({ ...cancelled, ...patch }, boundary)).toBe(false);
+    }
+  });
+  it("permits canonical usage at either snapshot without changing the productive work history", () => {
+    const stopped = [...events(), event(4, "turn.cancelled")];
+    const accounted = [...stopped, event(5, "item.completed", usage)];
+    for (const rows of [stopped, accounted]) expect(nativeCancellationHasNoLaterWork(rows, before())).toBe(true);
+    expect(nativeWorkEvents(accounted, before())).toEqual(nativeWorkEvents(stopped, before()));
+    expect(nativeWorkEvents([...accounted, event(6, "item.completed", usage)], before())).toEqual(nativeWorkEvents(accounted, before()));
+  });
+  it.each(["raw", "partial", "complete"].flatMap(kind => [1, 2].map(version => ({ kind, version }))))("permits bound canonical $kind usage in schema$version", ({ kind, version }) => {
+    const stopped = [...events(), event(4, "turn.cancelled")];
+    const payload = kind === "raw" ? { kind: "usage", item: { type: "usage" }, text: "", channel: "usage" }
+      : { kind: "usage", usage: { ...usage.usage, runDeltaComplete: kind === "complete" } };
+    const row = event(5, "item.completed", payload);
+    row.protocolSchemaVersion = version;
+    Object.assign(row.payload.prpEvent, { schema: `paperclip.prp.event.v${version}`, schemaVersion: version });
+    expect(nativeCancellationHasNoLaterWork([...stopped, row], before())).toBe(true);
+    expect(nativeWorkEvents([...stopped, row], before())).toEqual(nativeWorkEvents(stopped, before()));
+  });
+  it.each(["raw", "partial", "complete"])("permits self-consistent %s billing from a new accounting source", kind => {
+    const stopped = [...events(), event(4, "turn.cancelled")];
+    const payload = kind === "raw" ? { kind: "usage", item: { type: "usage" } }
+      : { kind: "usage", usage: { ...usage.usage, runDeltaComplete: kind === "complete" } };
+    const row = event(5, "item.completed", payload);
+    row.sourceInstanceId = row.payload.prpEvent.sourceInstanceId = "reconciler";
+    expect(nativeCancellationHasNoLaterWork([...stopped, row], before())).toBe(true);
+    expect(nativeWorkEvents([...stopped, row], before())).toEqual(nativeWorkEvents(stopped, before()));
+  });
+  it.each(["row-company", "row-agent", "row-run", "envelope-run", "row-instance", "envelope-instance", "missing-instance", "empty-instance",
+    "source-event", "missing-source-event", "row-source-event", "source-sequence", "missing-sequence", "zero-sequence", "fractional-sequence", "source-kind"])("does not discard usage with an unbound %s", kind => {
+    const stopped = [...events(), event(4, "turn.cancelled")], row = event(5, "item.completed", usage), envelope = row.payload.prpEvent;
+    if (kind === "row-company") row.companyId = "foreign";
+    if (kind === "row-agent") row.agentId = "foreign";
+    if (kind === "row-run") row.runId = "foreign";
+    if (kind === "envelope-run") envelope.runId = "foreign";
+    if (kind === "row-instance") row.sourceInstanceId = "foreign";
+    if (kind === "envelope-instance") envelope.sourceInstanceId = "foreign";
+    if (kind === "missing-instance") delete envelope.sourceInstanceId;
+    if (kind === "empty-instance") row.sourceInstanceId = envelope.sourceInstanceId = "";
+    if (kind === "source-event") envelope.sourceEventId = "foreign";
+    if (kind === "missing-source-event") delete envelope.sourceEventId;
+    if (kind === "row-source-event") delete row.sourceEventId;
+    if (kind === "source-sequence") envelope.sourceSeq++;
+    if (kind === "missing-sequence") delete row.sourceSeq;
+    if (kind === "zero-sequence") row.sourceSeq = envelope.sourceSeq = 0;
+    if (kind === "fractional-sequence") row.sourceSeq = envelope.sourceSeq = 1.5;
+    if (kind === "source-kind") envelope.sourceKind = "server";
+    expect(nativeCancellationHasNoLaterWork([...stopped, row], before())).toBe(false);
+    expect(nativeWorkEvents([...stopped, row], before())).not.toEqual(nativeWorkEvents(stopped, before()));
+  });
+  it("keeps usage strict when authoritative current-run context is absent", () => {
+    const stopped = [...events(), event(4, "turn.cancelled")], accounted = [...stopped, event(5, "item.completed", usage)];
+    expect(nativeCancellationHasNoLaterWork(accounted)).toBe(false);
+    expect(nativeWorkEvents(accounted)).not.toEqual(nativeWorkEvents(stopped));
+  });
+  it.each(["tool.execution.started", "tool.execution.progressed", "turn.started", "message.delta", "item.delta", "item.completed"]
+    .flatMap(kind => ["runner", "reconciler"].map(source => ({ kind, source }))))("rejects productive $kind from $source interleaved with post-Stop accounting", ({ kind, source }) => {
+    const stopped = [...events(), event(4, "turn.cancelled")];
+    const later = [...stopped, event(5, "item.completed", usage), event(6, kind, { kind: "message", text: "continued work" }),
+      event(7, "item.completed", usage)];
+    later[later.length - 2]!.sourceInstanceId = later[later.length - 2]!.payload.prpEvent.sourceInstanceId = source;
+    expect(nativeCancellationHasNoLaterWork(later, before())).toBe(false);
+    expect(nativeWorkEvents(later, before())).not.toEqual(nativeWorkEvents(stopped, before()));
+  });
+  it("does not hide a malformed or differently typed item behind a usage label", () => {
+    const stopped = [...events(), event(4, "turn.cancelled")];
+    const malformed = event(5, "item.completed", usage); malformed.protocolSchemaVersion = 2;
+    const mismatched = event(5, "item.completed", usage); mismatched.eventType = "message.delta";
+    for (const row of [malformed, mismatched, event(5, "item.delta", usage), event(5, "message.delta", usage)]) {
+      expect(nativeCancellationHasNoLaterWork([...stopped, row], before())).toBe(false);
+      expect(nativeWorkEvents([...stopped, row], before())).not.toEqual(nativeWorkEvents(stopped, before()));
+    }
+  });
+  it("retains the bounded command and exact three-run contract", () => {
+    expect(STOP_REVISION_WAIT).toContain("printf 'Ready for revision\\n'");
+    expect(STOP_REVISION_WAIT).toContain("Wait for that command's result. Then, in a separate shell tool call");
+    expect(STOP_REVISION_WAIT.indexOf("printf")).toBeLessThan(STOP_REVISION_WAIT.indexOf("sleep 90"));
+    expect(STOP_REVISION_WAIT).toContain("sleep 90 exactly once");
+    expect(STOP_REVISION_WAIT).toContain("Before changing any file");
+    expect(agentJourneyRunCountValid("stop-redirect", 3)).toBe(true);
+    for (const count of [0, 1, 2, 4]) expect(agentJourneyRunCountValid("stop-redirect", count)).toBe(false);
+  });
+});

tests/runner-e2e/product-journeys/stop-redirect-evidence.ts

@@ -0,0 +1,63 @@
+import { createHash } from "node:crypto";
+import { isBoundNativePrpEventRow } from "../native-event-envelope.js";
+import { activeFollowupBoundary, type ActiveFollowupBoundary } from "./active-followup-evidence.js";
+
+type Row = Record<string, any>;
+export interface StopWorkBoundary extends ActiveFollowupBoundary {
+  companyId: string;
+  agentId: string;
+  issueId: string;
+  turnId: string;
+  usageSeq: number;
+  usageRowSha256: string;
+  accountingSourceId: string;
+  accountingSequence: number;
+}
+const text = (value: unknown): value is string => typeof value === "string" && Boolean(value.trim());
+const time = (value: unknown) => typeof value === "string" ? Date.parse(value) : NaN;
+
+/** A model turn starting does not prove work or a provider usage observation. */
+export function activeStopWorkBoundary(run: Row, events: Row[]): StopWorkBoundary | null {
+  const command = activeFollowupBoundary(run, events);
+  if (!command) return null;
+  const commandRow = events.find(row => row.seq === command.startedSeq)!;
+  const turnId = commandRow.payload.prpEvent.turnId;
+  const issueId = run.nativeIssueId ?? run.issueId ?? run.contextSnapshot?.issueId;
+  const accounting = run.usageJson;
+  if (!text(turnId) || !text(issueId) || !text(accounting?.accountingReceiptSourceId)
+    || !Number.isSafeInteger(accounting.accountingReceiptSequence) || accounting.accountingReceiptSequence <= 0) return null;
+  const binding = { runId: run.id, companyId: run.companyId, agentId: run.agentId };
+  const bound = (row: Row) => isBoundNativePrpEventRow(row, binding, command.sourceInstanceId);
+  const turnStart = events.findLast(row => bound(row) && row.eventType === "turn.started");
+  if (!turnStart || turnStart.payload.prpEvent.turnId !== turnId || turnStart.seq >= command.startedSeq) return null;
+  const usage = events.findLast(row => bound(row) && row.seq > turnStart.seq && row.eventType === "item.completed"
+    && row.payload.prpEvent.turnId === turnId && row.payload.prpEvent.payload?.kind === "usage");
+  const report = usage?.payload.prpEvent.payload.usage;
+  const delta = report?.runDelta;
+  if (!usage || report?.runDeltaComplete !== true || report.usageBasis === "session_cumulative"
+    || report.usageSource === "session_cumulative" || !delta || typeof delta !== "object" || Array.isArray(delta)
+    || ![delta.inputTokens, delta.outputTokens].every(value => Number.isSafeInteger(value) && value >= 0)
+    || !Object.entries(delta).every(([key, value]) => !key.endsWith("Tokens") || Number.isSafeInteger(value) && Number(value) >= 0)
+    || !Object.entries(delta).some(([key, value]) => key.endsWith("Tokens") && Number(value) > 0)) return null;
+  return { ...command, companyId: run.companyId, agentId: run.agentId, issueId, turnId,
+    usageSeq: usage.seq, usageRowSha256: createHash("sha256").update(JSON.stringify(usage)).digest("hex"),
+    accountingSourceId: accounting.accountingReceiptSourceId, accountingSequence: accounting.accountingReceiptSequence };
+}
+
+/**
+ * Require this stopped run's actual settled accounting state, not a new receipt.
+ * A receipt can already be ready before Stop, and reconciliation can change its
+ * source without advancing its sequence. Neither proves or disproves settlement.
+ */
+export function stoppedRunAccountingFinalized(run: Row, before: StopWorkBoundary): boolean {
+  const usage = run.usageJson;
+  return run.id === before.runId && run.companyId === before.companyId && run.agentId === before.agentId
+    && (run.nativeIssueId ?? run.issueId ?? run.contextSnapshot?.issueId) === before.issueId
+    && run.runtimeMode === "native" && run.status === "cancelled"
+    && run.costAccountingPending === false && Number.isFinite(time(run.costAccountedAt))
+    && usage?.accountingReceiptReady === true && text(usage.accountingReceiptId)
+    && text(usage.accountingReceiptSourceId)
+    && Number.isSafeInteger(usage.accountingReceiptSequence) && usage.accountingReceiptSequence > 0;
+}
+
+export const STOP_REVISION_WAIT = "Before changing any file, first run the harmless shell command printf 'Ready for revision\\n' once. Wait for that command's result. Then, in a separate shell tool call, run sleep 90 exactly once. Wait for the sleep command to finish before making the requested change. This gives me time to stop the revision while you work.";

tests/runner-e2e/stop-redirect-boundary.browser.mjs

@@ -0,0 +1,119 @@
+import assert from "node:assert/strict";
+import { chromium } from "@playwright/test";
+import { nativeCancellationHasNoLaterWork, nativeWorkEvents, stopAndRedirectThroughUi } from "./stop-redirect-flow.ts";
+import { activeStopWorkBoundary, stoppedRunAccountingFinalized } from "./product-journeys/stop-redirect-evidence.ts";
+
+const base = { id: "run", companyId: "company", agentId: "agent", nativeIssueId: "issue", runtimeMode: "native", status: "running",
+  usageJson: { accountingReceiptSourceId: "source", accountingReceiptSequence: 5 } };
+const event = (seq, type, payload = {}) => ({ seq, eventType: type, runId: base.id, companyId: base.companyId, agentId: base.agentId,
+  protocolSchemaVersion: 1, sourceEventId: `event-${seq}`, sourceInstanceId: "runner", sourceSeq: seq,
+  payload: { prpEvent: { schema: "paperclip.prp.event.v1", schemaVersion: 1, eventType: type, runId: base.id, turnId: "turn",
+    sourceKind: "runner", sourceEventId: `event-${seq}`, sourceInstanceId: "runner", sourceSeq: seq, payload } } });
+const command = { schema: "paperclip.tool.execution.v1", executionId: "sleep", transport: "process", operation: "execute", status: "running" };
+const usage = { kind: "usage", usage: { runDeltaComplete: true, runDelta: { inputTokens: 120, outputTokens: 10 } } };
+let state = { phase: "startup", run: structuredClone(base), events: [event(1, "turn.started")] };
+let beforeStop = false, beforeRedirect = false, stops = 0, replies = 0, cancelled = false, boundary, journey, beforeEvents;
+const observed = new Set();
+const browser = await chromium.launch({ headless: true });
+const page = await browser.newPage();
+const until = async predicate => {
+  const deadline = Date.now() + 3000;
+  while (!predicate()) {
+    assert(Date.now() < deadline, "bounded observer did not advance");
+    await new Promise(done => setTimeout(done, 10));
+  }
+};
+const observe = async predicate => {
+  const deadline = Date.now() + 3000;
+  while (!cancelled && Date.now() < deadline) {
+    const snapshot = await page.evaluate(async () => (await fetch("/api/owned-observation")).json());
+    observed.add(snapshot.phase);
+    const result = predicate(snapshot);
+    if (result) return result;
+    await new Promise(done => setTimeout(done, 10));
+  }
+  throw new Error("bounded controlled observation did not qualify");
+};
+try {
+  await page.route("https://stop.fixture/**", route => route.request().url().endsWith("/api/owned-observation")
+    ? route.fulfill({ contentType: "application/json", body: JSON.stringify(state) })
+    : route.fulfill({ contentType: "text/html", body:
+      '<button data-testid="task-chat-composer-stop" onclick="window.stopped()">Stop</button>' }));
+  await page.exposeFunction("stopped", () => {
+    stops++;
+    state = { ...state, phase: "pending", events: [...state.events, event(7, "turn.cancelled")],
+      run: { ...base, status: "cancelled", costAccountingPending: true,
+      finishedAt: "2026-01-01T00:00:01Z", usageJson: { ...base.usageJson, accountingReceiptSequence: 6, accountingReceiptReady: false, accountingReceiptId: "receipt" } } };
+  });
+  await page.goto("https://stop.fixture/task");
+  journey = stopAndRedirectThroughUi({ page, nonce: "fixture",
+    beforeStop: async () => { beforeStop = true; boundary = await observe(s => activeStopWorkBoundary(s.run, s.events)); },
+    awaitRetirement: async () => {
+      assert.equal(stops, 1);
+      beforeEvents = await observe(s => nativeCancellationHasNoLaterWork(s.events, boundary) && nativeWorkEvents(s.events, boundary));
+    },
+    beforeRedirect: async () => {
+      beforeRedirect = true;
+      const final = await observe(s => stoppedRunAccountingFinalized(s.run, boundary) && s);
+      assert(nativeCancellationHasNoLaterWork(final.events, boundary));
+      assert.deepEqual(nativeWorkEvents(final.events, boundary), beforeEvents);
+    },
+    reply: async () => { replies++; return { id: "owned-comment" }; },
+  });
+  journey.catch(() => undefined);
+  await until(() => observed.has("startup") || stops > 0);
+  assert.equal(stops, 0, "startup or incomplete usage must not click Stop");
+  assert.equal(beforeStop, true);
+  state = { ...state, phase: "command-only", events: [event(1, "turn.started"), event(2, "tool.execution.started", command)] };
+  await until(() => observed.has("command-only")); assert.equal(stops, 0);
+  state = { ...state, phase: "usage-only", events: [event(1, "turn.started"), event(2, "item.completed", usage)] };
+  await until(() => observed.has("usage-only")); assert.equal(stops, 0);
+  const partial = { kind: "usage", usage: { ...usage.usage, runDeltaComplete: false } };
+  state = { ...state, phase: "partial-usage", events: [event(1, "turn.started"), event(2, "item.completed", partial), event(3, "tool.execution.started", command)] };
+  await until(() => observed.has("partial-usage")); assert.equal(stops, 0);
+  state = { ...state, phase: "hold-ended-before-usage", events: [event(1, "turn.started"), event(2, "tool.execution.started", command),
+    event(3, "tool.execution.completed", { ...command, status: "completed" }), event(4, "item.completed", usage)] };
+  await until(() => observed.has("hold-ended-before-usage")); assert.equal(stops, 0);
+  const readiness = { ...command, executionId: "readiness" };
+  state = { ...state, phase: "readiness-active", events: [event(1, "turn.started"), event(2, "tool.execution.started", readiness)] };
+  await until(() => observed.has("readiness-active")); assert.equal(stops, 0);
+  const readyUsage = [event(1, "turn.started"), event(2, "tool.execution.started", readiness),
+    event(3, "tool.execution.completed", { ...readiness, status: "completed" }), event(4, "item.completed", { kind: "usage" }),
+    event(5, "item.completed", usage)];
+  state = { ...state, phase: "readiness-complete", events: readyUsage };
+  await until(() => observed.has("readiness-complete")); assert.equal(stops, 0);
+  state = { ...state, phase: "ready", events: [...readyUsage, event(6, "tool.execution.started", command)] };
+  await until(() => beforeRedirect && observed.has("pending"));
+  assert.equal(stops, 1); assert.equal(replies, 0, "Stop/lease proof cannot substitute for settled accounting");
+  const raw = { kind: "usage", item: { type: "usage" }, text: "", channel: "usage" };
+  const reconciled = event(8, "item.completed", raw);
+  reconciled.sourceInstanceId = reconciled.payload.prpEvent.sourceInstanceId = "reconciler";
+  state = { ...state, phase: "ready-but-pending", events: [...state.events, reconciled,
+    event(9, "item.completed", partial), event(10, "item.completed", usage)],
+    run: { ...state.run, usageJson: { ...state.run.usageJson, accountingReceiptReady: true } } };
+  await until(() => observed.has("ready-but-pending")); assert.equal(replies, 0);
+  const current = await page.evaluate(async () => (await fetch("/api/owned-observation")).json());
+  for (const mutate of [row => { row.companyId = "foreign"; }, row => { row.agentId = "foreign"; },
+    row => { row.payload.prpEvent.runId = "foreign"; }, row => { row.sourceEventId = "foreign"; },
+    row => { row.sourceInstanceId = "foreign"; }, row => { row.sourceSeq++; }]) {
+    const foreign = event(11, "item.completed", raw); mutate(foreign);
+    assert.equal(nativeCancellationHasNoLaterWork([...current.events, foreign], boundary), false);
+    assert.notDeepEqual(nativeWorkEvents([...current.events, foreign], boundary), beforeEvents);
+  }
+  for (const type of ["tool.execution.started", "message.delta", "item.delta"]) {
+    const productive = event(11, type, { kind: "message" });
+    productive.sourceInstanceId = productive.payload.prpEvent.sourceInstanceId = "reconciler";
+    assert.equal(nativeCancellationHasNoLaterWork([...current.events, productive], boundary), false);
+  }
+  assert.equal(nativeCancellationHasNoLaterWork(current.events), false);
+  state = { ...state, phase: "missing-accounted-time", run: { ...state.run, costAccountingPending: false } };
+  await until(() => observed.has("missing-accounted-time")); assert.equal(replies, 0);
+  state = { ...state, phase: "settled", run: { ...state.run, costAccountedAt: "2026-01-01T00:00:02Z" } };
+  assert.equal((await journey).reply.id, "owned-comment");
+  assert.equal(stops, 1); assert.equal(replies, 1);
+  console.log("PASS: Chromium/controlled API rejects startup, command-only, usage-only, partial pre-Stop usage and a hold completed before its first usage; completed readiness usage plus a separate active hold permits one Stop; settled accounting permits one redirect; bound usage preserves work history and wrong bindings/productive output fail.");
+} finally {
+  cancelled = true;
+  await journey?.catch(() => undefined);
+  await browser.close();
+}

tests/runner-e2e/stop-redirect-flow.ts

@@ -1,17 +1,21 @@
 import type { Page } from "@playwright/test";
+import { isBoundNativePrpEventRow, type NativePrpRunBinding } from "./native-event-envelope.js";
 type Row = Record<string, any>;
 
 /** Work emitted by a retired turn, excluding accounting and control-plane bookkeeping. */
-export function nativeWorkEvents<T extends Row>(events: T[]): T[] {
+export function nativeWorkEvents<T extends Row>(events: T[], binding?: NativePrpRunBinding): T[] {
   return events.filter(row => {
-    const kind = row.payload?.prpEvent?.eventType;
+    const event = row.payload?.prpEvent;
+    const kind = event?.eventType;
+    if (kind === "item.completed" && row.eventType === kind
+      && isBoundNativePrpEventRow(row, binding) && event.payload?.kind === "usage") return false;
     return typeof kind === "string" && /^(tool\.execution\.|turn\.|message\.|item\.)/.test(kind);
   });
 }
 
 /** Detect post-cancellation output even if it arrived before the first snapshot. */
-export function nativeCancellationHasNoLaterWork(events: Row[]): boolean {
-  const work = nativeWorkEvents(events);
+export function nativeCancellationHasNoLaterWork(events: Row[], binding?: NativePrpRunBinding): boolean {
+  const work = nativeWorkEvents(events, binding);
   const cancellation = work.findIndex(row => row.payload.prpEvent.eventType === "turn.cancelled");
   return cancellation >= 0 && cancellation === work.length - 1;
 }
@@ -26,10 +30,12 @@ export function activeNativeTurnStarted(run: Row | undefined, events: Row[]): bo
 /** Shared story order; target adapters supply authoritative retirement observations. */
 export async function stopAndRedirectThroughUi(input: {
   page: Page; nonce: string;
+  beforeStop?(): Promise<void>;
   awaitRetirement(): Promise<void>;
   beforeRedirect(): Promise<void>;
   reply(message: string): Promise<unknown>;
 }) {
+  await input.beforeStop?.();
   await input.page.getByTestId("task-chat-composer-stop").last().click();
   await input.awaitRetirement();
   await input.beforeRedirect();