From d2fecdf711c5f88f85e910da00fb656ccd0c66e4 Mon Sep 17 00:00:00 2001
From: Harry Yep <git@hode.co.uk>
Date: Mon, 27 Jul 2026 13:05:04 +0800
Subject: [PATCH 1/7] fix: codex companion cwd warning and oversized event skip

---
 docs/codex-contract.md                    |  4 +--
 docs/companion-contract.md                |  2 +-
 plugins/codex/scripts/codex-companion.mjs | 22 ++++++++++--
 plugins/codex/scripts/lib/codex-exec.mjs  | 35 ++++++++++++++++---
 tests/codex-companion.test.mjs            | 41 +++++++++++++++++++++++
 tests/codex-exec.test.mjs                 | 18 +++++++++-
 tests/fake-codex                          |  4 +++
 7 files changed, 115 insertions(+), 11 deletions(-)

diff --git a/docs/codex-contract.md b/docs/codex-contract.md
index 6b4c4d4..e90f111 100644
--- a/docs/codex-contract.md
+++ b/docs/codex-contract.md
@@ -61,7 +61,7 @@ The companion does not add a shell command allow list and does not rewrite the u
 
 Model and reasoning effort remain unset unless explicitly requested. The Codex configuration remains authoritative for defaults. The companion explicitly sets web search to disabled and workspace-write network access to false by default. `--web` opts a read-only or write task into live Codex web search. `--network` requires `--write` and opts the workspace-write sandbox into network access. Task mode `--output-schema <path>` validates a regular JSON Schema file up to 256 KiB and parses the final agent message once, while the Codex CLI `review` subcommand accepts but silently ignores `--output-schema`, unchanged from 0.144.6 through the live-verified 0.145.0.
 
-The companion writes the final prompt to Codex stdin. Prompts are never interpolated into a shell command. Programmatic callers use `--request-stdin` and pipe the complete raw companion request through stdin, leaving fixed argv free of request bytes. Claude Code's Bash tool does not expose stdin, so slash commands and wrapper Agents retain a private token-named staging compatibility path: the command model writes the raw request through its Write tool before invoking a fixed companion command containing only that token. Once the staging write has landed, the parser preserves positional whitespace, newlines, quotes, and backslashes byte for byte while removing recognized adapter options. The staged compatibility path is not an end-to-end byte-exact transport because a model performs the Write step. SessionEnd removes verified staging files owned by that Claude session, and later transport creation prunes other verified staging files after one hour. Final response text comes from the bounded JSONL stream, so Codex never receives an unbounded fallback output path. Prompts, individual JSONL events, final responses, raw event ledgers, logs, and rendered diagnostics have explicit byte limits. Oversized input or output fails as `resource` before it can inflate durable state. Resume accepts only a persisted Codex thread ID or a companion-selected terminal task record. `--resume-last` is restricted to the current Claude session when its id is available and otherwise selects the newest eligible workspace task. Active jobs are never resume candidates. `--fresh` rejects combination with either resume form and forces a new thread.
+The companion writes the final prompt to Codex stdin. Prompts are never interpolated into a shell command. Programmatic callers use `--request-stdin` and pipe the complete raw companion request through stdin, leaving fixed argv free of request bytes. Claude Code's Bash tool does not expose stdin, so slash commands and wrapper Agents retain a private token-named staging compatibility path: the command model writes the raw request through its Write tool before invoking a fixed companion command containing only that token. Once the staging write has landed, the parser preserves positional whitespace, newlines, quotes, and backslashes byte for byte while removing recognized adapter options. The staged compatibility path is not an end-to-end byte-exact transport because a model performs the Write step. SessionEnd removes verified staging files owned by that Claude session, and later transport creation prunes other verified staging files after one hour. Final response text comes from the bounded JSONL stream, so Codex never receives an unbounded fallback output path. Prompts, individual JSONL events, final responses, raw event ledgers, logs, and rendered diagnostics have explicit byte limits. An oversized single JSONL event is skipped with a diagnostic and the run continues; the run fails as `resource` when the protocol cannot complete after a skip, and oversized prompts or final responses still fail as `resource` before they can inflate durable state. Resume accepts only a persisted Codex thread ID or a companion-selected terminal task record. `--resume-last` is restricted to the current Claude session when its id is available and otherwise selects the newest eligible workspace task. Active jobs are never resume candidates. `--fresh` rejects combination with either resume form and forces a new thread.
 
 ## Review execution
 
@@ -115,7 +115,7 @@ Protocol fixtures cover every supported event type, unknown extensions, malforme
 
 ## Fusion integration
 
-Codex records are the primary source for transport status, delivery, semantic status, model request, resolved model and effort, token usage, thread identity, Claude session identity, repository identity, and job duration. Without a state override, Fusion reads only the canonical `codex-claude-code-fusion/state` root. The old `codex-openai-codex/state` root is included only through `/fusion:stats --include-legacy` or `FUSION_CODEX_INCLUDE_LEGACY=1`, is always read only, and is never copied or mutated. The new `/codex:result` command cannot collect an old plugin deliverable. `CODEX_COMPANION_DATA` selects only its own state root, just as the Fusion state overrides select only one root. Fusion observes those records without taking ownership of their lifecycle. A `done` Codex job remains `unverified` until the orchestrator checks its completion criteria or verification command and records `accepted` or `rejected`. `/fusion:stats --record-acceptance` writes that verdict through the companion's `record-acceptance` subcommand. Marking a job accepted after transport failure requires the explicit `--accept-failed-transport` override, and the stats report exposes acceptance anomalies.
+Codex records are the primary source for transport status, delivery, semantic status, model request, resolved model and effort, token usage, thread identity, Claude session identity, repository identity, and job duration. Without a state override, Fusion reads only the canonical `codex-claude-code-fusion/state` root. The old `codex-openai-codex/state` root is included only through `/fusion:stats --include-legacy` or `FUSION_CODEX_INCLUDE_LEGACY=1`, is always read only, and is never copied or mutated. The new `/codex:result` command cannot collect an old plugin deliverable. `CODEX_COMPANION_DATA` selects only its own state root, just as the Fusion state overrides select only one root. Fusion observes those records without taking ownership of their lifecycle. A `done` Codex job remains `unverified` until the orchestrator checks its completion criteria or verification command and records `accepted` or `rejected`. `/fusion:stats --record` writes the verdict into the Codex job record through the companion's `record-acceptance` subcommand. Marking a job accepted after transport failure requires the explicit `--accept-failed-transport` override, and the stats report exposes acceptance anomalies.
 
 `record-acceptance` requires `--reason` for a rejected verdict and accepts `--failure-kind` with one of `intent_override`, `scope_rewrite`, `wrong_approach`, or `style_mismatch` to classify judgment shaped rejections; the kind lands in `semanticFailureKind`.
 
diff --git a/docs/companion-contract.md b/docs/companion-contract.md
index 38cccce..c200a7a 100644
--- a/docs/companion-contract.md
+++ b/docs/companion-contract.md
@@ -44,7 +44,7 @@ The shared failure kinds are:
 - `cancelled`: The plugin cancel command, companion signal forwarding, or session cleanup intentionally marked the job cancelled.
 - `input`: The request was rejected before job creation, for example because the working directory was invalid.
 - `died`: The driving process exited without recording an outcome, or no driving pid or engine child pid was recorded before the pidless launcher grace window elapsed.
-- `resource`: The adapter rejected an artifact that exceeded a bounded resource limit or could not acquire an exclusive runtime resource, including because of an active session lease or an ambiguous legacy id.
+- `resource`: The adapter rejected an artifact that exceeded a bounded resource limit, a run whose protocol was left incomplete by skipped oversized events, or could not acquire an exclusive runtime resource, including because of an active session lease or an ambiguous legacy id.
 
 Engine adapters may define additional typed failure kinds beyond this shared set. The current Codex instance adds `protocol`, for a required structured lifecycle stream that was malformed, incomplete, or incompatible with the adapter, `process`, for an engine process that exited because of an unexpected signal or other unrequested process termination, and `policy`, for a forbidden Codex collaboration tool call. The Grok instance adds `sandbox`, for sandbox initialization or enforcement failure, `transport`, for prompt delivery or structured envelope failure, and `policy`, for missing positive tool-policy evidence or a fallback, unmappable, or unmatched tool-policy warning.
 
diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs
index 7f1350f..bb04318 100644
--- a/plugins/codex/scripts/codex-companion.mjs
+++ b/plugins/codex/scripts/codex-companion.mjs
@@ -99,6 +99,7 @@ const RECORD_ACCEPTANCE_VALUES = new Set(["accepted", "rejected", "unverified"])
 const RECORD_ACCEPTANCE_SOURCES = new Set(["collector", "main-loop", "stats"]);
 const SEMANTIC_FAILURE_KINDS = new Set(["intent_override", "scope_rewrite", "wrong_approach", "style_mismatch"]);
 const SOL_FOREGROUND_WRITE_WARNING = "warning: sol p90 wall clock exceeds the 600s foreground cap. Split the brief or name gpt-5.6-terra.";
+const IMPLICIT_SUBDIRECTORY_CWD_WARNING = (cwd, workspaceRoot) => `Codex cwd ${cwd} sits below repository top level ${workspaceRoot} without an explicit --cwd; the sandbox roots at this subdirectory. Pass --cwd to pin the intended workspace.`;
 let activeCommandArgv = null;
 
 class CompanionError extends Error {
@@ -1208,7 +1209,7 @@ function inheritedRouting(record, options, defaultServiceTier) {
   return { effort, inherited, model, serviceTier };
 }
 
-function createReservedJob({ background, brief, codexVersion, cwd, dataDir, jobClass, mode, request }) {
+function createReservedJob({ background, brief, codexVersion, cwd, dataDir, explicitCwd, jobClass, mode, request }) {
   brief = boundedPrompt(brief);
   const ownerIdentity = background ? null : currentProcessIdentity();
   const reserveWorkspace = (reservedRequest) => withWorkspaceLock(dataDir, cwd, () => {
@@ -1239,7 +1240,7 @@ function createReservedJob({ background, brief, codexVersion, cwd, dataDir, jobC
       pid: background ? null : process.pid,
       pidIdentity: ownerIdentity,
       pidOwnsProcessGroup: false,
-      request: reservedRequest,
+      request: explicitCwd ? { ...reservedRequest, explicitCwd: true } : reservedRequest,
       workspaceRoot: canonicalWorkspaceRoot(cwd)
     });
     const file = jobFilePath(dataDir, cwd, id);
@@ -1342,6 +1343,13 @@ function structuredOutputOutcome(request, outcome) {
   }
 }
 
+function implicitSubdirectoryCwdWarning(record) {
+  if (record.request?.explicitCwd === true || record.repositoryTopLevel == null || record.workspaceRoot === record.cwd) {
+    return null;
+  }
+  return IMPLICIT_SUBDIRECTORY_CWD_WARNING(record.cwd, record.workspaceRoot);
+}
+
 function finishExecutionFailure(file, record, error, env = process.env) {
   const rawMessage = oneLine(error?.message ?? error, "Codex companion execution failed.");
   appendJobLog(record.logFile, rawMessage);
@@ -1463,6 +1471,10 @@ async function executeRecord(found) {
     if (isSolForegroundWriteTask(completedRecord)) {
       diagnostics.push({ type: "warning", message: SOL_FOREGROUND_WRITE_WARNING });
     }
+    const cwdWarning = implicitSubdirectoryCwdWarning(completedRecord);
+    if (cwdWarning) {
+      diagnostics.push({ type: "warning", message: cwdWarning });
+    }
     return finishJob(found.file, {
       cumulativeTokenUsage: outcome.cumulativeTokenUsage,
       diagnostics,
@@ -1685,6 +1697,10 @@ async function dispatchJob(found, asJson) {
   if (isSolForegroundWriteTask(completed)) {
     process.stderr.write(`${SOL_FOREGROUND_WRITE_WARNING}\n`);
   }
+  const cwdWarning = implicitSubdirectoryCwdWarning(completed);
+  if (cwdWarning) {
+    process.stderr.write(`${cwdWarning}\n`);
+  }
   renderRecord(completed, asJson);
   collectRenderedJob(found.file, completed);
   if (completed.status !== "done") {
@@ -1739,6 +1755,7 @@ async function handleTask(rawArgv, transport = {}) {
     codexVersion: probe.version,
     cwd,
     dataDir,
+    explicitCwd: options.cwd != null,
     jobClass: "task",
     mode: write ? "write" : "consult",
     request
@@ -1830,6 +1847,7 @@ async function handleReview(rawArgv, adversarial = false, transport = {}) {
     codexVersion: probe.version,
     cwd,
     dataDir,
+    explicitCwd: options.cwd != null,
     jobClass: adversarial ? "adversarial-review" : "review",
     mode: "consult",
     request
diff --git a/plugins/codex/scripts/lib/codex-exec.mjs b/plugins/codex/scripts/lib/codex-exec.mjs
index 97e8ecc..9b13b59 100644
--- a/plugins/codex/scripts/lib/codex-exec.mjs
+++ b/plugins/codex/scripts/lib/codex-exec.mjs
@@ -966,6 +966,10 @@ function classifyDiagnostic(state, event) {
   if (!diagnostic || typeof diagnostic.message !== "string") {
     return;
   }
+  recordDiagnostic(state, diagnostic);
+}
+
+function recordDiagnostic(state, diagnostic) {
   state.diagnostics.push({ ...diagnostic, message: safeDiagnosticMessage(diagnostic.message) });
   if (state.diagnostics.length > MAX_DIAGNOSTICS) {
     state.diagnostics.splice(0, state.diagnostics.length - MAX_DIAGNOSTICS);
@@ -1240,6 +1244,7 @@ export async function runCodex(options = {}) {
     eventsTruncated: false,
     expectedResumeThreadId: typeof options.resumeThreadId === "string" && options.resumeThreadId.trim() ? options.resumeThreadId.trim() : null,
     finalResponse: "",
+    oversizedEventsSkipped: 0,
     protocolError: null,
     resourceError: null,
     resumeThreadMismatch: false,
@@ -1471,6 +1476,7 @@ export async function runCodex(options = {}) {
     await spawnCheckpointPromise;
     let segments = [];
     let segmentBytes = 0;
+    let skippingOversizedEvent = false;
     try {
       for await (const rawChunk of child.stdout) {
         const chunk = Buffer.from(rawChunk);
@@ -1479,11 +1485,27 @@ export async function runCodex(options = {}) {
           const newline = chunk.indexOf(10, offset);
           const end = newline === -1 ? chunk.length : newline;
           const segment = chunk.subarray(offset, end);
+          if (skippingOversizedEvent) {
+            if (newline === -1) {
+              break;
+            }
+            skippingOversizedEvent = false;
+            offset = newline + 1;
+            continue;
+          }
           if (segmentBytes + segment.length > eventLineMaxBytes) {
-            recordResourceError(state, `Codex emitted a JSONL event larger than the ${eventLineMaxBytes} byte limit.`);
-            void requestTermination();
-            child.stdout.destroy();
-            return;
+            state.oversizedEventsSkipped += 1;
+            if (state.oversizedEventsSkipped === 1) {
+              recordDiagnostic(state, { type: "warning", message: `Skipped a JSONL event larger than the ${eventLineMaxBytes} byte limit.` });
+            }
+            segments = [];
+            segmentBytes = 0;
+            if (newline === -1) {
+              skippingOversizedEvent = true;
+              break;
+            }
+            offset = newline + 1;
+            continue;
           }
           if (segment.length > 0) {
             segments.push(segment);
@@ -1529,7 +1551,10 @@ export async function runCodex(options = {}) {
     fs.closeSync(eventsFd);
   }
   if (!state.timedOut && !state.cancelled && !spawnError && !state.callbackError && processOutcome.exitCode === 0 && !processOutcome.signal) {
-    if (!state.threadStarted) {
+    const protocolIncomplete = !state.threadStarted || !state.turnStarted || !state.terminalEvent;
+    if (protocolIncomplete && state.oversizedEventsSkipped > 0) {
+      recordResourceError(state, `Codex emitted a JSONL event larger than the ${eventLineMaxBytes} byte limit.`);
+    } else if (!state.threadStarted) {
       recordProtocolError(state, "Codex exited without thread.started.");
     } else if (!state.turnStarted) {
       recordProtocolError(state, "Codex exited without turn.started.");
diff --git a/tests/codex-companion.test.mjs b/tests/codex-companion.test.mjs
index dcaf029..a4c1508 100644
--- a/tests/codex-companion.test.mjs
+++ b/tests/codex-companion.test.mjs
@@ -72,6 +72,47 @@ test("task header parsing captures model and effort only from a pipe-separated h
   assert.deepEqual(parseTaskHeader("lane: codex | model: gpt-5.6-terra"), { headerModel: "gpt-5.6-terra", headerEffort: null });
 });
 
+test("task warns when an implicit cwd is below a repository top level", (t) => {
+  const sandbox = makeSandbox(t);
+  const subdirectory = path.join(sandbox.workDir, "apps", "api");
+  fs.mkdirSync(subdirectory, { recursive: true });
+  execFileSync("git", ["init", "--quiet"], { cwd: sandbox.workDir });
+
+  const cwd = fs.realpathSync(subdirectory);
+  const workspaceRoot = fs.realpathSync(sandbox.workDir);
+  const warning = `Codex cwd ${cwd} sits below repository top level ${workspaceRoot} without an explicit --cwd; the sandbox roots at this subdirectory. Pass --cwd to pin the intended workspace.`;
+  const result = runCompanion(["task", "inspect the API"], { cwd: subdirectory, env: envFor(sandbox) });
+
+  assert.equal(result.status, 0, result.stderr);
+  assert.equal(result.stderr, `${warning}\n`);
+  assert.equal(jobRecords(sandbox)[0].diagnostics.filter((diagnostic) => diagnostic.type === "warning" && diagnostic.message === warning).length, 1);
+});
+
+test("task does not warn for an explicit cwd below a repository top level", (t) => {
+  const sandbox = makeSandbox(t);
+  const subdirectory = path.join(sandbox.workDir, "apps", "api");
+  fs.mkdirSync(subdirectory, { recursive: true });
+  execFileSync("git", ["init", "--quiet"], { cwd: sandbox.workDir });
+
+  const result = runCompanion(["task", "--cwd", ".", "inspect the API"], { cwd: subdirectory, env: envFor(sandbox) });
+
+  assert.equal(result.status, 0, result.stderr);
+  assert.equal(result.stderr, "");
+  assert.equal(jobRecords(sandbox)[0].diagnostics.some((diagnostic) => diagnostic.type === "warning"), false);
+});
+
+test("task does not warn for an implicit cwd outside a Git repository", (t) => {
+  const sandbox = makeSandbox(t);
+  const subdirectory = path.join(sandbox.workDir, "apps", "api");
+  fs.mkdirSync(subdirectory, { recursive: true });
+
+  const result = runCompanion(["task", "inspect the API"], { cwd: subdirectory, env: envFor(sandbox) });
+
+  assert.equal(result.status, 0, result.stderr);
+  assert.equal(result.stderr, "");
+  assert.equal(jobRecords(sandbox)[0].diagnostics.some((diagnostic) => diagnostic.type === "warning"), false);
+});
+
 test("job records capture the repository top level without requiring a Git repository", (t) => {
   const sandbox = makeSandbox(t);
   const repository = path.join(sandbox.root, "repository");
diff --git a/tests/codex-exec.test.mjs b/tests/codex-exec.test.mjs
index 42de1d5..5a6c9d6 100644
--- a/tests/codex-exec.test.mjs
+++ b/tests/codex-exec.test.mjs
@@ -525,7 +525,7 @@ test("a closed stdin fails even when Codex emits a completed lifecycle", async (
   assert.match(outcome.errorMessage, /closed stdin before accepting the complete prompt/i);
 });
 
-test("prompt, JSONL event, and final response limits fail with a resource error", async (t) => {
+test("prompt, incomplete oversized JSONL events, and final responses fail with a resource error", async (t) => {
   await assert.rejects(
     runFixture(t, "completed", { prompt: "x".repeat(65), promptMaxBytes: 64 }),
     (error) => error?.failureKind === "resource"
@@ -536,6 +536,10 @@ test("prompt, JSONL event, and final response limits fail with a resource error"
   });
   assert.equal(oversizedEvent.outcome.status, "error");
   assert.equal(oversizedEvent.outcome.failureKind, "resource");
+  assert.equal(oversizedEvent.outcome.errorMessage, "Codex emitted a JSONL event larger than the 128 byte limit.");
+  assert.deepEqual(oversizedEvent.outcome.diagnostics, [
+    { type: "warning", message: "Skipped a JSONL event larger than the 128 byte limit." }
+  ]);
   const oversizedResult = await runFixture(t, "completed", {
     resultMaxBytes: 64,
     env: { FAKE_CODEX_MESSAGE_BYTES: "1024" }
@@ -544,6 +548,18 @@ test("prompt, JSONL event, and final response limits fail with a resource error"
   assert.equal(oversizedResult.outcome.failureKind, "resource");
 });
 
+test("oversized JSONL events are skipped when the lifecycle completes", async (t) => {
+  const { outcome } = await runFixture(t, "oversized-midrun-event", {
+    eventLineMaxBytes: 128,
+    env: { FAKE_CODEX_EVENT_BYTES: "1024" }
+  });
+  assert.equal(outcome.status, "done");
+  assert.equal(outcome.failureKind, null);
+  assert.deepEqual(outcome.diagnostics, [
+    { type: "warning", message: "Skipped a JSONL event larger than the 128 byte limit." }
+  ]);
+});
+
 test("final responses come only from the bounded JSONL stream", async (t) => {
   const { args, files, outcome } = await runFixture(t, "fallback");
   assert.equal(outcome.status, "done");
diff --git a/tests/fake-codex b/tests/fake-codex
index a6594d3..54b3524 100755
--- a/tests/fake-codex
+++ b/tests/fake-codex
@@ -199,6 +199,10 @@ switch (mode) {
     complete();
     break;
   case "oversized-event":
+    start();
+    emit({ type: "future.event", payload: "x".repeat(Number(process.env.FAKE_CODEX_EVENT_BYTES || 4096)) });
+    break;
+  case "oversized-midrun-event":
     start();
     emit({ type: "future.event", payload: "x".repeat(Number(process.env.FAKE_CODEX_EVENT_BYTES || 4096)) });
     complete();

From aa2ef4f17f3d5caf20820c73e6abd139e02624fb Mon Sep 17 00:00:00 2001
From: Harry Yep <git@hode.co.uk>
Date: Mon, 27 Jul 2026 13:05:05 +0800
Subject: [PATCH 2/7] feat: worker budget grace, package type derivation, stop
 gate reverify

---
 plugins/fusion/agents/deep-reasoner.md      |   2 +-
 plugins/fusion/agents/fast-worker.md        |   2 +-
 plugins/fusion/agents/trivial-worker.md     |   2 +-
 plugins/fusion/scripts/worker-lifecycle.mjs |  74 ++++++++++--
 tests/fusion-worker-lifecycle.test.mjs      | 127 +++++++++++++++++++-
 5 files changed, 190 insertions(+), 17 deletions(-)

diff --git a/plugins/fusion/agents/deep-reasoner.md b/plugins/fusion/agents/deep-reasoner.md
index ded5bfe..7d6ac99 100644
--- a/plugins/fusion/agents/deep-reasoner.md
+++ b/plugins/fusion/agents/deep-reasoner.md
@@ -14,4 +14,4 @@ Read whatever code you need. Never edit files, execute an implementation brief,
 
 Reply with exactly four sections: Conclusion, Reasoning (compressed to what a tech lead needs to judge it), Recommended actions, Risks and open questions. State your confidence. Do not pad; the reader pays for every token. End with `delivery: complete` and `coverage: complete` on separate lines.
 
-The supplied `fusion-brief: v1` envelope is your entire task context. Do not retrieve or reconstruct the parent conversation. Stop and return a partial result when the lifecycle guard reports a wall clock, turn, or token limit. You may make at most one retry after a lifecycle completion check.
+The supplied `fusion-brief: v1` envelope is your entire task context. Do not retrieve or reconstruct the parent conversation. Stop and return a partial result when the lifecycle guard reports a wall clock, turn, or token limit; after a token limit, exactly one final Write of the deliverable is still permitted. You may make at most one retry after a lifecycle completion check.
diff --git a/plugins/fusion/agents/fast-worker.md b/plugins/fusion/agents/fast-worker.md
index 78128f3..ef1d8df 100644
--- a/plugins/fusion/agents/fast-worker.md
+++ b/plugins/fusion/agents/fast-worker.md
@@ -12,7 +12,7 @@ You execute resolved, bounded implementation briefs. A plan section handed down
 
 Always run the verification command from the spec before reporting. Reply with a file level summary of what changed, the tail of the verification output, and anything you were asked to do but could not. End a successful report with `delivery: complete` and `verification: passed` on separate lines.
 
-The supplied `fusion-brief: v1` envelope is your entire task context. Do not retrieve or reconstruct the parent conversation. Stop and return a partial result when the lifecycle guard reports a wall clock, no-progress, turn, or token limit. You may make at most one retry after a lifecycle completion check.
+The supplied `fusion-brief: v1` envelope is your entire task context. Do not retrieve or reconstruct the parent conversation. Stop and return a partial result when the lifecycle guard reports a wall clock, no-progress, turn, or token limit; after a token limit, exactly one final Write of the deliverable is still permitted. You may make at most one retry after a lifecycle completion check.
 
 ## Execution speed
 
diff --git a/plugins/fusion/agents/trivial-worker.md b/plugins/fusion/agents/trivial-worker.md
index 6e27d86..c1de76e 100644
--- a/plugins/fusion/agents/trivial-worker.md
+++ b/plugins/fusion/agents/trivial-worker.md
@@ -12,4 +12,4 @@ You execute small, exactly specified tasks quickly. Follow the spec exactly; if
 
 For a change brief, run its verification command and reply with what changed plus the verification output. For an analysis, drafting, or research brief, check the result against its explicit acceptance criteria and return the requested deliverable. End with `delivery: complete` plus `verification: passed` for a change or `coverage: complete` for analysis.
 
-The supplied `fusion-brief: v1` envelope is your entire task context. Do not retrieve or reconstruct the parent conversation. Stop and return a partial result when the lifecycle guard reports a wall clock, no-progress, turn, or token limit. You may make at most one retry after a lifecycle completion check.
+The supplied `fusion-brief: v1` envelope is your entire task context. Do not retrieve or reconstruct the parent conversation. Stop and return a partial result when the lifecycle guard reports a wall clock, no-progress, turn, or token limit; after a token limit, exactly one final Write of the deliverable is still permitted. You may make at most one retry after a lifecycle completion check.
diff --git a/plugins/fusion/scripts/worker-lifecycle.mjs b/plugins/fusion/scripts/worker-lifecycle.mjs
index f03a76d..cfc9cbe 100644
--- a/plugins/fusion/scripts/worker-lifecycle.mjs
+++ b/plugins/fusion/scripts/worker-lifecycle.mjs
@@ -122,7 +122,8 @@ function allowAgentWithTaskId(toolInput, taskId, additionalContext = null) {
   const prompt = promptText(toolInput);
   const marker = `fusion-task-id: ${taskId}`;
   const lines = prompt.split(/\r?\n/).filter((line) => !/^fusion-task-id:\s*/i.test(line.trim()));
-  lines.splice(lines[0]?.trim() === "fusion-brief: v1" ? 1 : 0, 0, marker);
+  const firstNonBlankLine = lines.findIndex((line) => line.trim());
+  lines.splice(lines[firstNonBlankLine]?.trim() === "fusion-brief: v1" ? firstNonBlankLine + 1 : 0, 0, marker);
   return { hookSpecificOutput: { hookEventName: "PreToolUse", permissionDecision: "allow", updatedInput: { ...toolInput, prompt: lines.join("\n") }, ...(additionalContext ? { additionalContext } : {}) } };
 }
 
@@ -388,6 +389,24 @@ function createDispatch(input, agentType, userBackgroundAuthorized, env, validat
   const taskId = createWorkerTaskId(input.session_id, input.tool_use_id);
   const prompt = promptText(input.tool_input);
   const collectorIdentity = canonicalWorkerAgentType(agentType) === "fusion:job-collector" ? collectorRequestIdentity(prompt) : null;
+  const fallbackPackageType = () => {
+    if (!PEER_WRAPPER_AGENTS.has(agentType)) {
+      return /^verification:\s*\S/im.test(prompt) ? "implementation" : "consult";
+    }
+    const packageTypeLines = prompt.split(/\r?\n/).filter((line) => /^package-type\s*:/i.test(line.trim()));
+    const explicitPackageType = packageTypeLines.length === 1
+      ? packageTypeLines[0].match(/^package-type\s*:\s*(implementation|consult|review|research|design)\s*$/i)?.[1]?.toLowerCase()
+      : null;
+    if (explicitPackageType) {
+      return explicitPackageType;
+    }
+    const separator = /(?:^|\s)--(?=\s|$)/.exec(prompt);
+    const peerPromptPrefix = separator ? prompt.slice(0, separator.index) : prompt;
+    if (/(?:^|\s)--(?:write|transport-default-write)(?=\s|$)/.test(peerPromptPrefix)) {
+      return "implementation";
+    }
+    return /^verification:\s*\S/im.test(prompt) ? "implementation" : "consult";
+  };
   let parentTranscriptBytesAtDispatch = null;
   try {
     parentTranscriptBytesAtDispatch = fs.statSync(input.transcript_path).size;
@@ -405,7 +424,7 @@ function createDispatch(input, agentType, userBackgroundAuthorized, env, validat
     userBackgroundAuthorized,
     parentTranscriptPath: typeof input.transcript_path === "string" ? input.transcript_path : null,
     parentTranscriptBytesAtDispatch,
-    packageType: validation.packageType ?? (/^verification:\s*\S/im.test(prompt) ? "implementation" : "consult"),
+    packageType: validation.packageType ?? fallbackPackageType(),
     briefBytes: validation.briefBytes ?? null,
     completionContract: completionContract(agentType, prompt),
     ...collectorIdentity,
@@ -725,7 +744,23 @@ function handlePreToolUse(input, env) {
   const failure = workerBudgetFailure(refreshed);
   if (failure) {
     markBudgetFailure(refreshed, failure, env);
-    writeOutput(denyTool(`Fusion worker ${refreshed.taskId} must stop: ${failure.reason}. Return a concise partial result now; do not retry or call more tools.`));
+    if (failure.failureKind === "token_limit" && input.tool_name === "Write" && !refreshed.terminalWriteGraceUsedAt) {
+      const now = new Date().toISOString();
+      let terminalWriteGraceUsed = false;
+      updateLifecycleWorkerRecord(refreshed.taskId, env, (current) => {
+        if (!current || isTerminalWorkerStatus(current.transportStatus) || current.terminalWriteGraceUsedAt) {
+          return null;
+        }
+        terminalWriteGraceUsed = true;
+        return { ...current, terminalWriteGraceUsedAt: now };
+      });
+      if (terminalWriteGraceUsed) {
+        writeOutput({ hookSpecificOutput: { hookEventName: "PreToolUse", permissionDecision: "allow", additionalContext: "Final deliverable write permitted; every further tool call will be denied." } });
+        return;
+      }
+    }
+    const terminalWriteHint = failure.failureKind === "token_limit" ? " If the deliverable file is not yet written, one final Write is permitted." : "";
+    writeOutput(denyTool(`Fusion worker ${refreshed.taskId} must stop: ${failure.reason}. Return a concise partial result now; do not retry or call more tools.${terminalWriteHint}`));
     return;
   }
   const now = new Date().toISOString();
@@ -1133,7 +1168,8 @@ function handlePostToolUse(input, env, failed = false) {
     return;
   }
   const now = new Date().toISOString();
-  let emitWindDown = false;
+  let emitTurnWindDown = false;
+  let emitTokenWindDown = false;
   updateLifecycleWorkerRecord(record.taskId, env, (current) => {
     if (!current || isTerminalWorkerStatus(current.transportStatus)) {
       return null;
@@ -1141,18 +1177,23 @@ function handlePostToolUse(input, env, failed = false) {
     const selectedTranscript = isAttributedTranscriptPath(input.transcript_path, record) ? input.transcript_path : null;
     const refreshed = selectedTranscript ? refreshWorkerTranscript(current, selectedTranscript) : current;
     const progress = !failed && !READ_ONLY_TOOLS.has(input.tool_name);
-    const windDownThreshold = Math.max(0, (refreshed.limits?.maxTurns ?? Number.POSITIVE_INFINITY) - 2);
-    emitWindDown = !isTerminalWorkerStatus(refreshed.transportStatus) && refreshed.windDownContextSentAt == null && (refreshed.turns ?? 0) >= windDownThreshold;
+    const turnWindDownThreshold = Math.max(0, (refreshed.limits?.maxTurns ?? Number.POSITIVE_INFINITY) - 2);
+    const tokenWindDownThreshold = (refreshed.limits?.maxOutputTokens ?? Number.POSITIVE_INFINITY) * 0.85;
+    emitTokenWindDown = !isTerminalWorkerStatus(refreshed.transportStatus) && refreshed.tokenWindDownSentAt == null && (refreshed.usage?.outputTokens ?? 0) >= tokenWindDownThreshold;
+    emitTurnWindDown = !emitTokenWindDown && !isTerminalWorkerStatus(refreshed.transportStatus) && refreshed.windDownContextSentAt == null && (refreshed.turns ?? 0) >= turnWindDownThreshold;
     return {
       ...refreshed,
       lastActivityAt: now,
       inFlightSince: undefined,
       ...(!failed ? { lastLivenessAt: now } : {}),
       ...(progress ? { lastProgressAt: now, progressEvents: (refreshed.progressEvents ?? 0) + 1 } : {}),
-      ...(emitWindDown ? { windDownContextSentAt: now } : {})
+      ...(emitTurnWindDown ? { windDownContextSentAt: now } : {}),
+      ...(emitTokenWindDown ? { tokenWindDownSentAt: now } : {})
     };
   });
-  if (emitWindDown) {
+  if (emitTokenWindDown) {
+    writeOutput(hookOutput(input.hook_event_name, "Token budget wind-down: stop making tool calls and write your final deliverable now."));
+  } else if (emitTurnWindDown) {
     writeOutput(hookOutput(input.hook_event_name, "Turn budget wind-down: stop making tool calls and write your final deliverable now."));
   }
 }
@@ -1191,7 +1232,7 @@ function handleSubagentStart(input, env) {
     : record.completionContract === "coverage"
       ? `End the completed analysis with separate lines \`delivery: complete\` and \`coverage: complete\`. ${verdictEnvelope}`
       : `End a successful execution report with separate lines \`delivery: complete\` and \`verification: passed\`. ${verdictEnvelope}`;
-  writeOutput(hookOutput("SubagentStart", `Fusion task id: ${record.taskId}. Work only from the supplied isolated brief. Do not request or reconstruct the parent transcript. Budgets: ${limits.wallClockMs}ms wall clock, ${limits.stallMs}ms without successful tool activity, ${limits.maxTurns} turns, ${limits.maxOutputTokens} output tokens, ${limits.maxUncachedTokens} uncached tokens. Retry at most once. ${delivery}`));
+  writeOutput(hookOutput("SubagentStart", `Fusion task id: ${record.taskId}. Work only from the supplied isolated brief. Do not request or reconstruct the parent transcript. Budgets: ${limits.wallClockMs}ms wall clock, ${limits.stallMs}ms without successful tool activity, ${limits.maxTurns} turns, ${limits.maxOutputTokens} output tokens, ${limits.maxUncachedTokens} uncached tokens. Retry at most once. At 85 percent of the output token budget, stop making tool calls and write the final deliverable; after the token limit, exactly one final Write of the deliverable is permitted. ${delivery}`));
 }
 
 function handleSubagentStop(input, env) {
@@ -1832,6 +1873,12 @@ export function reverifyInFlightRecords(records, tasks, env = process.env) {
     .filter((record) => record && !terminalCollectedRecord(record) && !terminalTransportObserved(record, runtimeTaskForRecord(record, tasks)));
 }
 
+export function reverifyCancellationRecords(records, env = process.env) {
+  return records
+    .map((record) => readWorkerRecord(record.taskId, env))
+    .filter((record) => record && !isTerminalWorkerStatus(record.transportStatus) && !terminalCollectedRecord(record));
+}
+
 function claimStopAdvisory(sessionId, kind, signature, env) {
   let claimed = false;
   updateWorkerSessionState(sessionId, env, (current) => {
@@ -1904,9 +1951,10 @@ function handleStop(input, env) {
     pending.push(updated);
   }
   const cancellations = pending.filter((record) => record.transportStatus === "cancel_requested" || (!record.userBackgroundAuthorized && !["ready_uncollected", "ready_background"].includes(record.transportStatus) && (record.stopBlockCount ?? 0) >= 6));
-  if (cancellations.length > 0) {
+  const verifiedCancellations = reverifyCancellationRecords(cancellations, env);
+  if (verifiedCancellations.length > 0) {
     const now = new Date().toISOString();
-    const missingRuntimeIds = cancellations.filter((record) => !record.backgroundTaskId && !record.agentId);
+    const missingRuntimeIds = verifiedCancellations.filter((record) => !record.backgroundTaskId && !record.agentId);
     for (const record of missingRuntimeIds) {
       updateLifecycleWorkerRecord(record.taskId, env, (current) => {
         if (!current || isTerminalWorkerStatus(current.transportStatus)) {
@@ -1916,7 +1964,7 @@ function handleStop(input, env) {
       });
     }
     const reapedTaskIds = [];
-    for (const record of cancellations.filter((candidate) => candidate.backgroundTaskId || candidate.agentId)) {
+    for (const record of verifiedCancellations.filter((candidate) => candidate.backgroundTaskId || candidate.agentId)) {
       if (hasBackgroundTasks ? runtimeTaskForRecord(record, tasks) : (record.cancelAttemptCount ?? 0) < 2) {
         continue;
       }
@@ -1934,7 +1982,7 @@ function handleStop(input, env) {
     }
     const reapedSentence = reapedTaskIds.length > 0 ? `Fusion task IDs ${reapedTaskIds.join(", ")} were settled as task_reaped because the harness no longer tracks their runtime tasks.` : "";
     const reapedSuffix = reapedSentence ? ` ${reapedSentence}` : "";
-    const cancelIds = cancellations.filter((record) => !reapedTaskIds.includes(record.taskId)).map((record) => record.backgroundTaskId ?? record.agentId).filter(Boolean);
+    const cancelIds = verifiedCancellations.filter((record) => !reapedTaskIds.includes(record.taskId)).map((record) => record.backgroundTaskId ?? record.agentId).filter(Boolean);
     if (cancelIds.length > 0) {
       writeOutput(blockStop(`Call TaskStop for over-budget task${cancelIds.length === 1 ? "" : "s"} ${cancelIds.join(", ")} and report the cancellation before finishing.${reapedSuffix}`));
       return;
diff --git a/tests/fusion-worker-lifecycle.test.mjs b/tests/fusion-worker-lifecycle.test.mjs
index d9a5485..cc39da8 100644
--- a/tests/fusion-worker-lifecycle.test.mjs
+++ b/tests/fusion-worker-lifecycle.test.mjs
@@ -7,7 +7,7 @@ import { test } from "node:test";
 
 import { fusionRepositoryKey } from "../plugins/fusion/scripts/fusion-stats.mjs";
 import { createWorkerRecord, readWorkerSessionState, readWorkerRecords, recordWorkerAcceptance, updateWorkerRecord, WORKER_COLLECTION_METHODS } from "../plugins/fusion/scripts/lib/worker-state.mjs";
-import { reverifyInFlightRecords, validateWorkerBrief, workerBudgetFailure, workerLimits } from "../plugins/fusion/scripts/worker-lifecycle.mjs";
+import { reverifyCancellationRecords, reverifyInFlightRecords, validateWorkerBrief, workerBudgetFailure, workerLimits } from "../plugins/fusion/scripts/worker-lifecycle.mjs";
 
 const repoRoot = path.join(import.meta.dirname, "..");
 const script = path.join(repoRoot, "plugins", "fusion", "scripts", "worker-lifecycle.mjs");
@@ -409,6 +409,25 @@ test("package-type envelopes validate, default, and persist their byte length",
   assert.strictEqual(JSON.parse(launched.stdout).hookSpecificOutput.permissionDecision, "allow");
   assert.strictEqual(record(box).packageType, "review");
   assert.strictEqual(record(box).briefBytes, Buffer.byteLength(prompt));
+
+  const peerCases = [
+    ["rescue prompt --write", "implementation"],
+    ["rescue prompt --transport-default-write", "implementation"],
+    ["rescue prompt --write\npackage-type: review", "review"],
+    ["bounded peer consult", "consult"],
+    ["rescue prompt -- --write", "consult"]
+  ];
+  for (const [peerPrompt, packageType] of peerCases) {
+    const peer = sandbox(t);
+    const peerDispatch = dispatch(peer, { subagent_type: "codex:codex-rescue", prompt: peerPrompt });
+    run(peer, peerDispatch);
+    run(peer, {
+      ...peerDispatch,
+      hook_event_name: "PostToolUse",
+      tool_response: { isAsync: true, status: "async_launched", agentId: `peer-package-${packageType}` }
+    });
+    assert.strictEqual(record(peer).packageType, packageType);
+  }
 });
 
 test("verification manifest advisories identify missing suites and stay silent for complete verification", (t) => {
@@ -507,6 +526,12 @@ test("the dispatch guard requires a minimal isolated brief and explicit user bac
   assert.strictEqual(JSON.parse(authorized.stdout).hookSpecificOutput.permissionDecision, "allow");
   assert.match(JSON.parse(authorized.stdout).hookSpecificOutput.updatedInput.prompt, /^fusion-brief: v1\nfusion-task-id: fusion-/);
   assert.strictEqual(record(box).expectedDelivery, "manual-background");
+
+  const leadingBlank = sandbox(t);
+  const leadingBlankPrompt = `\n\n${brief()}`;
+  const leadingBlankLaunch = JSON.parse(run(leadingBlank, dispatch(leadingBlank, { prompt: leadingBlankPrompt })).stdout).hookSpecificOutput;
+  assert.strictEqual(leadingBlankLaunch.permissionDecision, "allow");
+  assert.match(leadingBlankLaunch.updatedInput.prompt, /^\n\nfusion-brief: v1\nfusion-task-id: fusion-/);
 });
 
 test("SubagentStart requires verdict envelopes for execution and coverage workers but not collectors", (t) => {
@@ -527,6 +552,8 @@ test("SubagentStart requires verdict envelopes for execution and coverage worker
   assert.match(executionInstruction, /pass and fail counts/);
   assert.match(executionInstruction, /environment findings/);
   assert.match(executionInstruction, /long unified diffs, full test logs, and long quoted file contents/);
+  assert.match(executionInstruction, /At 85 percent of the output token budget/);
+  assert.match(executionInstruction, /exactly one final Write of the deliverable is permitted/);
 
   const coverage = sandbox(t);
   run(coverage, dispatch(coverage, {
@@ -3396,6 +3423,50 @@ test("worker tool calls are denied after the wall clock budget and cancellation
   assert.strictEqual(record(box).failureKind, "timeout");
 });
 
+test("a token limit permits exactly one final deliverable Write", (t) => {
+  const box = sandbox(t);
+  run(box, dispatch(box));
+  run(box, {
+    hook_event_name: "SubagentStart",
+    session_id: "session-1",
+    cwd: box.cwd,
+    transcript_path: box.transcript,
+    agent_id: "token-grace",
+    agent_type: "fusion:fast-worker"
+  });
+  const taskId = record(box).taskId;
+  updateWorkerRecord(taskId, envFor(box), (current) => ({
+    ...current,
+    usage: { inputTokens: 0, cacheCreationInputTokens: 0, cacheReadInputTokens: 0, outputTokens: 48_000, totalTokens: 48_000, uncachedTokens: 48_000 }
+  }));
+  const write = {
+    hook_event_name: "PreToolUse",
+    session_id: "session-1",
+    cwd: box.cwd,
+    transcript_path: box.transcript,
+    agent_id: "token-grace",
+    agent_type: "fusion:fast-worker",
+    tool_name: "Write",
+    tool_input: { file_path: path.join(box.cwd, "deliverable.md"), content: "deliverable" }
+  };
+
+  const firstWrite = JSON.parse(run(box, write).stdout).hookSpecificOutput;
+  assert.strictEqual(firstWrite.permissionDecision, "allow");
+  assert.strictEqual(firstWrite.additionalContext, "Final deliverable write permitted; every further tool call will be denied.");
+  const graceUsedAt = record(box).terminalWriteGraceUsedAt;
+  assert.ok(graceUsedAt);
+
+  const expectedReason = `Fusion worker ${taskId} must stop: output token budget reached (48000). Return a concise partial result now; do not retry or call more tools. If the deliverable file is not yet written, one final Write is permitted.`;
+  const secondWrite = JSON.parse(run(box, write).stdout).hookSpecificOutput;
+  assert.strictEqual(secondWrite.permissionDecision, "deny");
+  assert.strictEqual(secondWrite.permissionDecisionReason, expectedReason);
+  assert.strictEqual(record(box).terminalWriteGraceUsedAt, graceUsedAt);
+
+  const bash = JSON.parse(run(box, { ...write, tool_name: "Bash", tool_input: { command: "true" } }).stdout).hookSpecificOutput;
+  assert.strictEqual(bash.permissionDecision, "deny");
+  assert.strictEqual(bash.permissionDecisionReason, expectedReason);
+});
+
 test("an explicitly authorized background worker still requires TaskStop after its budget expires", (t) => {
   const box = sandbox(t);
   const limits = { FUSION_WORKER_WALL_CLOCK_MS: "1", FUSION_WORKER_STALL_MS: "999999" };
@@ -3983,6 +4054,46 @@ test("PostToolUse emits turn wind-down context once at two turns below the cap",
   assert.strictEqual(record(box).windDownContextSentAt, notifiedAt);
 });
 
+test("PostToolUse emits token wind-down context once at 85 percent of the cap", (t) => {
+  const box = sandbox(t);
+  const workerDispatch = dispatch(box);
+  run(box, workerDispatch);
+  run(box, {
+    hook_event_name: "SubagentStart",
+    session_id: "session-1",
+    cwd: box.cwd,
+    transcript_path: box.transcript,
+    agent_id: "token-wind-down",
+    agent_type: "fusion:fast-worker"
+  });
+  const workerTranscript = path.join(box.root, "agent-token-wind-down.jsonl");
+  fs.writeFileSync(workerTranscript, `${JSON.stringify({ type: "assistant", requestId: "token-before", message: { usage: { output_tokens: 40_799 } } })}\n`, "utf8");
+  const payload = {
+    hook_event_name: "PostToolUse",
+    session_id: "session-1",
+    cwd: box.cwd,
+    transcript_path: workerTranscript,
+    agent_id: "token-wind-down",
+    agent_type: "fusion:fast-worker",
+    tool_name: "Read"
+  };
+
+  const before = run(box, payload);
+  assert.strictEqual(before.stdout, "");
+  assert.strictEqual(record(box).tokenWindDownSentAt, undefined);
+
+  fs.appendFileSync(workerTranscript, `${JSON.stringify({ type: "assistant", requestId: "token-threshold", message: { usage: { output_tokens: 1 } } })}\n`, "utf8");
+  const threshold = JSON.parse(run(box, payload).stdout).hookSpecificOutput;
+  assert.strictEqual(threshold.additionalContext, "Token budget wind-down: stop making tool calls and write your final deliverable now.");
+  assert.strictEqual(record(box).usage.outputTokens, 40_800);
+  const notifiedAt = record(box).tokenWindDownSentAt;
+  assert.ok(notifiedAt);
+
+  const repeated = run(box, payload);
+  assert.strictEqual(repeated.stdout, "");
+  assert.strictEqual(record(box).tokenWindDownSentAt, notifiedAt);
+});
+
 test("Stop combines multi-record collection and settlement instructions", (t) => {
   const box = sandbox(t);
   const outputFile = path.join(box.root, "peer-terminal.output");
@@ -4324,6 +4435,20 @@ test("in-flight advisory re-verification excludes a record collected after the i
   assert.deepStrictEqual(reverifyInFlightRecords([candidate], [{ id: "reverify-collected", status: "running" }], envFor(box)), []);
 });
 
+test("cancellation demand re-verification excludes a record that turned done on disk", (t) => {
+  const box = sandbox(t);
+  const worker = createWorkerRecord({
+    taskId: `fusion-${"d".repeat(24)}`,
+    sessionId: "session-1",
+    agentType: "fusion:fast-worker",
+    workspaceRoot: box.cwd
+  }, envFor(box));
+  const candidate = updateWorkerRecord(worker.taskId, envFor(box), (current) => ({ ...current, agentId: "reverify-cancellation", backgroundTaskId: "reverify-cancellation", transportStatus: "cancel_requested", failureKind: "timeout" }));
+  updateWorkerRecord(worker.taskId, envFor(box), (current) => ({ ...current, transportStatus: "done", finishedAt: new Date().toISOString() }));
+
+  assert.deepStrictEqual(reverifyCancellationRecords([candidate], envFor(box)), []);
+});
+
 test("queued accepted verdicts re-arm settlement after cancelled and incomplete hook transitions", (t) => {
   const cancelled = sandbox(t);
   run(cancelled, dispatch(cancelled));

From 176aca1bcebf8d49dd5216a101ed3463868cc4f9 Mon Sep 17 00:00:00 2001
From: Harry Yep <git@hode.co.uk>
Date: Mon, 27 Jul 2026 13:05:05 +0800
Subject: [PATCH 3/7] feat: settlement grammar and near cap telemetry

---
 plugins/fusion/scripts/codex-jobs-monitor.mjs |  41 +-----
 plugins/fusion/scripts/fusion-stats.mjs       | 130 +++++++++++++++++-
 plugins/fusion/skills/stats/SKILL.md          |   2 +-
 tests/codex-jobs-monitor.test.mjs             |   4 +
 tests/fusion-stats.test.mjs                   |  95 ++++++++++++-
 tests/raw-transport-surface.test.mjs          |   2 +-
 6 files changed, 221 insertions(+), 53 deletions(-)

diff --git a/plugins/fusion/scripts/codex-jobs-monitor.mjs b/plugins/fusion/scripts/codex-jobs-monitor.mjs
index d08aa4c..d94ea13 100644
--- a/plugins/fusion/scripts/codex-jobs-monitor.mjs
+++ b/plugins/fusion/scripts/codex-jobs-monitor.mjs
@@ -9,10 +9,10 @@ import { fileURLToPath, pathToFileURL } from "node:url";
 
 import {
   FILE_ENGINE_DESCRIPTORS,
+  appendModelAuditObservation,
   appendTokenUsageObservation,
   fusionRepositoryKey,
   loadModelAuditObservations,
-  modelObservationRank,
   normalizeCodexTokenUsage,
   resolveFusionDataDir,
   fusionWorkspaceKey,
@@ -308,43 +308,6 @@ function ensurePrivateDirectory(directory) {
   }
 }
 
-function appendModelAuditObservation(sidecarPath, observation) {
-  ensurePrivateDirectory(path.dirname(sidecarPath));
-  const lockPath = `${sidecarPath}.lock`;
-  let lock;
-  try {
-    lock = fs.openSync(lockPath, fs.constants.O_CREAT | fs.constants.O_EXCL | fs.constants.O_WRONLY, 0o600);
-    fs.fchmodSync(lock, 0o600);
-  } catch (error) {
-    if (error?.code === "EEXIST") {
-      return false;
-    }
-    throw error;
-  }
-  try {
-    if (fs.existsSync(sidecarPath)) {
-      fs.chmodSync(sidecarPath, 0o600);
-    }
-    const observations = loadModelAuditObservations(sidecarPath);
-    const current = observations.get(observation.jobId);
-    if (!current || modelObservationRank(observation) > modelObservationRank(current)) {
-      observations.set(observation.jobId, observation);
-    }
-    const replacementPath = `${sidecarPath}.${process.pid}.${Date.now()}.tmp`;
-    try {
-      fs.writeFileSync(replacementPath, `${[...observations.values()].map((entry) => JSON.stringify(entry)).join("\n")}\n`, { encoding: "utf8", mode: 0o600 });
-      fs.chmodSync(replacementPath, 0o600);
-      fs.renameSync(replacementPath, sidecarPath);
-    } finally {
-      fs.rmSync(replacementPath, { force: true });
-    }
-    return true;
-  } finally {
-    fs.closeSync(lock);
-    fs.rmSync(lockPath, { force: true });
-  }
-}
-
 function resolveSessionsDir(env = process.env) {
   const override = env[SESSIONS_DIR_ENV];
   if (override && String(override).trim()) {
@@ -624,6 +587,8 @@ function terminalLedgerRecord(record, observations, scopeRoot) {
     createdAt: record.createdAt ?? null,
     startedAt: record.startedAt ?? null,
     finishedAt: record.finishedAt ?? record.completedAt ?? record.updatedAt ?? null,
+    timeoutMs: record.timeoutMs ?? null,
+    failureKind: record.failureKind ?? null,
     observedAt: new Date().toISOString(),
     model: observations.model,
     modelSource: observations.modelSource,
diff --git a/plugins/fusion/scripts/fusion-stats.mjs b/plugins/fusion/scripts/fusion-stats.mjs
index d6d1028..12ee27e 100644
--- a/plugins/fusion/scripts/fusion-stats.mjs
+++ b/plugins/fusion/scripts/fusion-stats.mjs
@@ -266,6 +266,43 @@ export function appendTokenUsageObservation(sidecarPath, observation) {
   });
 }
 
+export function appendModelAuditObservation(sidecarPath, observation) {
+  ensurePrivateDirectory(path.dirname(sidecarPath));
+  const lockPath = `${sidecarPath}.lock`;
+  let lock;
+  try {
+    lock = fs.openSync(lockPath, fs.constants.O_CREAT | fs.constants.O_EXCL | fs.constants.O_WRONLY, 0o600);
+    fs.fchmodSync(lock, 0o600);
+  } catch (error) {
+    if (error?.code === "EEXIST") {
+      return false;
+    }
+    throw error;
+  }
+  try {
+    if (fs.existsSync(sidecarPath)) {
+      fs.chmodSync(sidecarPath, 0o600);
+    }
+    const observations = loadModelAuditObservations(sidecarPath);
+    const current = observations.get(observation.jobId);
+    if (!current || modelObservationRank(observation) > modelObservationRank(current)) {
+      observations.set(observation.jobId, observation);
+    }
+    const replacementPath = `${sidecarPath}.${process.pid}.${Date.now()}.tmp`;
+    try {
+      fs.writeFileSync(replacementPath, `${[...observations.values()].map((entry) => JSON.stringify(entry)).join("\n")}\n`, { encoding: "utf8", mode: 0o600 });
+      fs.chmodSync(replacementPath, 0o600);
+      fs.renameSync(replacementPath, sidecarPath);
+    } finally {
+      fs.rmSync(replacementPath, { force: true });
+    }
+    return true;
+  } finally {
+    fs.closeSync(lock);
+    fs.rmSync(lockPath, { force: true });
+  }
+}
+
 export function resolveGitWorkspaceRoot(cwd) {
   const resolved = path.resolve(cwd);
   const result = spawnSync("git", ["-C", resolved, "rev-parse", "--show-toplevel"], { encoding: "utf8" });
@@ -443,6 +480,8 @@ function terminalLedgerRecord(raw, { workspaceRoot = null, workspaceKey = null,
     startedAt: raw?.startedAt ?? null,
     finishedAt: raw?.finishedAt ?? raw?.completedAt ?? raw?.observedAt ?? null,
     completedAt: raw?.finishedAt ?? raw?.completedAt ?? raw?.observedAt ?? null,
+    timeoutMs: raw?.timeoutMs ?? null,
+    failureKind: raw?.failureKind ?? null,
     request: model || effort ? { model, effort } : null,
     _fusionObservedModel: raw?.modelSource === "rollout-turn-context" ? model : null,
     _fusionObservedEffort: raw?.effortSource === "rollout-turn-context" ? effort : null,
@@ -1018,11 +1057,14 @@ function summarizeCodexSkuTelemetry(entries) {
       if (entry.createdAtMs < cutoff) {
         continue;
       }
-      const row = trendRows.get(entry.sku) ?? { sku: entry.sku, jobs: 0, outputTokens: 0, outputTokenOverflow: false, timeouts: 0 };
+      const row = trendRows.get(entry.sku) ?? { sku: entry.sku, jobs: 0, outputTokens: 0, outputTokenOverflow: false, timeouts: 0, nearCap: 0 };
       row.jobs += 1;
       if (entry.failureKind === "timeout") {
         row.timeouts += 1;
       }
+      if (entry.failureKind === "timeout" || (entry.durationSeconds != null && entry.timeoutMs != null && entry.timeoutMs / 1000 - entry.durationSeconds <= 30)) {
+        row.nearCap += 1;
+      }
       if (entry.outputTokens != null && !row.outputTokenOverflow) {
         const outputTokens = row.outputTokens + entry.outputTokens;
         if (Number.isSafeInteger(outputTokens)) {
@@ -1060,7 +1102,7 @@ function summarizeCodexSkuTelemetry(entries) {
   return {
     last7DaysBySku: [...trendRows.values()]
       .sort((left, right) => left.sku.localeCompare(right.sku))
-      .map((row) => ({ ...row, timeoutShare: row.timeouts / row.jobs })),
+      .map((row) => ({ ...row, timeoutShare: row.timeouts / row.jobs, nearCapShare: row.nearCap / row.jobs })),
     laneSignalDrift: laneSignalDrift.sort((left, right) => left.sku.localeCompare(right.sku))
   };
 }
@@ -1152,7 +1194,9 @@ export function fileBasedEngineStats(descriptor, { all = false, env = process.en
           terminal: CODEX_TERMINAL_STATUSES.has(job.status),
           acceptance: semanticAcceptance(raw),
           failureKind: nonEmptyString(raw?.failureKind),
-          outputTokens: codexUsage.usage?.outputTokens ?? null
+          outputTokens: codexUsage.usage?.outputTokens ?? null,
+          durationSeconds: job.durationSeconds,
+          timeoutMs: raw?.timeoutMs ?? null
         });
       }
     }
@@ -1954,11 +1998,37 @@ function scanDeadWorkspaces(descriptor, env) {
   return candidates;
 }
 
+function findUnusedAcceptanceObservations(env) {
+  const root = path.join(resolveFusionDataDir(env), "observations");
+  let workspaces;
+  try {
+    workspaces = fs.readdirSync(root, { withFileTypes: true });
+  } catch {
+    return [];
+  }
+  const candidates = [];
+  for (const workspace of workspaces) {
+    if (!workspace.isDirectory()) {
+      continue;
+    }
+    const file = path.join(root, workspace.name, "acceptance.jsonl");
+    try {
+      if (fs.lstatSync(file).isFile()) {
+        candidates.push({ key: workspace.name, file });
+      }
+    } catch {
+      void 0;
+    }
+  }
+  return candidates;
+}
+
 export function findDeadWorkspaces(env = process.env) {
   const dead = {};
   for (const descriptor of WORKSPACE_ENGINE_DESCRIPTORS) {
     dead[descriptor.id] = scanDeadWorkspaces(descriptor, env);
   }
+  dead.acceptance = findUnusedAcceptanceObservations(env);
   return dead;
 }
 
@@ -1969,6 +2039,9 @@ export function pruneDeadWorkspaces({ env = process.env, yes = false, beforeRemo
   }
   const removed = {};
   for (const [engineId, candidates] of Object.entries(dead)) {
+    if (engineId === "acceptance") {
+      continue;
+    }
     removed[engineId] = [];
     const descriptor = WORKSPACE_ENGINE_DESCRIPTORS.find((entry) => entry.id === engineId);
     for (const candidate of candidates) {
@@ -1982,6 +2055,18 @@ export function pruneDeadWorkspaces({ env = process.env, yes = false, beforeRemo
       removed[engineId].push(candidate);
     }
   }
+  removed.acceptance = [];
+  for (const candidate of dead.acceptance ?? []) {
+    try {
+      if (!fs.lstatSync(candidate.file).isFile()) {
+        continue;
+      }
+      fs.rmSync(candidate.file, { force: true });
+      removed.acceptance.push(candidate);
+    } catch {
+      void 0;
+    }
+  }
   return { applied: true, dead: removed };
 }
 
@@ -2064,9 +2149,9 @@ function renderCodexSkuTrend(lines, rows) {
   if (!Array.isArray(rows) || rows.length === 0) {
     return;
   }
-  lines.push("", "By SKU, last 7 days:", "SKU | jobs | output tokens | timeout share");
+  lines.push("", "By SKU, last 7 days:", "SKU | jobs | output tokens | timeout share | near-cap share");
   for (const row of rows) {
-    lines.push(`${row.sku} | ${row.jobs} | ${row.outputTokenOverflow ? "overflow" : row.outputTokens} | ${formatShare(row.timeoutShare)}`);
+    lines.push(`${row.sku} | ${row.jobs} | ${row.outputTokenOverflow ? "overflow" : row.outputTokens} | ${formatShare(row.timeoutShare)} | ${formatShare(row.nearCapShare)}`);
   }
 }
 
@@ -2476,6 +2561,14 @@ function parseRecordArguments(argv, { allowPerPairReasons = false } = {}) {
     if (token === "--record") {
       records.push(parseRecordPair(argv[index + 1]));
       index += 1;
+      while (typeof argv[index + 1] === "string" && !argv[index + 1].startsWith("--")) {
+        try {
+          records.push(parseRecordPair(argv[index + 1]));
+        } catch {
+          break;
+        }
+        index += 1;
+      }
       continue;
     }
     if (token === "--reason-for") {
@@ -2597,6 +2690,15 @@ function isStrictDirectRecordArguments(argv) {
         parseRecordPair(argv[index + 1]);
         records += 1;
         index += 1;
+        while (typeof argv[index + 1] === "string" && !argv[index + 1].startsWith("--")) {
+          try {
+            parseRecordPair(argv[index + 1]);
+          } catch {
+            break;
+          }
+          records += 1;
+          index += 1;
+        }
         continue;
       }
       if (token === "--source") {
@@ -2625,7 +2727,21 @@ function isStrictDirectRecordArguments(argv) {
 
 function isStrictDirectReportOrMaintenanceArguments(argv) {
   const flags = new Set(["--json", "--audit", "--all", "--prune-dead"]);
-  return argv.length > 0 && argv.every((token) => flags.delete(token));
+  return argv.every((token) => flags.delete(token));
+}
+
+function hasRejectedRecordPair(argv) {
+  for (let index = 0; index < argv.length; index += 1) {
+    if (argv[index] !== "--record") {
+      continue;
+    }
+    for (let pairIndex = index + 1; typeof argv[pairIndex] === "string" && !argv[pairIndex].startsWith("--"); pairIndex += 1) {
+      if (parseRecordPair(argv[pairIndex]).verdict === "rejected") {
+        return true;
+      }
+    }
+  }
+  return false;
 }
 
 function writeRecordConfirmation({ kind, engine = null, jobId = null, worker = null, queued = false, asJson, stdout }) {
@@ -2818,7 +2934,7 @@ function runCli(argv = process.argv.slice(2)) {
     return;
   }
   if (isStrictDirectRecordArguments(argv) || isStrictDirectReportOrMaintenanceArguments(argv)) {
-    if (argv.some((token, index) => token === "--record" && parseRecordPair(argv[index + 1]).verdict === "rejected")) {
+    if (hasRejectedRecordPair(argv)) {
       process.stderr.write("Rejected verdicts require --reason through the raw-args transport. For batch settlements, use --reason-for <id> <text> for each rejected pair.\n");
       process.exitCode = 1;
       return;
diff --git a/plugins/fusion/skills/stats/SKILL.md b/plugins/fusion/skills/stats/SKILL.md
index 6cb7f88..10b46f8 100644
--- a/plugins/fusion/skills/stats/SKILL.md
+++ b/plugins/fusion/skills/stats/SKILL.md
@@ -5,7 +5,7 @@ argument-hint: '[--all] [--session [id]] [--trace] [--audit [--days <n>]] [--rec
 allowed-tools: Bash(node:*), Read, Write
 ---
 
-Treat the complete raw request as opaque data and never place any part of it in Bash. Run `node "${CLAUDE_PLUGIN_ROOT}/scripts/fusion-stats.mjs" transport-create` in one foreground Bash call, parse the returned JSON, accept only its 48 character lowercase hexadecimal token, and use `Read` once on the returned file, requiring it to be empty. Use `Write` to replace that same file with the raw request exactly as received. Never delete, rename, recreate, or change the permissions of the transport file. Run `node "${CLAUDE_PLUGIN_ROOT}/scripts/fusion-stats.mjs" --raw-args-token TOKEN` in one foreground Bash call, replacing only `TOKEN` with the validated token. If Read fails, the file is not empty, or Write fails, run `node "${CLAUDE_PLUGIN_ROOT}/scripts/fusion-stats.mjs" transport-discard --raw-args-token TOKEN`. Never use shell redirection, encoding, command substitution, backticks, environment variables, or heredocs for the raw request. The sole carve-outs are two strict direct forms: a verdict settlement made only of repeatable `--record <id>=<accepted|rejected|unverified>[:accept-failed-transport]` pairs, optional `--source collector|main-loop`, and the value free `--json` and `--accept-failed-transport` flags; and a value free report or maintenance request made only of any combination of `--json`, `--audit`, `--all`, and `--prune-dead`, each at most once. Those arguments may be passed directly. Anything else, including every `--reason` text value, every `--failure-kind` and `--failure-kind-for` value, and `--session <id>`, still uses the raw-args transport exactly as described above. A nonterminal worker settlement prints `Queued accepted for <taskId>; applies when the record turns terminal.` with the selected verdict in place of `accepted`; queued accepted applies only when terminal transport is done, while any other terminal outcome clears it, records `pendingVerdictError`, and re-arms settlement.
+Treat the complete raw request as opaque data and never place any part of it in Bash. Run `node "${CLAUDE_PLUGIN_ROOT}/scripts/fusion-stats.mjs" transport-create` in one foreground Bash call, parse the returned JSON, accept only its 48 character lowercase hexadecimal token, and use `Read` once on the returned file, requiring it to be empty. Use `Write` to replace that same file with the raw request exactly as received. Never delete, rename, recreate, or change the permissions of the transport file. Run `node "${CLAUDE_PLUGIN_ROOT}/scripts/fusion-stats.mjs" --raw-args-token TOKEN` in one foreground Bash call, replacing only `TOKEN` with the validated token. If Read fails, the file is not empty, or Write fails, run `node "${CLAUDE_PLUGIN_ROOT}/scripts/fusion-stats.mjs" transport-discard --raw-args-token TOKEN`. Never use shell redirection, encoding, command substitution, backticks, environment variables, or heredocs for the raw request. The sole carve-outs are two strict direct forms: a verdict settlement made only of repeatable `--record` flags, each carrying one or more `<id>=<accepted|rejected|unverified>[:accept-failed-transport]` pairs, optional `--source collector|main-loop`, and the value free `--json` and `--accept-failed-transport` flags; and a value free report or maintenance request made only of any combination of `--json`, `--audit`, `--all`, and `--prune-dead`, each at most once. Those arguments may be passed directly. Anything else, including every `--reason` text value, every `--failure-kind` and `--failure-kind-for` value, and `--session <id>`, still uses the raw-args transport exactly as described above. A nonterminal worker settlement prints `Queued accepted for <taskId>; applies when the record turns terminal.` with the selected verdict in place of `accepted`; queued accepted applies only when terminal transport is done, while any other terminal outcome clears it, records `pendingVerdictError`, and re-arms settlement.
 
 Present report output to the user as-is. It is already compact markdown; do not add prose around it. An unavailable engine section is expected when that peer's plugin is not installed or has no job state; do not treat it as an error. The underlying script accepts `--json` for machine readable output. `--audit` summarizes the last seven days of retained inline guard events by default; add `--all` for the full retained window or `--session <id>` for one session. `/fusion:stats --record` writes the verdict into the Codex job record through the companion's `record-acceptance` subcommand. An `accepted` verdict for a job whose transport failed requires `--accept-failed-transport`. The acceptance forms are used only after collection and verification; return the confirmation line without turning transport completion into acceptance on its own. A judgment shaped rejection may carry `--failure-kind` with one of `intent_override`, `scope_rewrite`, `wrong_approach`, or `style_mismatch`; it rides the raw args transport like `--reason` and lands in the worker's `acceptanceFailureKind` and the engine record's `semanticFailureKind`. Preserve the report's acceptance anomalies block when it appears.
 
diff --git a/tests/codex-jobs-monitor.test.mjs b/tests/codex-jobs-monitor.test.mjs
index ce5d213..3808b2c 100644
--- a/tests/codex-jobs-monitor.test.mjs
+++ b/tests/codex-jobs-monitor.test.mjs
@@ -1141,6 +1141,8 @@ test("canonical terminal records use direct model and token evidence without rol
     status: "running",
     threadId,
     turnId,
+    timeoutMs: 60000,
+    failureKind: "timeout",
     request: { model: "requested-model", effort: "low" },
     resolvedModel: "actual-model",
     resolvedEffort: "xhigh"
@@ -1171,6 +1173,8 @@ test("canonical terminal records use direct model and token evidence without rol
   });
   assert.strictEqual(terminal.transportStatus, "done");
   assert.strictEqual(terminal.finishedAt, finishedAt);
+  assert.strictEqual(terminal.timeoutMs, 60000);
+  assert.strictEqual(terminal.failureKind, "timeout");
   assert.strictEqual(terminal.model, "actual-model");
   assert.strictEqual(terminal.modelSource, "job-record");
   assert.strictEqual(terminal.effort, "xhigh");
diff --git a/tests/fusion-stats.test.mjs b/tests/fusion-stats.test.mjs
index 00ba460..497747e 100644
--- a/tests/fusion-stats.test.mjs
+++ b/tests/fusion-stats.test.mjs
@@ -288,6 +288,8 @@ test("aggregates the canonical Codex lifecycle and direct token usage", (t) => {
     createdAt: "2026-07-02T06:01:00.000Z",
     startedAt: "2026-07-02T06:01:00.000Z",
     finishedAt: "2026-07-02T06:01:03.000Z",
+    timeoutMs: 3000,
+    failureKind: "timeout",
     tokenUsageAvailability: "unavailable"
   });
 
@@ -301,6 +303,10 @@ test("aggregates the canonical Codex lifecycle and direct token usage", (t) => {
   assert.strictEqual(stats.tokenUsage.jobsWithUsage, 1);
   assert.strictEqual(stats.tokenUsage.jobsWithoutUsage, 1);
   assert.deepStrictEqual(stats.tokenUsage.totals, { inputTokens: 120, cachedInputTokens: 40, outputTokens: 30, reasoningOutputTokens: 12, totalTokens: 150 });
+  assert.deepStrictEqual(stats.last7DaysBySku, [
+    { sku: "gpt-5.4@high", jobs: 1, outputTokens: 30, outputTokenOverflow: false, timeouts: 0, nearCap: 0, timeoutShare: 0, nearCapShare: 0 },
+    { sku: "unknown@unavailable", jobs: 1, outputTokens: 0, outputTokenOverflow: false, timeouts: 1, nearCap: 1, timeoutShare: 1, nearCapShare: 1 }
+  ]);
 });
 
 test("Codex stats fail closed when exact token aggregation exceeds the safe integer range", (t) => {
@@ -414,6 +420,9 @@ test("Codex renders per-SKU trends from the seven days ending at the newest reco
     status: "done",
     jobClass: "task",
     createdAt: "2026-07-19T00:00:00.000Z",
+    startedAt: "2026-07-19T00:00:00.000Z",
+    finishedAt: "2026-07-19T00:00:40.000Z",
+    timeoutMs: 60000,
     request: { model: "gpt-trend", effort: "xhigh" },
     tokenUsageAvailability: "available",
     tokenUsage: usage(11)
@@ -431,11 +440,11 @@ test("Codex renders per-SKU trends from the seven days ending at the newest reco
 
   const stats = codexStats({ env: { FUSION_CODEX_STATE: stateRoot, FUSION_DATA_DIR: fusionData }, cwd: dir });
   assert.deepStrictEqual(stats.last7DaysBySku, [
-    { sku: "gpt-trend@high", jobs: 1, outputTokens: 7, outputTokenOverflow: false, timeouts: 0, timeoutShare: 0 },
-    { sku: "gpt-trend@xhigh", jobs: 2, outputTokens: 24, outputTokenOverflow: false, timeouts: 1, timeoutShare: 0.5 }
+    { sku: "gpt-trend@high", jobs: 1, outputTokens: 7, outputTokenOverflow: false, timeouts: 0, nearCap: 0, timeoutShare: 0, nearCapShare: 0 },
+    { sku: "gpt-trend@xhigh", jobs: 2, outputTokens: 24, outputTokenOverflow: false, timeouts: 1, nearCap: 2, timeoutShare: 0.5, nearCapShare: 1 }
   ]);
   const rendered = renderFusionStats({ scope: dir, codex: stats });
-  assert.match(rendered, /By SKU, last 7 days:\nSKU \| jobs \| output tokens \| timeout share\ngpt-trend@high \| 1 \| 7 \| 0\.0%\ngpt-trend@xhigh \| 2 \| 24 \| 50\.0%/);
+  assert.match(rendered, /By SKU, last 7 days:\nSKU \| jobs \| output tokens \| timeout share \| near-cap share\ngpt-trend@high \| 1 \| 7 \| 0\.0% \| 0\.0%\ngpt-trend@xhigh \| 2 \| 24 \| 50\.0% \| 100\.0%/);
   assert.doesNotMatch(rendered, /gpt-old@low \|/);
 });
 
@@ -983,6 +992,26 @@ test("--prune-dead --yes removes only the all-dead workspace dirs and spares liv
   assert.strictEqual(after.grok.length, 0);
 });
 
+test("--prune-dead lists and removes unused acceptance observations", (t) => {
+  const dir = sandbox(t);
+  const stateRoot = path.join(dir, "state");
+  const fusionData = path.join(dir, "fusion-data");
+  const key = "retired-acceptance";
+  const file = path.join(fusionData, "observations", key, "acceptance.jsonl");
+  fs.mkdirSync(path.dirname(file), { recursive: true });
+  fs.writeFileSync(file, "{\"jobId\":\"obsolete\"}\n");
+
+  const dryRun = run({ cwd: dir, codexState: stateRoot }, ["--prune-dead", "--json"], { FUSION_DATA_DIR: fusionData });
+  assert.strictEqual(dryRun.status, 0, dryRun.stderr);
+  assert.deepStrictEqual(JSON.parse(dryRun.stdout).dead.acceptance, [{ key, file }]);
+  assert.ok(fs.existsSync(file));
+
+  const confirmed = run({ cwd: dir, codexState: stateRoot }, ["--prune-dead", "--yes", "--json"], { FUSION_DATA_DIR: fusionData });
+  assert.strictEqual(confirmed.status, 0, confirmed.stderr);
+  assert.deepStrictEqual(JSON.parse(confirmed.stdout).dead.acceptance, [{ key, file }]);
+  assert.strictEqual(fs.existsSync(file), false);
+});
+
 test("pruneDeadWorkspaces never touches a workspace with any running job even without a live cwd", (t) => {
   const dir = sandbox(t);
   const grokDataDir = path.join(dir, "grok-data");
@@ -1233,13 +1262,15 @@ test("terminal ledgers supplement cleaned jobs and safely dedupe live state", (t
     {
       schemaVersion: 1,
       jobId: "cleaned-job",
-      transportStatus: "done",
+      transportStatus: "error",
       workspaceRoot: dir,
       sessionId: "session-ledger",
       kind: "review",
       createdAt: "2026-07-14T00:00:00.000Z",
       startedAt: "2026-07-14T00:00:01.000Z",
       finishedAt: "2026-07-14T00:00:11.000Z",
+      timeoutMs: 60000,
+      failureKind: "timeout",
       observedAt: "2026-07-14T00:00:12.000Z",
       model: "ledger-model",
       effort: "high",
@@ -1250,11 +1281,14 @@ test("terminal ledgers supplement cleaned jobs and safely dedupe live state", (t
 
   const recovered = codexStats({ env: { FUSION_CODEX_STATE: stateRoot, FUSION_DATA_DIR: path.join(dir, "fusion") }, cwd: dir });
   assert.strictEqual(recovered.totalJobs, 1);
-  assert.deepStrictEqual(recovered.byTransportStatus, { done: 1 });
+  assert.deepStrictEqual(recovered.byTransportStatus, { error: 1 });
   assert.deepStrictEqual(recovered.byAcceptance, { unverified: 1 });
   assert.deepStrictEqual(recovered.byModel, { "ledger-model@high": 1 });
   assert.strictEqual(recovered.evidence.recoveredTerminalJobs, 1);
   assert.deepStrictEqual(recovered.tokenUsage.totals, { inputTokens: 100, cachedInputTokens: 40, outputTokens: 20, reasoningOutputTokens: 5, totalTokens: 120 });
+  assert.deepStrictEqual(recovered.last7DaysBySku, [
+    { sku: "ledger-model@high", jobs: 1, outputTokens: 20, outputTokenOverflow: false, timeouts: 1, nearCap: 1, timeoutShare: 1, nearCapShare: 1 }
+  ]);
 
   writeCodexJob(stateRoot, dir, "cleaned-job", {
     status: "done",
@@ -1476,6 +1510,15 @@ test("direct --audit prints the audit report", (t) => {
   assert.match(result.stdout, /# Fusion inline audit/);
 });
 
+test("bare direct invocation prints the Fusion stats report", (t) => {
+  const dir = sandbox(t);
+
+  const result = runDirect({ cwd: dir, codexState: path.join(dir, "missing") });
+
+  assert.strictEqual(result.status, 0, result.stderr);
+  assert.match(result.stdout, /# Fusion stats/);
+});
+
 test("direct --json --all emits the all-workspace JSON report", (t) => {
   const dir = sandbox(t);
   const stateRoot = path.join(dir, "codex-state");
@@ -1558,6 +1601,26 @@ test("--record settles a Fusion task and its Codex peer with direct strict argv"
   assert.strictEqual(readWorkerRecord(taskId, env).acceptance, "accepted");
 });
 
+test("--record settles multiple direct pairs after one flag", (t) => {
+  const dir = sandbox(t);
+  const stateDir = path.join(dir, "worker-state");
+  const firstTaskId = `fusion-${"1".repeat(24)}`;
+  const secondTaskId = `fusion-${"2".repeat(24)}`;
+  const env = { FUSION_WORKER_STATE_DIR: stateDir };
+  createTerminalWorker({ taskId: firstTaskId, env, workspaceRoot: dir });
+  createTerminalWorker({ taskId: secondTaskId, env, workspaceRoot: dir });
+
+  const result = runDirect(
+    { cwd: dir, codexState: path.join(dir, "missing") },
+    ["--record", `${firstTaskId}=accepted`, `${secondTaskId}=unverified`],
+    env
+  );
+
+  assert.strictEqual(result.status, 0, result.stderr);
+  assert.strictEqual(readWorkerRecord(firstTaskId, env).acceptance, "accepted");
+  assert.strictEqual(readWorkerRecord(secondTaskId, env).acceptance, "unverified");
+});
+
 test("--record leaves a terminal worker unsettled when its engine companion fails", (t) => {
   const dir = sandbox(t);
   const stateDir = path.join(dir, "worker-state");
@@ -1760,6 +1823,26 @@ test("--record batches mixed verdicts with per-pair rejection reasons and failur
   assert.deepStrictEqual(JSON.parse(fs.readFileSync(argsFile, "utf8")), ["record-acceptance", "--job-id", jobId, "--acceptance", "rejected", "--source", "main-loop", "--reason", "The result did not satisfy the requested behavior.", "--failure-kind", "scope_rewrite"]);
 });
 
+test("--record settles multiple transported pairs after one flag", (t) => {
+  const dir = sandbox(t);
+  const stateDir = path.join(dir, "worker-state");
+  const firstTaskId = `fusion-${"c".repeat(24)}`;
+  const secondTaskId = `fusion-${"d".repeat(24)}`;
+  const env = { FUSION_WORKER_STATE_DIR: stateDir };
+  createTerminalWorker({ taskId: firstTaskId, env, workspaceRoot: dir });
+  createTerminalWorker({ taskId: secondTaskId, env, workspaceRoot: dir });
+
+  const result = run(
+    { cwd: dir, codexState: path.join(dir, "missing") },
+    ["--record", `${firstTaskId}=accepted`, `${secondTaskId}=unverified`],
+    env
+  );
+
+  assert.strictEqual(result.status, 0, result.stderr);
+  assert.strictEqual(readWorkerRecord(firstTaskId, env).acceptance, "accepted");
+  assert.strictEqual(readWorkerRecord(secondTaskId, env).acceptance, "unverified");
+});
+
 test("--record rejects a raw batch with a rejected pair that has no reason before writing", (t) => {
   const dir = sandbox(t);
   const stateDir = path.join(dir, "worker-state");
@@ -1894,7 +1977,7 @@ test("--record rejects a reason attached to multiple pairs and requires transpor
   const env = { FUSION_WORKER_STATE_DIR: stateDir };
   createTerminalWorker({ taskId: firstTaskId, env, workspaceRoot: dir });
   createTerminalWorker({ taskId: secondTaskId, env, workspaceRoot: dir });
-  const args = ["--record", `${firstTaskId}=accepted`, "--record", `${secondTaskId}=rejected`, "--reason", "mixed verdicts"];
+  const args = ["--record", `${firstTaskId}=accepted`, `${secondTaskId}=rejected`, "--reason", "mixed verdicts"];
 
   const direct = runDirect({ cwd: dir, codexState: path.join(dir, "missing") }, args, env);
   assert.notStrictEqual(direct.status, 0);
diff --git a/tests/raw-transport-surface.test.mjs b/tests/raw-transport-surface.test.mjs
index 8348734..32944f9 100644
--- a/tests/raw-transport-surface.test.mjs
+++ b/tests/raw-transport-surface.test.mjs
@@ -91,7 +91,7 @@ test("every executable private transport surface follows its declared Write tran
     assert.match(content, /file is not empty/, file);
     if (file === "plugins/fusion/skills/stats/SKILL.md") {
       assert.match(content, /sole carve-outs are two strict direct forms: a verdict settlement/, file);
-      assert.match(content, /repeatable `--record <id>=<accepted\|rejected\|unverified>\[:accept-failed-transport\]` pairs, optional `--source collector\|main-loop`, and the value free `--json` and `--accept-failed-transport` flags/, file);
+      assert.match(content, /repeatable `--record` flags, each carrying one or more `<id>=<accepted\|rejected\|unverified>\[:accept-failed-transport\]` pairs, optional `--source collector\|main-loop`, and the value free `--json` and `--accept-failed-transport` flags/, file);
       assert.match(content, /a value free report or maintenance request made only of any combination of `--json`, `--audit`, `--all`, and `--prune-dead`, each at most once/, file);
       assert.match(content, /Anything else, including every `--reason` text value, every `--failure-kind` and `--failure-kind-for` value, and `--session <id>`, still uses the raw-args transport exactly as described above\./, file);
     }

From 14a6d17cf19c66825d827e85cca612c4c921209b Mon Sep 17 00:00:00 2001
From: Harry Yep <git@hode.co.uk>
Date: Mon, 27 Jul 2026 13:05:05 +0800
Subject: [PATCH 4/7] feat: grok observability parity and decline aware narrow
 wave advisory

---
 plugins/fusion/scripts/grok-jobs-observer.mjs | 19 +++-
 .../scripts/inline-delegation-guard.mjs       | 88 ++++++++++++++++++-
 tests/grok-jobs-observer.test.mjs             | 47 +++++++++-
 tests/inline-delegation-guard.test.mjs        | 66 ++++++++++++++
 4 files changed, 216 insertions(+), 4 deletions(-)

diff --git a/plugins/fusion/scripts/grok-jobs-observer.mjs b/plugins/fusion/scripts/grok-jobs-observer.mjs
index c318e0d..e03d4e8 100644
--- a/plugins/fusion/scripts/grok-jobs-observer.mjs
+++ b/plugins/fusion/scripts/grok-jobs-observer.mjs
@@ -4,7 +4,7 @@ import fs from "node:fs";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 
-import { FILE_ENGINE_DESCRIPTORS, appendTokenUsageObservation, resolveFusionDataDir, tokenUsageSidecarPath } from "./fusion-stats.mjs";
+import { FILE_ENGINE_DESCRIPTORS, appendModelAuditObservation, appendTokenUsageObservation, fusionRepositoryKey, modelAuditSidecarPath, resolveFusionDataDir, tokenUsageSidecarPath } from "./fusion-stats.mjs";
 
 const TERMINAL_STATUSES = new Set(["done", "error", "cancelled"]);
 const DEFAULT_POLL_INTERVAL_MS = 15000;
@@ -120,6 +120,7 @@ export function observeGrokJobs({ env = process.env, now = () => new Date().toIS
       continue;
     }
     const workspaceRoot = path.resolve(record.cwd);
+    const repositoryKey = fusionRepositoryKey(workspaceRoot);
     const available = hasUsageObject(record);
     const observedAt = now();
     const appended = appendTokenUsageObservation(tokenUsageSidecarPath(workspaceRoot, env), {
@@ -127,6 +128,7 @@ export function observeGrokJobs({ env = process.env, now = () => new Date().toIS
       jobId: record.id,
       engine: "grok",
       workspaceRoot,
+      repositoryKey,
       availability: available ? "available" : "unavailable",
       tokenUsage: available ? grokTokenUsageObservation(record.usage) : null,
       source: "grok-job-record",
@@ -135,6 +137,21 @@ export function observeGrokJobs({ env = process.env, now = () => new Date().toIS
     if (!appended) {
       continue;
     }
+    const resolvedModel = typeof record.resolvedModel === "string" && record.resolvedModel.trim() ? record.resolvedModel.trim() : null;
+    if (resolvedModel) {
+      const resolvedEffort = typeof record.resolvedEffort === "string" && record.resolvedEffort.trim() ? record.resolvedEffort.trim() : null;
+      appendModelAuditObservation(modelAuditSidecarPath(workspaceRoot, env), {
+        schemaVersion: 1,
+        jobId: record.id,
+        engine: "grok",
+        model: resolvedModel,
+        effort: resolvedEffort,
+        workspaceRoot,
+        repositoryKey,
+        source: "grok-job-record",
+        observedAt
+      });
+    }
     observations += 1;
     if (available) {
       const becameObserved = !observedJobIds.has(record.id);
diff --git a/plugins/fusion/scripts/inline-delegation-guard.mjs b/plugins/fusion/scripts/inline-delegation-guard.mjs
index cf95e89..99a8cac 100644
--- a/plugins/fusion/scripts/inline-delegation-guard.mjs
+++ b/plugins/fusion/scripts/inline-delegation-guard.mjs
@@ -56,6 +56,7 @@ const NO_OP_HEARTBEAT_REASON = "Fusion tasks are in flight for this session, so
 const REAPED_WORKER_REDIRECT_AUDIT_DESCRIPTION = "reaped-worker-redirect";
 const NARROW_WAVE_ADVISORY_SUFFIX = "consecutive narrow waves; fleet default applies: consider /fusion:ultra for the remaining packages or state fleet-decline: <reason>.";
 const NARROW_WAVE_ADVISORY_REASON = "narrow-wave-advisory";
+const TRANSCRIPT_TAIL_MAX_BYTES = 262144;
 const MISSING_LAUNCH_RECOVERY_REASON = "missing-launch-recovered";
 const TAIL_ALLOW_AUDIT_EVENT = "tail-allowed";
 const ZERO_DISPATCH_SOFTENED_AUDIT_EVENT = "zero-dispatch-softened";
@@ -111,6 +112,88 @@ function resolveMode(env = process.env) {
   return String(env[MODE_ENV] ?? "").trim().toLowerCase() === "advisory" ? "advisory" : "enforce";
 }
 
+function transcriptEntryContent(entry) {
+  if (Array.isArray(entry?.message?.content)) {
+    return entry.message.content;
+  }
+  return Array.isArray(entry?.content) ? entry.content : [];
+}
+
+function isExternalUserMessage(entry) {
+  if (entry?.type !== "user" || entry.toolUseResult != null || entry.sourceToolAssistantUUID || entry.isMeta === true || entry?.message?.isMeta === true) {
+    return false;
+  }
+  const content = entry?.message?.content ?? entry?.content;
+  if (typeof content === "string") {
+    return true;
+  }
+  return (
+    Array.isArray(content) &&
+    content.length > 0 &&
+    content.every((block) => block?.type === "text" && typeof block.text === "string")
+  );
+}
+
+function hasFleetDeclineAfterLastUserMessage(transcriptPath) {
+  if (typeof transcriptPath !== "string" || !transcriptPath) {
+    return false;
+  }
+  let descriptor;
+  try {
+    descriptor = fs.openSync(transcriptPath, "r");
+    const stat = fs.fstatSync(descriptor);
+    if (!stat.isFile() || stat.size === 0) {
+      return false;
+    }
+    const bytesToRead = Math.min(TRANSCRIPT_TAIL_MAX_BYTES, stat.size);
+    const start = stat.size - bytesToRead;
+    const buffer = Buffer.allocUnsafe(bytesToRead);
+    const bytesRead = fs.readSync(descriptor, buffer, 0, buffer.length, start);
+    let text = buffer.subarray(0, bytesRead).toString("utf8");
+    if (start > 0) {
+      const firstNewline = text.indexOf("\n");
+      text = firstNewline === -1 ? "" : text.slice(firstNewline + 1);
+    }
+    const entries = [];
+    for (const line of text.split("\n")) {
+      if (!line.trim()) {
+        continue;
+      }
+      try {
+        entries.push(JSON.parse(line));
+      } catch {
+        void 0;
+      }
+    }
+    const lastUserIndex = entries.findLastIndex(isExternalUserMessage);
+    if (lastUserIndex === -1) {
+      return false;
+    }
+    const assistantText = [];
+    for (const entry of entries.slice(lastUserIndex + 1)) {
+      if (entry?.type !== "assistant") {
+        continue;
+      }
+      for (const block of transcriptEntryContent(entry)) {
+        if (block?.type === "text" && typeof block.text === "string") {
+          assistantText.push(block.text);
+        }
+      }
+    }
+    return /^\s*fleet-decline:/mi.test(assistantText.join("\n"));
+  } catch {
+    return false;
+  } finally {
+    if (descriptor != null) {
+      try {
+        fs.closeSync(descriptor);
+      } catch {
+        void 0;
+      }
+    }
+  }
+}
+
 function resolveAuditDir(env = process.env) {
   const override = env[AUDIT_DIR_ENV];
   if (override && String(override).trim()) {
@@ -566,6 +649,7 @@ function normalizeAuditEvent(event) {
           lane,
           tool: event.tool,
           reason: event.reason,
+          ...(event.reason === NARROW_WAVE_ADVISORY_REASON && event.declineStated === true ? { declineStated: true } : {}),
           ...(event.suppressed === true ? { suppressed: true } : {})
         }
       : null;
@@ -1220,6 +1304,7 @@ function runHook(env = process.env, input = readHookInput()) {
       env
     );
     if (confirmation.shouldAdvise) {
+      const declineStated = hasFleetDeclineAfterLastUserMessage(input.transcript_path);
       const advisory = claimAdvisory(
         file,
         "narrow-wave",
@@ -1237,11 +1322,12 @@ function runHook(env = process.env, input = readHookInput()) {
             lane: confirmation.lane,
             tool: toolName,
             reason: NARROW_WAVE_ADVISORY_REASON,
+            ...(declineStated ? { declineStated: true } : {}),
             ...(advisory.suppressed ? { suppressed: true } : {})
           },
           env
         );
-        if (!advisory.suppressed) {
+        if (!advisory.suppressed && !declineStated) {
           process.stdout.write(`${JSON.stringify(postToolAdvisoryOutput(`${confirmation.streak} ${NARROW_WAVE_ADVISORY_SUFFIX}`))}\n`);
         }
       }
diff --git a/tests/grok-jobs-observer.test.mjs b/tests/grok-jobs-observer.test.mjs
index 264e0b5..7cb9511 100644
--- a/tests/grok-jobs-observer.test.mjs
+++ b/tests/grok-jobs-observer.test.mjs
@@ -7,7 +7,7 @@ import path from "node:path";
 import { fileURLToPath } from "node:url";
 import { test } from "node:test";
 
-import { tokenUsageSidecarPath } from "../plugins/fusion/scripts/fusion-stats.mjs";
+import { fusionRepositoryKey, modelAuditSidecarPath, tokenUsageSidecarPath } from "../plugins/fusion/scripts/fusion-stats.mjs";
 import { grokJobsObserverStatePath, observeGrokJobs } from "../plugins/fusion/scripts/grok-jobs-observer.mjs";
 
 const observerScript = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..", "plugins", "fusion", "scripts", "grok-jobs-observer.mjs");
@@ -46,6 +46,8 @@ test("reobserves unavailable terminal jobs until usage becomes available", (t) =
   const env = { GROK_COMPANION_DATA: grokData, FUSION_DATA_DIR: fusionData };
   writeGrokJob(grokData, workspaceRoot, "grok-available", {
     status: "done",
+    resolvedModel: "grok-4",
+    resolvedEffort: "high",
     usage: {
       input_tokens: 120,
       cache_read_input_tokens: 40,
@@ -54,13 +56,20 @@ test("reobserves unavailable terminal jobs until usage becomes available", (t) =
       total_tokens: 150
     }
   });
-  writeGrokJob(grokData, workspaceRoot, "grok-unavailable", { status: "error" });
+  writeGrokJob(grokData, workspaceRoot, "grok-unavailable", {
+    status: "error",
+    resolvedModel: "grok-4",
+    resolvedEffort: null
+  });
   writeGrokJob(grokData, workspaceRoot, "grok-running", {
     status: "running",
+    resolvedModel: "grok-4",
+    resolvedEffort: "low",
     usage: { input_tokens: 1, cache_read_input_tokens: 0, output_tokens: 1, reasoning_tokens: 0, total_tokens: 2 }
   });
 
   assert.strictEqual(observeGrokJobs({ env, now: () => "2026-07-22T00:00:00.000Z" }), 2);
+  const repositoryKey = fusionRepositoryKey(workspaceRoot);
   const observations = readJsonLines(tokenUsageSidecarPath(workspaceRoot, env));
   assert.deepStrictEqual(observations, [
     {
@@ -68,6 +77,7 @@ test("reobserves unavailable terminal jobs until usage becomes available", (t) =
       jobId: "grok-available",
       engine: "grok",
       workspaceRoot,
+      repositoryKey,
       availability: "available",
       tokenUsage: { inputTokens: 120, cachedInputTokens: 40, outputTokens: 30, reasoningOutputTokens: 12, totalTokens: 150 },
       source: "grok-job-record",
@@ -78,21 +88,51 @@ test("reobserves unavailable terminal jobs until usage becomes available", (t) =
       jobId: "grok-unavailable",
       engine: "grok",
       workspaceRoot,
+      repositoryKey,
       availability: "unavailable",
       tokenUsage: null,
       source: "grok-job-record",
       observedAt: "2026-07-22T00:00:00.000Z"
     }
   ]);
+  const modelAuditPath = modelAuditSidecarPath(workspaceRoot, env);
+  const modelAudits = readJsonLines(modelAuditPath);
+  assert.deepStrictEqual(modelAudits, [
+    {
+      schemaVersion: 1,
+      jobId: "grok-available",
+      engine: "grok",
+      model: "grok-4",
+      effort: "high",
+      workspaceRoot,
+      repositoryKey,
+      source: "grok-job-record",
+      observedAt: "2026-07-22T00:00:00.000Z"
+    },
+    {
+      schemaVersion: 1,
+      jobId: "grok-unavailable",
+      engine: "grok",
+      model: "grok-4",
+      effort: null,
+      workspaceRoot,
+      repositoryKey,
+      source: "grok-job-record",
+      observedAt: "2026-07-22T00:00:00.000Z"
+    }
+  ]);
   const observerState = JSON.parse(fs.readFileSync(grokJobsObserverStatePath(env), "utf8"));
   assert.strictEqual(observerState.schemaVersion, 2);
   assert.deepStrictEqual(observerState.observedJobIds, ["grok-available"]);
   assert.deepStrictEqual(observerState.unavailableObservedAt, { "grok-unavailable": "2026-07-22T00:00:00.000Z" });
   assert.strictEqual(observeGrokJobs({ env, now: () => "2026-07-22T00:01:00.000Z" }), 1);
   assert.deepStrictEqual(readJsonLines(tokenUsageSidecarPath(workspaceRoot, env)), observations);
+  assert.deepStrictEqual(readJsonLines(modelAuditPath), modelAudits);
 
   writeGrokJob(grokData, workspaceRoot, "grok-unavailable", {
     status: "error",
+    resolvedModel: "grok-4",
+    resolvedEffort: null,
     usage: { input_tokens: 90, cache_read_input_tokens: 20, output_tokens: 10, reasoning_tokens: 4, total_tokens: 120 }
   });
   assert.strictEqual(observeGrokJobs({ env, now: () => "2026-07-22T00:02:00.000Z" }), 1);
@@ -103,14 +143,17 @@ test("reobserves unavailable terminal jobs until usage becomes available", (t) =
       jobId: "grok-unavailable",
       engine: "grok",
       workspaceRoot,
+      repositoryKey,
       availability: "available",
       tokenUsage: { inputTokens: 90, cachedInputTokens: 20, outputTokens: 10, reasoningOutputTokens: 4, totalTokens: 120 },
       source: "grok-job-record",
       observedAt: "2026-07-22T00:02:00.000Z"
     }
   ]);
+  assert.deepStrictEqual(readJsonLines(modelAuditPath), modelAudits);
   assert.deepStrictEqual(JSON.parse(fs.readFileSync(grokJobsObserverStatePath(env), "utf8")).observedJobIds, ["grok-available", "grok-unavailable"]);
   assert.strictEqual(observeGrokJobs({ env, now: () => "2026-07-22T00:03:00.000Z" }), 0);
+  assert.deepStrictEqual(readJsonLines(modelAuditPath), modelAudits);
 });
 
 test("marks an unavailable terminal job only after its observation TTL", (t) => {
diff --git a/tests/inline-delegation-guard.test.mjs b/tests/inline-delegation-guard.test.mjs
index 0f98484..84601ac 100644
--- a/tests/inline-delegation-guard.test.mjs
+++ b/tests/inline-delegation-guard.test.mjs
@@ -150,6 +150,16 @@ function completeActiveWave(sandbox, gapMs) {
   fs.writeFileSync(file, JSON.stringify(state), "utf8");
 }
 
+function triggerNarrowWaveAdvisory(sandbox, extraEnv = {}, payload = dispatchPayload(sandbox)) {
+  const gapMs = Number(extraEnv.FUSION_FLEET_WAVE_GAP_MS);
+  assert.ok(Number.isFinite(gapMs) && gapMs > 0);
+  assert.strictEqual(run(sandbox, dispatchPayload(sandbox), extraEnv).stdout, "");
+  completeActiveWave(sandbox, gapMs);
+  assert.strictEqual(run(sandbox, dispatchPayload(sandbox), extraEnv).stdout, "");
+  completeActiveWave(sandbox, gapMs);
+  return run(sandbox, payload, extraEnv);
+}
+
 function auditFiles(sandbox) {
   if (!fs.existsSync(sandbox.auditDir)) {
     return [];
@@ -557,6 +567,62 @@ test("two narrow dispatch waves attach the fleet advisory to the following dispa
   assert.strictEqual(readState(sandbox, "session-1").consecutiveNarrowWaves, 2);
 });
 
+test("a fleet decline after the last user message suppresses the narrow-wave context and remains audited", (t) => {
+  const sandbox = makeSandbox(t);
+  const transcript = path.join(sandbox.root, "transcript.jsonl");
+  fs.writeFileSync(
+    transcript,
+    `${[
+      { type: "user", message: { content: "continue with the package" } },
+      { type: "assistant", message: { content: [{ type: "text", text: "fleet-decline: codex single flight" }] } }
+    ]
+      .map((entry) => JSON.stringify(entry))
+      .join("\n")}\n`,
+    "utf8"
+  );
+
+  const result = triggerNarrowWaveAdvisory(sandbox, { FUSION_FLEET_WAVE_GAP_MS: "10000" });
+  assert.strictEqual(result.stdout, "");
+  assert.strictEqual(readState(sandbox, "session-1").lastAdvisedNarrowWaveStreak, 2);
+  const warnings = readAuditRecords(sandbox).filter((record) => record.event === "warn" && record.reason === "narrow-wave-advisory");
+  assert.strictEqual(warnings.length, 1);
+  assert.strictEqual(warnings[0].declineStated, true);
+});
+
+test("a fleet decline before the last user message does not suppress the narrow-wave context", (t) => {
+  const sandbox = makeSandbox(t);
+  const transcript = path.join(sandbox.root, "transcript.jsonl");
+  fs.writeFileSync(
+    transcript,
+    `${[
+      { type: "user", message: { content: "continue with the package" } },
+      { type: "assistant", message: { content: [{ type: "text", text: "fleet-decline: codex single flight" }] } },
+      { type: "user", message: { content: "continue with the next package" } }
+    ]
+      .map((entry) => JSON.stringify(entry))
+      .join("\n")}\n`,
+    "utf8"
+  );
+
+  const result = triggerNarrowWaveAdvisory(sandbox, { FUSION_FLEET_WAVE_GAP_MS: "10000" });
+  assert.strictEqual(
+    JSON.parse(result.stdout).hookSpecificOutput.additionalContext,
+    "2 consecutive narrow waves; fleet default applies: consider /fusion:ultra for the remaining packages or state fleet-decline: <reason>."
+  );
+});
+
+test("a missing transcript does not suppress the narrow-wave context", (t) => {
+  const sandbox = makeSandbox(t);
+  const payload = dispatchPayload(sandbox);
+  payload.transcript_path = path.join(sandbox.root, "missing-transcript.jsonl");
+
+  const result = triggerNarrowWaveAdvisory(sandbox, { FUSION_FLEET_WAVE_GAP_MS: "10000" }, payload);
+  assert.strictEqual(
+    JSON.parse(result.stdout).hookSpecificOutput.additionalContext,
+    "2 consecutive narrow waves; fleet default applies: consider /fusion:ultra for the remaining packages or state fleet-decline: <reason>."
+  );
+});
+
 test("a wide dispatch wave resets the narrow-wave advisory run", (t) => {
   const sandbox = makeSandbox(t);
   const gapMs = 10000;

From 3961220f52d8eb57262b0f8d47c655fe1f53da1f Mon Sep 17 00:00:00 2001
From: Harry Yep <git@hode.co.uk>
Date: Mon, 27 Jul 2026 13:05:05 +0800
Subject: [PATCH 5/7] test: hermetic breaker check suite

---
 tests/breaker-check.test.mjs | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/tests/breaker-check.test.mjs b/tests/breaker-check.test.mjs
index e591f4b..8ada4ba 100644
--- a/tests/breaker-check.test.mjs
+++ b/tests/breaker-check.test.mjs
@@ -14,6 +14,9 @@ function makeSandbox(t) {
   t.after(() => fs.rmSync(root, { recursive: true, force: true }));
   return {
     root,
+    home: path.join(root, "home"),
+    workerState: path.join(root, "worker-state"),
+    dataDir: path.join(root, "fusion-data"),
     grokData: path.join(root, "grok-data"),
     codexState: path.join(root, "codex-state")
   };
@@ -29,10 +32,19 @@ function writeRecord(file, record) {
 }
 
 function envFor(sandbox, extra = {}) {
+  const inherited = Object.fromEntries(
+    Object.entries(process.env).filter(
+      ([key]) => key !== "HOME" && key !== "CLAUDE_CODE_SESSION_ID" && !/^(?:FUSION|GROK|CODEX)_/.test(key)
+    )
+  );
   return {
-    ...process.env,
-    GROK_COMPANION_DATA: sandbox.grokData,
+    ...inherited,
+    HOME: sandbox.home,
+    FUSION_DATA_DIR: sandbox.dataDir,
+    FUSION_WORKER_STATE_DIR: sandbox.workerState,
+    FUSION_CODEX_STATE: sandbox.codexState,
     FUSION_CODEX_STATE_DIR: sandbox.codexState,
+    GROK_COMPANION_DATA: sandbox.grokData,
     ...extra
   };
 }

From 72c5155a9b16d12b0891559753bd4bd0442d141b Mon Sep 17 00:00:00 2001
From: Harry Yep <git@hode.co.uk>
Date: Mon, 27 Jul 2026 13:05:05 +0800
Subject: [PATCH 6/7] docs: claude worker briefs are envelope first

---
 plugins/fusion/rules-manifest.json    | 1 +
 plugins/fusion/rules/orchestration.md | 2 +-
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/plugins/fusion/rules-manifest.json b/plugins/fusion/rules-manifest.json
index e94e942..4f7f810 100644
--- a/plugins/fusion/rules-manifest.json
+++ b/plugins/fusion/rules-manifest.json
@@ -26,6 +26,7 @@
     "b3649bf8c8fb1f3ede0ccfc7fb727a8adf1e13628a18f4a84d08efbee75bd934",
     "b5283c0356c9004385157f3360ca81a3601253e370216b390049d069a29a2250",
     "c435cc1486eeff00f5b3a1c76017d46d8f05d30ebd248435d575a1461d7da5d4",
+    "c5cb1e9d17cd2771e0f320e760a249188d69658ef3a63b31556000a64fed88e1",
     "c68785a9d73cc2cb83b740db7a80744ddd293e6da011c6a80a17b045f9219e17",
     "dd2c1475815da92b62b28ab9cf6599df7ccab7fa88617e367e8343a6ed515434",
     "e1feb3d17af193a1b8b34b090d4d504c594efb87bbe521ed20f948aba4567e8a",
diff --git a/plugins/fusion/rules/orchestration.md b/plugins/fusion/rules/orchestration.md
index 511edcd..c7f99e7 100644
--- a/plugins/fusion/rules/orchestration.md
+++ b/plugins/fusion/rules/orchestration.md
@@ -169,7 +169,7 @@ Every brief is self contained: goal, constraints, relevant paths, and what done
 - Parallel packages declare each other: every brief names sibling packages' files as intended in flight changes and forbids reverting, restoring, or cleaning anything outside its own list. A worker's end state check covers its own files only.
 - A worktree isolated peer package is collected by merging its diff back into the main tree and re-running its verification command there before the package counts as done; for Codex, record semantic acceptance only after that main tree verification. The orchestrator creates a real Git worktree first and launches the peer with that canonical directory as its actual process cwd. Grok headless `--worktree` and `--worktree-ref` are not substitutes because the current headless path does not reliably create, select, or forward those worktree settings. Worktree runs leave disposable per-path state directories in the engine's data root, and /fusion:stats --prune-dead is the cleanup path once the worktree itself is gone. After each release, `/fusion:stats --prune-dead` runs as part of the release cadence. Worktrees hosting detached peer jobs are created and removed by the orchestrator with git worktree add and git worktree remove; never tie such a worktree to a wrapper agent's lifetime, because harness managed agent worktrees are auto removed when the wrapper agent exits, which deletes the workspace out from under a still running detached job.
 - State write permission explicitly in every peer brief. Every Grok mode uses strict. Its upstream `system_read` roots include `/var` and `/tmp` on Linux, and `/private` plus the entire `~/Library` on macOS, so shared temporary content and substantial user application configuration and caches may be readable. Its write roots include the workspace and entire Grok home, always include `/tmp` and `/var/tmp`, and on macOS also include `/private/tmp`, `/private/var/tmp`, and all of `/private/var/folders`. The run's unique private `TMPDIR` is the adapter's preferred and handshake-verified temporary path, not the only writable temporary surface. User toolchains outside the readable roots may therefore be unavailable, and their absence never authorizes a workspace downgrade. Strict requests child process network restriction. Linux enforces it through seccomp, while macOS network blocking is a no-op and a write shell can still reach the network; `--web` controls only the built in web tools. Grok consult briefs and /grok:review add a hard read only tool set limited to file read, list, and search, plus web search and fetch when `--web` is present. They cannot invoke shell commands, tests, git, builds, edits, MCP tools, or subagents through model tools. The sandbox still permits consult `read_file` to reach `~/.grok/auth.json`, config, and sessions because strict exposes the entire Grok home, and write terminal commands can receive preserved xAI authentication variables. The companion adds best-effort Read denies for `auth.json`, `mcp_credentials.json`, `config.toml`, `sessions`, `memory`, `logs`, and `debug` through absolute paths and standard `**/.grok` patterns, but raw path variants, symbolic links, and shell or indirect scripts can bypass those tool-level rules. These Read denies govern model tool calls and do not stop upstream's internal first-turn injection for a direct task with explicit `--memory`; injected memory enters the model context sent to xAI. Environment scrubbing and path-pattern denies do not isolate credentials or memory content. Model-facing meta-tool denies do not prove that native MCP servers, plugins, or hooks under `~/.grok` did not start during agent construction or had no side effects; those bridge variables disable compatibility imports, not native Grok configuration. Narrower exposure requires a dedicated sandbox profile, an isolated Grok home, an authentication broker, or upstream path-level authorization. Only briefs asking for repository changes receive the strict write tool set. Before prompt delivery, the companion requires a new `ProfileApplied` event in the incremental shared `sandbox-events.jsonl` delta that contains this run's private temporary path and matches the canonical workspace, strict profile, enforced state, and requested `restrict_network: true` configuration. Upstream events have no run id or pid, so the unique private path correlates positive evidence; unrelated `ProfileApplied` and `ApplyFailed` records are not attributed, and failure records are only auxiliary diagnostics. The matching event proves profile and configuration application, not actual platform network isolation. A matching warning on the owned stderr stream, handshake timeout, missing or malformed matching evidence, shared-log disappearance or rotation, or any matching field mismatch terminates the process and fails closed; stderr cannot prove successful enforcement.
-- Every peer brief opens with a single header line that names the lane, the effort or model tier whenever the routing row requires one, and the verification command for an implementation or change brief, or explicit coverage or acceptance criteria for a consult, research, or review brief; every automatically routed Grok header also includes `grok-role: <role>`. Pass the brief directly in the Agent prompt. Never create a transport prompt file with Write or place a brief under plugin data. The Grok companion must use `--prompt-file /dev/stdin` and pipe the prompt to the child because Grok applies the sandbox before reading a prompt file, which makes an external staged brief unreadable under strict mode. Companion adapters own any private staging required by their transports. Collection review verifies consult, research, and review criteria. A brief missing that header is not ready to dispatch. The header line is audit prose: engine model and effort selection happens only through companion options such as `--model` and `--effort` placed before the `--` separator in the direct prompt, and a routing row that names a model tier is satisfied only when the dispatch passes those options.
+- Every peer brief opens with a single header line that names the lane, the effort or model tier whenever the routing row requires one, and the verification command for an implementation or change brief, or explicit coverage or acceptance criteria for a consult, research, or review brief; every automatically routed Grok header also includes `grok-role: <role>`. Pass the brief directly in the Agent prompt. Never create a transport prompt file with Write or place a brief under plugin data. The Grok companion must use `--prompt-file /dev/stdin` and pipe the prompt to the child because Grok applies the sandbox before reading a prompt file, which makes an external staged brief unreadable under strict mode. Companion adapters own any private staging required by their transports. Collection review verifies consult, research, and review criteria. A brief missing that header is not ready to dispatch. Claude worker briefs are not peer briefs: their first line is always the `fusion-brief: v1` envelope, and lane or header prose rides inside the envelope after that line. The header line is audit prose: engine model and effort selection happens only through companion options such as `--model` and `--effort` placed before the `--` separator in the direct prompt, and a routing row that names a model tier is satisfied only when the dispatch passes those options.
 - Grok review requires the installed headless `--json-schema` contract and consumes `structuredOutput` from the same turn. A schema error or explicit null structured result is a failed review, not permission to trust text or spend a second call. Compatibility parsing of `text` is allowed only when the structured output field is completely absent. Preserve request id, session id, turn count, token usage, per-model usage, cost, partial-cost, and incomplete-usage indicators in job records whenever the CLI reports them. A `model_usage` row reports only input, output, cache-read, and model-call counts plus optional cost, so never synthesize aggregate reasoning or total token channels. `usage_is_incomplete` means the usage ledger may have missed open subagents, usage application, or a drain timeout; present values are observed lower bounds, and both job-total token and cost coverage fail closed to incomplete. Accept only positive complete top-level cost pairs and use ticks as authoritative at 10000000000 ticks per USD. These fields improve observability but never replace independent semantic acceptance.
 - A future Grok ACP adapter must use one process pool per canonical cwd and sandbox profile. Each daemon starts directly with `grok --cwd <canonical> --sandbox <profile> agent --no-leader stdio`; a global bridge or leader reuse is forbidden because sandbox installation happens once at process start and an existing leader can retain a different cwd or sandbox. ACP permission requests and live tool events improve visibility, but do not weaken the companion's tool, worktree, background, collection, or semantic acceptance boundaries.
 - Never delegate to codex or grok anything touching secrets, credentials, or context that cannot be compressed into a brief.

From 25e993d1ada5490dd17c5e8d754cfe715de4c124 Mon Sep 17 00:00:00 2001
From: Harry Yep <git@hode.co.uk>
Date: Mon, 27 Jul 2026 13:05:05 +0800
Subject: [PATCH 7/7] chore: release 0.0.41

---
 .claude-plugin/marketplace.json           |  8 ++++----
 CHANGELOG.md                              | 10 ++++++++++
 plugins/codex/.claude-plugin/plugin.json  |  2 +-
 plugins/fusion/.claude-plugin/plugin.json |  2 +-
 plugins/grok/.claude-plugin/plugin.json   |  2 +-
 5 files changed, 17 insertions(+), 7 deletions(-)

diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json
index e35bc6d..215c7cd 100644
--- a/.claude-plugin/marketplace.json
+++ b/.claude-plugin/marketplace.json
@@ -6,14 +6,14 @@
     "email": "hi@okis.dev"
   },
   "description": "Multi-model orchestration marketplace for Claude Code.",
-  "version": "0.0.40",
+  "version": "0.0.41",
   "plugins": [
     {
       "name": "grok",
       "source": "./plugins/grok",
       "displayName": "Grok Companion",
       "description": "Local Grok CLI delegation: task, review, resumable history, best-of-n tournaments, background jobs, stats, and setup health checks.",
-      "version": "0.0.40",
+      "version": "0.0.41",
       "author": {
         "name": "Harry Yep"
       },
@@ -34,7 +34,7 @@
       "source": "./plugins/codex",
       "displayName": "Codex Companion",
       "description": "First party local Codex CLI delegation for tasks, reviews, resumable threads, and durable background jobs.",
-      "version": "0.0.40",
+      "version": "0.0.41",
       "author": {
         "name": "Harry Yep"
       },
@@ -55,7 +55,7 @@
       "source": "./plugins/fusion",
       "displayName": "Fusion Orchestrator",
       "description": "Multi-model orchestration: tier agents, routing rules, blind panel, ultra fleet, model config, and drift doctor.",
-      "version": "0.0.40",
+      "version": "0.0.41",
       "author": {
         "name": "Harry Yep"
       },
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 97529bb..2a8cb86 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,15 @@
 # changelog
 
+## 0.0.41
+
+- the worker token budget stops eating deliverables: a one shot wind down lands at 85 percent of the output token budget telling the worker to stop calling tools and write the deliverable, and after the limit trips exactly one final `Write` is still permitted (`terminalWriteGraceUsedAt`) while every other tool stays denied with the deny reason naming the grace; the subagent start announcement and all three worker prompts state both (observed live: three of four mining workers lost their final report write to the old hard deny)
+- settlement grammar sheds its two most repeated stumbles: a bare `/fusion:stats` invocation now renders the plain report because the empty argument list joins the value free direct family, and a single `--record` flag accepts multiple `<id>=<verdict>` pairs identically in the strict direct family and the raw args transport parser (seven contract errors across four sessions in one mining window motivated both)
+- codex telemetry gets discriminating about time: per SKU trends add a near cap share (timeout deaths plus finishes within 30 seconds of the 570 second deadline), terminal ledger records carry `timeoutMs` and `failureKind` so ledger recovered timeouts stop understating `timeoutShare`, and peer wrapper ledger records derive `packageType` from an explicit `package-type:` line, then a `--write` or `--transport-default-write` token before the `--` separator, then the verification heuristic, instead of blanket `consult` (120 of 131 live jobs were mislabeled)
+- grok reaches observability parity: the jobs observer emits `model-audit.jsonl` rows for terminal grok jobs (`appendModelAuditObservation` moves to `fusion-stats.mjs` exports) and its token rows gain `repositoryKey`; `--prune-dead` also sweeps orphaned `acceptance.jsonl` sidecars, dead since the 0.0.27 settlement cutover bypassed their writer and 0.0.35 deleted it
+- the codex companion survives its two observed run killers: a job whose cwd resolves inside a repository subdirectory without an explicit `--cwd` gets a warning diagnostic plus stderr line instead of silently sandboxing to the subtree (two zero delivery rejections came from a leaked shell cwd), and an oversized JSONL event is skipped with one diagnostic while the run continues, failing as `resource` only when the protocol cannot complete after a skip, which also removes the broken pipe exit 143 race; both contract docs restate the behavior and the stale `--record-acceptance` reference is gone
+- the narrow wave advisory becomes decline aware: when the transcript tail shows a `fleet-decline:` line after the last user message the stdout nudge is suppressed while the audit warn records `declineStated: true` (19 of 27 live advisories were firing at sessions with stated declines); the stop gate rereads cancellation records from disk before demanding `TaskStop` so a worker that turned terminal after the snapshot is no longer demanded, and the task id splice tolerates leading blank lines in an envelope brief
+- breaker-check tests are hermetic: the suite builds a sandbox home with stripped `FUSION_`, `GROK_`, and `CODEX_` inheritance so live lane failures can no longer redden a local run, and the orchestration rules state that claude worker briefs are envelope first (`fusion-brief: v1` on line one, header prose after) with the rules manifest regenerated
+
 ## 0.0.40
 
 - routing becomes capability first: a package routes by its load bearing capability instead of its surface task type, capability monopolies hard route ahead of every task type default (X and live platform reads to grok `live-web` even when digest shaped, Claude tool surface and privacy to Claude workers, judgment dense authorship to the main session), and lane defaults are named as cached capability observations that fresh observations override per package; grounded in a same day controlled probe where grok `web_fetch` returned a full x.com post body while the codex web surface returned empty HTML for the same URL
diff --git a/plugins/codex/.claude-plugin/plugin.json b/plugins/codex/.claude-plugin/plugin.json
index eabe52d..b9f53b0 100644
--- a/plugins/codex/.claude-plugin/plugin.json
+++ b/plugins/codex/.claude-plugin/plugin.json
@@ -2,7 +2,7 @@
   "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json",
   "name": "codex",
   "displayName": "Codex Companion",
-  "version": "0.0.40",
+  "version": "0.0.41",
   "description": "First party local Codex CLI delegation for tasks, reviews, resumable threads, and durable background jobs.",
   "author": {
     "name": "Harry Yep"
diff --git a/plugins/fusion/.claude-plugin/plugin.json b/plugins/fusion/.claude-plugin/plugin.json
index 7515772..baead71 100644
--- a/plugins/fusion/.claude-plugin/plugin.json
+++ b/plugins/fusion/.claude-plugin/plugin.json
@@ -2,7 +2,7 @@
   "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json",
   "name": "fusion",
   "displayName": "Fusion Orchestrator",
-  "version": "0.0.40",
+  "version": "0.0.41",
   "description": "Multi-model orchestration: tier agents, routing rules, blind panel, ultra fleet, model config, and drift doctor.",
   "author": {
     "name": "Harry Yep"
diff --git a/plugins/grok/.claude-plugin/plugin.json b/plugins/grok/.claude-plugin/plugin.json
index 05dee52..20eb971 100644
--- a/plugins/grok/.claude-plugin/plugin.json
+++ b/plugins/grok/.claude-plugin/plugin.json
@@ -2,7 +2,7 @@
   "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json",
   "name": "grok",
   "displayName": "Grok Companion",
-  "version": "0.0.40",
+  "version": "0.0.41",
   "description": "Local Grok CLI delegation: task, review, resumable history, background jobs, stats, and setup health checks.",
   "author": {
     "name": "Harry Yep"