fix(harness): mark model activity before final answers (#797)

BUILD-LOG.md

@@ -6193,3 +6193,12 @@ Before: clean isolated checkout at e9271841f3af23bf2a72aa17d9d67a1ec5abe017; no
 After: personally authored three-file skills/ms-ux-improvement coordinator package, complementary to existing frontend execution guidance; final candidate-02 content manifest b93d7836ed7c70db62386192d6e1db5c4f867f841aaa7645c3c861def5a9b140. Retained 19 bounded clean-context behavioral samples / 186 raw files: six unguided baselines, seven candidate-01 samples including ready-review control, six candidate-02 samples. Candidate-01's two unnecessary frozen-method recovery conditions and already-correct control remain recorded; narrowed wording and final tests show no extra gate, without causal/statistical improvement claim. Fresh package-focused skill-creator/resource validation passed for workspace, frozen snapshot and scratch copy; patch/whitespace, arithmetic, integrity, unchanged baseline and append-only ownership checks passed. Entry 637 words (2,210 total) and one-model synthetic/procedural-scope limits retained as author concerns. Exact source, patch, hashes, commands, raw evidence and independent-review brief: /mnt/storage/scratch/mosaic-lead/skill-ux-improvement/author/run-20261105T022125Z/HANDOFF.md. Author task complete; independent source approval pending. No application suites, commits, forge/queue writes, publication, installation, runtime activation or protected actions.
 
 2026-11-04 | #818 / Gitea #1845 canonical skill integration verified | Exact3 reviewed Markdown files match source hashes; canonical skill-creator validator and whitespace checks pass. Independent review APPROVE/no findings,14 focused checks and a fresh bounded forward-use handoff retained. All prior canonical and author log bytes preserved. Package is available in source by explicit path; no managed/personal installation, activation, publication or user-acceptance claim. MVP activation remains pinned to parentbd8120318, independent of this skill commit.
+
+
+## 2026-11-05 — MVP 3 coder first-activity watchdog correction (#797), implementation handoff
+
+Before: isolated branch fix/mvp3-coder-stall-797 starts at 05fdfef9ea1fdddf58c725c04f229ca8559fc054. Claude text-mode stdout appeared only at the final answer, so the 120-second first-output watchdog could stop productive thinking/tool work (D2 in planning-gate-fix-r1-report.md). Explicit child charter authorizes a local implementation/test/commit only; queue next codex refused this linked worktree, with no workaround or queue mutation.
+
+After: Claude's private stream-json reader marks real upstream assistant events and emits only the successful final text, matching real CLI newline semantics byte for byte. Pi's existing extension also marks model text/thinking/tool deltas; mock remains a test-only immediate echo. Both reuse the existing exclusive 0600 proxy-first-output marker with fixed output\n content; init/status, synthetic CLI errors, tool results, stderr and workspace writes do not count. Runner supplies only its existing run directory; helper gets a read-only harness mount and bundle source digest. No timeout, schema, role authority, credential, proxy management or live state changed. Silent runs still stop with proxy-stall; interrupted first/resumed turns emit no partial answer or successful new session ID.
+
+Verification: Node v24.22.1 with PATH pinned to /mnt/storage/scratch/tmp/z62zsJ/node24/node_modules/node-linux-x64/bin and fresh TMPDIR from mktemp -d /mnt/storage/scratch/tmp/XXXXXX. Final full 28-file command: 376 passed, 0 failed/cancelled/skipped/todo, test-concurrency=4 (coder-stall-final-r2.tap). Includes full launcher/model-stall, harness runner/bundle, seat session/sandbox, real Claude/Pi adapter, confinement, installer/copy, source-digest and import-closure checks. Delayed tool/thinking/header/assistant fixtures use 0.8-second injected watchdog and 1.4-second pre-answer work. Initial tests retained: missing test import and failed-result exit handling, real text-mode trailing-newline difference, then a combined-gate grandchild-reaping assertion race; all corrected and full final gate green. Whitespace and baseline log-prefix preservation checks pass. Report, exact commands, SHA256 manifest, scans and all TAP receipts: /mnt/storage/scratch/mosaic-lead/mvp3/coder-stall-fix-r1-report.md and coder-stall-* siblings. Local commit follows these green suites. Independent exact-commit review, CI/integration and deployment/live GLM acceptance remain with the lead; no push, stash, fleet operation or external message.

adapters/README.md

@@ -76,6 +76,7 @@ session's workspace and its own process group. Same contract, plus:
 | `MOSAIC_POLICY_FILE`, `MOSAIC_TOOLS_FILE` | The bundle's gate policy and typed tools. |
 | `MOSAIC_TOOL_SOCKET` | The runner's tool socket, which the typed tools call. |
 | `MOSAIC_TURN_MARKER` | `pi`: the extension writes it at `agent_end`; a pi turn that exits 0 without it failed. |
+| `MOSAIC_MODEL_ACTIVITY_DIR` | Managed runner's existing writable run directory. Trusted model streaming events publish the fixed `output\n` token to `proxy-first-output` (0600, exclusive create). No event contents or credentials enter it. Claude's private stream reader and Pi's extension signal thinking, tool-use and text activity before final stdout; init/status, tool results and workspace writes do not count. |
 | `MOSAIC_EXTENSIONS` | `pi` only: extension files loaded with `-e`; a missing one exits 2. The pi module names two: the gate and typed tools extension, then the unslop check's `tools/unslop-hook/extension.ts` (row 66). |
 | `MOSAIC_UNSLOP_LOG` | The unslop check's JSONL log, `<run>/unslop.jsonl`, which the runner sets for both adapters and reads after each turn (row 66). |
 | `MOSAIC_CLAUDE_SETTINGS`, `MOSAIC_CLAUDE_MCP_CONFIG` | `claude` only: the bundle's gate hook and MCP server; required. |
@@ -85,6 +86,14 @@ session's workspace and its own process group. Same contract, plus:
 The layers each adapter relies on, and what they don't cover, are in
 `packages/harness/README.md`.
 
+Claude uses `--output-format stream-json --verbose --include-partial-messages`.
+The adapter's `stream.mjs` consumes those events and prints only the successful
+final result, with the same newline as the CLI's text format. Malformed or
+incomplete streams fail the turn; a failed or interrupted first turn still
+does not persist a reusable session ID. The helper is mounted read-only and
+its digest is recorded in the launch bundle. Pi retains print mode and signals
+activity through its existing extension. Mock remains a test-only echo.
+
 ### A role's skills (row 64)
 
 A managed session's skills reach both adapters as text in the bundle's

adapters/claude/adapter.sh

@@ -11,6 +11,7 @@
 # wrapped as `timeout -k 2 10 <gate> || exit 2`, and the tool limit (--tools).
 # --bare is never passed: it turns settings hooks off.
 set -eu
+STREAM="$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)/stream.mjs"
 
 need() {
     eval "v=\${$1:-}"
@@ -60,7 +61,9 @@ fi
 
 # Flags:
 #   -p                       one-shot: print the answer and exit
-#   --output-format text     stdout is the answer only
+#   --output-format stream-json  private model events; stream.mjs emits only
+#   --verbose --include-partial-messages  the final result on stdout and marks
+#                            first model activity before long tool work
 #   --system-prompt-file     the generated prompt replaces the default, given
 #                            as a file so no local user reads it in
 #                            /proc/<pid>/cmdline (row 213); absolute, since
@@ -84,8 +87,8 @@ fi
 #                            text in the system prompt (row 64)
 ALLOWED=mcp__mosaic
 [ -z "${MOSAIC_TOOLS:-}" ] || ALLOWED="$MOSAIC_TOOLS,mcp__mosaic"
-"$CLAUDE" -p "$MOSAIC_REQUEST" \
-    --output-format text \
+node "$STREAM" "$CLAUDE" -p "$MOSAIC_REQUEST" \
+    --output-format stream-json --verbose --include-partial-messages \
     --system-prompt-file "$PROMPT_FILE" \
     --model "$MOSAIC_MODEL" \
     --effort "$MOSAIC_THINKING" \

adapters/claude/stream.mjs

@@ -0,0 +1,72 @@
+// Consume Claude Code's SDK event stream privately. Only the successful result
+// reaches stdout, ensuring a trailing newline like --output-format text.
+import { spawn } from 'node:child_process';
+import { firstOutputMarker } from '../../packages/harness/src/first-output.mjs';
+
+const mark = process.env.MOSAIC_MODEL_ACTIVITY_DIR
+  ? firstOutputMarker(process.env.MOSAIC_MODEL_ACTIVITY_DIR) : () => {};
+const [binary, ...args] = process.argv.slice(2);
+const child = spawn(binary, args, { stdio: ['ignore', 'pipe', 'inherit'] });
+let pending = '', answer, resultSeen = false, resultFailed = false, failed = false, killTimer;
+const fail = () => {
+  if (failed) return;
+  failed = true;
+  // Never put event bodies, provider errors or credentials in diagnostics.
+  process.stderr.write('claude adapter: invalid or incomplete model stream\n');
+  child.kill('SIGTERM');
+  killTimer = setTimeout(() => child.kill('SIGKILL'), 5000).unref();
+};
+function event(line) {
+  if (!line.trim()) return;
+  const item = JSON.parse(line);
+  const model = item.type === 'stream_event' ? item.event : null;
+  // Init/status, hooks, user/tool results and arbitrary local writes aren't
+  // model activity. message_start is the upstream assistant response header;
+  // partial blocks also cover models that first respond with thinking/tools.
+  const start = model?.type === 'message_start' && model.message?.role === 'assistant';
+  const block = model?.type === 'content_block_start'
+    && ['text', 'thinking', 'tool_use'].includes(model.content_block?.type);
+  const delta = model?.type === 'content_block_delta'
+    && ['text_delta', 'thinking_delta', 'input_json_delta'].includes(model.delta?.type);
+  const assistant = item.type === 'assistant' && !item.error
+    && item.message?.model !== '<synthetic>' && item.message?.role === 'assistant'
+    && item.message.content?.some(c => ['text', 'thinking', 'tool_use'].includes(c.type));
+  if (start || block || delta || assistant) mark('activity');
+  if (item.type === 'result') {
+    if (resultSeen) throw Error();
+    resultSeen = true;
+    // Let a normal CLI failure exit on its own, retaining its exit code.
+    resultFailed = item.subtype !== 'success' || !!item.is_error;
+    if (!resultFailed) {
+      if (typeof item.result !== 'string') throw Error();
+      answer = item.result;
+    }
+  }
+}
+child.stdout.setEncoding('utf8');
+child.stdout.on('data', chunk => {
+  if (failed) return;
+  try {
+    pending += chunk;
+    let end;
+    while ((end = pending.indexOf('\n')) !== -1) {
+      if (end > 16 * 1024 * 1024) throw Error();
+      event(pending.slice(0, end));
+      pending = pending.slice(end + 1);
+    }
+    if (pending.length > 16 * 1024 * 1024) throw Error();
+  } catch { fail(); }
+});
+child.on('error', fail);
+child.on('close', (code, signal) => {
+  clearTimeout(killTimer);
+  if (!failed && pending) {
+    try { event(pending); } catch { fail(); }
+  }
+  if (failed || resultFailed || code !== 0 || signal || answer === undefined) {
+    if (!failed && code === 0 && !signal) fail();
+    process.exitCode = code || 2;
+  } else {
+    process.stdout.write(answer.endsWith('\n') ? answer : answer + '\n');
+  }
+});

docs/SESSIONS.md

@@ -2064,3 +2064,5 @@ are never rewritten or removed; corrections are new entries.
 
 2026-11-05 | codex-ux-skill-author | Vikunja #818 UX improvement skill author handoff | Candidate-02 three-file package personally authored; package/copy/patch/arithmetic/integrity checks passed; 19 synthetic behavioral samples and all 186 raw artifacts retained with candidate-01 issue/correction and limits. Exact manifest b93d7836ed7c70db62386192d6e1db5c4f867f841aaa7645c3c861def5a9b140; evidence /mnt/storage/scratch/mosaic-lead/skill-ux-improvement/author/run-20261105T022125Z/HANDOFF.md. Independent source review pending; no commit, app suites, forge/queue writes, installation, activation or publication.
 2026-11-04 | Mos, skill integrator | #818 / Gitea #1845 | Integrated exact independently approved ms-ux-improvement candidate02 into source;3hashes/validator/whitespace pass and author histories preserved. Separate commit from MVP; its activation target staysbd8120318. No install, activation, publication or live action.
+
+2026-11-05 | Codex implementation child | MVP 3 #797 D2 coder-stall fix on fix/mvp3-coder-stall-797 from 05fdfef9e | Claude private streaming activity marker + answer-only stdout; Pi model-event coverage; silent/interrupted controls preserved. Final Node 24.22.1 full 28-file gate: 376 pass, 0 fail/skip/cancel. Queue refused linked worktree; no workaround. Local commit only, no push/stash/live actions. Report: /mnt/storage/scratch/mosaic-lead/mvp3/coder-stall-fix-r1-report.md; independent review remains with lead.

packages/harness/src/first-output.mjs

@@ -1,6 +1,7 @@
 import { writeFileSync, openSync, closeSync, linkSync, unlinkSync } from 'node:fs';
 import { join } from 'node:path';
-// The host needs an output-exists signal, never the output or a model credential.
+// The host needs a model-activity signal, never event/output text or a credential.
+// The historical filename/token stays stable for the host watchdog protocol.
 export function firstOutputMarker(runDir) {
   let seen = false;
   return chunk => {

packages/harness/src/pi-extension.mjs

@@ -11,6 +11,7 @@
 import { readFileSync, writeFileSync } from "node:fs";
 import { decide } from "./gate.mjs";
 import { callTool, replyText } from "./tools.mjs";
+import { firstOutputMarker } from "./first-output.mjs";
 
 function need(name) {
   const value = process.env[name];
@@ -25,6 +26,16 @@ export default function (pi) {
   const marker = need("MOSAIC_TURN_MARKER");
   if (policy.harness !== "pi" || !Array.isArray(policy.tools) || !Array.isArray(policy.typed)) throw new Error("mosaic extension: policy file is not a pi policy");
 
+  // Print mode emits only the last answer, after all tool work. Use model
+  // streaming events, never tool output or workspace changes, for liveness.
+  const markActivity = process.env.MOSAIC_MODEL_ACTIVITY_DIR
+    ? firstOutputMarker(process.env.MOSAIC_MODEL_ACTIVITY_DIR) : () => {};
+  pi.on("message_update", async (event) => {
+    if (event.message?.role === "assistant" && ["text_delta", "thinking_delta", "toolcall_start", "toolcall_delta"].includes(event.assistantMessageEvent?.type)) {
+      markActivity("activity");
+    }
+  });
+
   pi.on("tool_call", async (event) => {
     const verdict = decide(policy, event.toolName, event.input);
     return verdict.allow ? undefined : { block: true, reason: verdict.reason };

packages/harness/src/runner.mjs

@@ -415,6 +415,8 @@ export function adapterEnv(session, env) {
     MOSAIC_PROVIDER: session.provider ?? "",
     MOSAIC_MODEL: session.model,
     ...session.harnessEnv,
+    // Fixed marker metadata from adapter model events, before answer stdout.
+    ...(session.runDir ? { MOSAIC_MODEL_ACTIVITY_DIR: session.runDir } : {}),
     // Row 66: after the module's variables, so the log is always the run's own.
     ...(session.runDir ? { MOSAIC_UNSLOP_LOG: unslopLog(session.runDir) } : {}),
   };

packages/harness/tests/bundle.test.mjs

@@ -4,7 +4,7 @@ import { createHash } from "node:crypto";
 import { mkdirSync, readFileSync, statSync, writeFileSync } from "node:fs";
 import { join } from "node:path";
 import { BundleError, buildBundle, sessionModel, SOURCES } from "../src/bundle.mjs";
-import { HARNESS, REPO_SKILLS, resolvedFor, scratch } from "./helpers.mjs";
+import { HARNESS, REPO, REPO_SKILLS, resolvedFor, scratch } from "./helpers.mjs";
 import { decide } from "../src/gate.mjs";
 
 const UNSLOP_DIR = new URL("../../../tools/unslop-hook/", import.meta.url).pathname;
@@ -150,7 +150,9 @@ test("a claude-code bundle adds the wrapped gate hook and the MCP config", (t) =
   assert.deepEqual(m.gateTimeout, { n: 10, k: 2, hook: 20 });
   assert.ok(m.builtins.includes("Bash"));
   assert.ok(m.typed.every((n) => n.startsWith("mcp__mosaic__")));
-  assert.deepEqual(Object.keys(m.sources).sort(), ["claudeGate", "gate", "mcpServer", "tools", "unslopCheck", "unslopHook", "unslopLists"]);
+  assert.equal(m.sources.claudeStream.path, join(REPO, "adapters/claude/stream.mjs"));
+  assert.equal(m.sources.claudeStream.sha256, sha(m.sources.claudeStream.path));
+  assert.deepEqual(Object.keys(m.sources).sort(), ["claudeGate", "claudeStream", "gate", "mcpServer", "tools", "unslopCheck", "unslopHook", "unslopLists"]);
   // Row 66 (#1559): the unslop hooks run after a reply and before the next prompt, and never block.
   for (const [event, which] of [["Stop", "stop"], ["UserPromptSubmit", "prompt"]]) {
     assert.equal(settings.hooks[event].length, 1, event);

packages/harness/tests/claude-session.test.mjs

@@ -8,7 +8,7 @@ import { fixtureCacheEnv } from '../../../scripts/testing/lane-fixtures.mjs';
 import { test } from "node:test";
 import assert from "node:assert/strict";
 import { spawn, spawnSync } from "node:child_process";
-import { chmodSync, existsSync, mkdirSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs";
+import { chmodSync, existsSync, mkdirSync, readFileSync, realpathSync, rmSync, symlinkSync, writeFileSync } from "node:fs";
 import { basename, dirname, join } from "node:path";
 import { buildBundle } from "../src/bundle.mjs";
 import { adapterEnv, unslopCounts, unslopLog, unslopNotes } from "../src/runner.mjs";
@@ -142,13 +142,29 @@ test("claude: a second turn resumes the first turn's session", { skip }, async (
   assert.equal(readFileSync(join(s.sessionDir, "claude-session-id"), "utf8"), id);
 });
 
+for (const answer of ['Answer', '  Answer é 🦆\nsecond line\n', 'Answer\n\n']) {
+test(`claude: streaming preserves the real CLI text answer byte for byte (${JSON.stringify(answer)})`, { skip }, async t => {
+  const { dir, workspace, env } = await session(t, [], [], () => answer);
+  const textAdapter = join(dir, 'text-adapter.sh');
+  writeFileSync(textAdapter, readFileSync(ADAPTER, 'utf8')
+    .replace('node "$STREAM" "$CLAUDE"', '"$CLAUDE"')
+    .replace('--output-format stream-json --verbose --include-partial-messages', '--output-format text'));
+  const text = await turn({ ...env, MOSAIC_SESSION_DIR: join(dir, 'text-session') }, 'answer', workspace, textAdapter);
+  const stream = await turn(env, 'answer', workspace);
+  assert.equal(text.code, 0, text.stderr);
+  assert.equal(stream.code, 0, stream.stderr);
+  assert.equal(stream.stdout, text.stdout);
+  assert.ok(stream.stdout.includes(answer));
+});
+}
+
 // --restricted is what keeps CLAUDE.md files and auto-memory out of the
 // session's prompt (README "Limits"). This one runs without claude: a
 // stand-in records the adapter's argv.
 test("claude adapter: --restricted is always passed", (t) => {
   const dir = scratch(t);
   mkdirSync(join(dir, "bin"));
-  writeFileSync(join(dir, "bin", "claude"), '#!/bin/sh\nprintf "%s\\n" "$@" > "$ARGV_OUT"\necho ok\n');
+  writeFileSync(join(dir, "bin", "claude"), '#!/bin/sh\nprintf "%s\\n" "$@" > "$ARGV_OUT"\necho \'{"type":"result","subtype":"success","result":"ok"}\'\n');
   chmodSync(join(dir, "bin", "claude"), 0o755);
   for (const f of ["prompt.md", "settings.json", "mcp.json"]) writeFileSync(join(dir, f), "{}");
   const r = spawnSync("/bin/sh", [ADAPTER], {
@@ -179,7 +195,7 @@ test("claude adapter: MOSAIC_CLAUDE_BIN wins over PATH and must be an absolute e
   const dir = scratch(t);
   mkdirSync(join(dir, "bin"));
   const stand = (p, word) => {
-    writeFileSync(p, `#!/bin/sh\necho ${word} > "$WHO_OUT"\necho ok\n`);
+    writeFileSync(p, `#!/bin/sh\necho ${word} > "$WHO_OUT"\necho \'{"type":"result","subtype":"success","result":"ok"}\'\n`);
     chmodSync(p, 0o755);
     return p;
   };
@@ -229,7 +245,7 @@ test("claude adapter: a directory pin exits 2 before it runs; a pin with a space
   const dir = scratch(t);
   mkdirSync(join(dir, "bin"));
   const stand = (p, word) => {
-    writeFileSync(p, `#!/bin/sh\necho ${word} > "$WHO_OUT"\necho ok\n`);
+    writeFileSync(p, `#!/bin/sh\necho ${word} > "$WHO_OUT"\necho \'{"type":"result","subtype":"success","result":"ok"}\'\n`);
     chmodSync(p, 0o755);
     return p;
   };
@@ -293,6 +309,7 @@ test("claude: CLAUDE.md files and auto-memory don't reach the model; without --r
   assert.deepEqual(await seen(ADAPTER), []);
   // The control: the same adapter without --restricted lets all four in.
   const loose = join(dir, "adapter-loose.sh");
+  symlinkSync(join(REPO, "adapters/claude/stream.mjs"), join(dir, "stream.mjs"));
   writeFileSync(loose, readFileSync(ADAPTER, "utf8").replace("    --restricted \\\n", ""));
   assert.notEqual(readFileSync(loose, "utf8"), readFileSync(ADAPTER, "utf8"));
   assert.deepEqual(await seen(loose), Object.keys(plant));

packages/harness/tests/creative.test.mjs

@@ -174,7 +174,7 @@ function runAdapter(t, harness, b, { prompt = b.files.prompt, cwd } = {}) {
   const bin = join(b.dir, "bin");
   mkdirSync(bin, { recursive: true });
   const cli = join(bin, harness === "pi" ? "pi" : "claude");
-  writeFileSync(cli, '#!/bin/sh\nprintf "%s\\0" "$@" > "$ARGV_OUT"\necho ok\n');
+  writeFileSync(cli, '#!/bin/sh\nprintf "%s\\0" "$@" > "$ARGV_OUT"\necho \'{"type":"result","subtype":"success","result":"ok"}\'\n');
   chmodSync(cli, 0o755);
   const s = { harness, provider: harness === "pi" ? "probe" : null, model: "claude-sonnet-5-5", workspace: b.workspace, sessionDir: join(b.dir, "session"), toolSocket: join(b.dir, "tools.sock"), turnMarker: join(b.dir, "turn.done"), bundle: { ...b.files, prompt }, harnessEnv: b.env };
   const env = { ...adapterEnv(s, { PATH: `${bin}:/usr/bin:/bin`, HOME: join(b.dir, "home") }), ARGV_OUT: join(b.dir, "argv"), MOSAIC_REQUEST: "x" };

packages/harness/tests/fixtures/claude-activity.mjs

@@ -0,0 +1,42 @@
+// No network: a Claude CLI stand-in that distinguishes text and stream modes.
+import { writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+const args = process.argv.slice(2);
+const answer = '  Answer é 🦆\nsecond line\n';
+const mode = process.env.FAKE_CLAUDE_MODE ?? 'tool';
+writeFileSync(join(process.env.MOSAIC_WORKSPACE, 'cli-pid'), String(process.pid));
+const emit = item => process.stdout.write(JSON.stringify(item) + '\n');
+if (args[args.indexOf('--output-format') + 1] === 'text') {
+  process.stdout.write(answer.endsWith('\n') ? answer : answer + '\n');
+} else {
+  if (!args.includes('--verbose') || !args.includes('--include-partial-messages')) process.exit(9);
+  emit({ type: 'system', subtype: 'init', private: 'credential-canary' });
+  emit({ type: 'user', message: { role: 'user', content: [{ type: 'tool_result', content: 'local work' }] } });
+  emit({ type: 'assistant', error: 'api_error', message: { role: 'assistant', model: '<synthetic>', content: [{ type: 'text', text: 'local CLI error' }] } });
+  // Local file writes and diagnostic text must never cancel the watchdog.
+  writeFileSync(join(process.env.MOSAIC_WORKSPACE, 'draft'), 'workspace-canary');
+  process.stderr.write('fixture diagnostic\n');
+  if (mode === 'malformed') {
+    process.stdout.write('invalid credential-canary\n');
+  } else if (mode === 'missing') {
+    // Exit 0 without a result still fails the adapter.
+  } else if (mode === 'failed') {
+    emit({ type: 'result', subtype: 'error_during_execution', is_error: true, errors: ['credential-canary'] });
+    process.exitCode = 7;
+  } else {
+    if (mode !== 'silent') {
+      const event = mode === 'thinking'
+        ? { type: 'content_block_delta', delta: { type: 'thinking_delta', thinking: 'private reasoning' } }
+        : mode === 'header'
+          ? { type: 'message_start', message: { role: 'assistant' } }
+          : mode === 'assistant'
+            ? null
+            : { type: 'content_block_start', content_block: { type: 'tool_use', id: 'tool-1', name: 'Write', input: {} } };
+      if (event) emit({ type: 'stream_event', event });
+      else emit({ type: 'assistant', message: { role: 'assistant', content: [{ type: 'text', text: 'private draft' }] } });
+    }
+    setTimeout(() => {
+      emit({ type: 'result', subtype: 'success', is_error: false, result: answer });
+    }, mode === 'interrupt' || mode === 'silent' ? 30_000 : Number(process.env.FAKE_CLAUDE_DELAY ?? 100));
+  }
+}

packages/harness/tests/model-activity.test.mjs

@@ -0,0 +1,153 @@
+// Real adapter and watchdog, with a fake CLI: no models, proxy or credentials.
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { spawn, spawnSync } from 'node:child_process';
+import { existsSync, mkdirSync, readFileSync, statSync, unlinkSync, writeFileSync } from 'node:fs';
+import { dirname, join } from 'node:path';
+import { watchModelOutput } from '../../cli/src/model-stall.mjs';
+import { firstTurnMarker } from '../src/first-output.mjs';
+import { adapterEnv } from '../src/runner.mjs';
+import piExtension from '../src/pi-extension.mjs';
+import { REPO, scratch } from './helpers.mjs';
+
+const ADAPTER = join(REPO, 'adapters/claude/adapter.sh');
+const FAKE = new URL('./fixtures/claude-activity.mjs', import.meta.url).pathname;
+const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
+function fixture(t, mode = 'tool') {
+  const runDir = scratch(t), workspace = join(runDir, 'ws');
+  mkdirSync(workspace);
+  for (const name of ['prompt', 'settings', 'mcp']) writeFileSync(join(runDir, name), '{}');
+  const cli = join(runDir, 'claude');
+  writeFileSync(cli, `#!/bin/sh\nexec '${process.execPath}' '${FAKE}' "$@"\n`, { mode: 0o700 });
+  const session = {
+    runDir, workspace, sessionDir: join(runDir, 'session'), model: 'proxy-model',
+    bundle: { prompt: join(runDir, 'prompt') },
+    harnessEnv: { MOSAIC_CLAUDE_SETTINGS: join(runDir, 'settings'), MOSAIC_CLAUDE_MCP_CONFIG: join(runDir, 'mcp'), MOSAIC_THINKING: 'high' },
+  };
+  const env = adapterEnv(session, {
+    PATH: `${dirname(process.execPath)}:/usr/bin:/bin`, MOSAIC_CLAUDE_BIN: cli,
+    MOSAIC_REQUEST: 'write a file', FAKE_CLAUDE_MODE: mode,
+  });
+  return { runDir, workspace, env, cli };
+}
+function start(t, f) {
+  const child = spawn('/bin/sh', [ADAPTER], { env: f.env, stdio: ['ignore', 'pipe', 'pipe'], detached: true });
+  const kill = () => { try { process.kill(-child.pid, 'SIGTERM'); } catch {} };
+  let stdout = Buffer.alloc(0), stderr = '';
+  child.stdout.on('data', b => { stdout = Buffer.concat([stdout, b]); });
+  child.stderr.on('data', b => { stderr += b; });
+  const done = new Promise((resolve, reject) => {
+    child.once('error', reject);
+    child.once('close', (code, signal) => resolve({ code, signal, stdout, stderr }));
+  });
+  t.after(async () => { kill(); await done; });
+  return { child, done, kill, stdout: () => stdout };
+}
+
+for (const mode of ['tool', 'thinking', 'header', 'assistant']) {
+  test(`Claude ${mode} activity survives a short watchdog window before final stdout`, { timeout: 10_000 }, async t => {
+    const f = fixture(t, mode);
+    f.env.FAKE_CLAUDE_DELAY = '1400';
+    firstTurnMarker(f.runDir)();
+    const turn = start(t, f), receipts = [];
+    let stops = 0;
+    const watch = watchModelOutput({ runDir: f.runDir, seconds: 0.8, armAtFirstTurn: true, keyId: '0123456789ab',
+      stop: async () => { stops++; turn.kill(); }, receipt: r => receipts.push(r) });
+    t.after(watch.cancel);
+    await watch.done;
+    assert.equal(stops, 0);
+    assert.equal(turn.stdout().length, 0, 'model events stay private while the tool turn is in progress');
+    assert.equal(readFileSync(join(f.runDir, 'proxy-first-output'), 'utf8'), 'output\n');
+    assert.equal(statSync(join(f.runDir, 'proxy-first-output')).mode & 0o777, 0o600);
+    const result = await turn.done;
+    assert.equal(result.code, 0, result.stderr);
+    // Compare exact bytes against the same CLI's original text format.
+    const text = spawnSync(f.cli, ['--output-format', 'text'], { env: f.env });
+    assert.equal(text.status, 0);
+    assert.deepEqual(result.stdout, text.stdout);
+    assert.deepEqual(receipts, []);
+    assert.match(readFileSync(join(f.runDir, 'session/claude-session-id'), 'utf8'), /^[0-9a-f-]{36}\n$/);
+  });
+}
+
+test('silent Claude is stopped with proxy-stall despite init, tool results, stderr and workspace writes', { timeout: 10_000 }, async t => {
+  const f = fixture(t, 'silent'), turn = start(t, f), receipts = [];
+  firstTurnMarker(f.runDir)();
+  const watch = watchModelOutput({ runDir: f.runDir, seconds: 0.8, armAtFirstTurn: true, keyId: '0123456789ab',
+    stop: async () => turn.kill(), receipt: r => receipts.push(r) });
+  t.after(watch.cancel);
+  await watch.done;
+  const result = await turn.done;
+  assert.notEqual(result.code, 0);
+  assert.equal(result.stdout.length, 0);
+  assert.equal(existsSync(join(f.workspace, 'draft')), true, 'workspace activity alone is insufficient');
+  assert.equal(existsSync(join(f.runDir, 'proxy-first-output')), false);
+  assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
+  assert.deepEqual(receipts, [{ action: 'proxy-stall', keyIds: ['0123456789ab'] }]);
+});
+
+for (const resumed of [false, true]) {
+test(`interrupted Claude activity emits no partial answer, preserves session state and kills its CLI (resume=${resumed})`, { timeout: 10_000 }, async t => {
+  const f = fixture(t, 'interrupt');
+  const sessionFile = join(f.runDir, 'session/claude-session-id');
+  if (resumed) {
+    mkdirSync(join(f.runDir, 'session'));
+    writeFileSync(sessionFile, 'previous-session-id\n');
+  }
+  const turn = start(t, f);
+  const deadline = Date.now() + 5000;
+  while (!existsSync(join(f.runDir, 'proxy-first-output'))) {
+    assert.ok(Date.now() < deadline, 'activity marker appears');
+    await delay(10);
+  }
+  const pid = Number(readFileSync(join(f.workspace, 'cli-pid'), 'utf8'));
+  turn.kill();
+  const result = await turn.done;
+  assert.notEqual(result.code, 0);
+  assert.equal(result.stdout.length, 0);
+  if (resumed) assert.equal(readFileSync(sessionFile, 'utf8'), 'previous-session-id\n');
+  else assert.equal(existsSync(sessionFile), false);
+  // The CLI is a grandchild; its pipe can close just before the OS reaps it.
+  const reapDeadline = Date.now() + 5000;
+  for (;;) {
+    try { process.kill(pid, 0); }
+    catch (error) { assert.equal(error.code, 'ESRCH'); break; }
+    assert.ok(Date.now() < reapDeadline, 'interrupted CLI is reaped');
+    await delay(10);
+  }
+});
+}
+
+for (const mode of ['malformed', 'missing', 'failed']) {
+  test(`Claude ${mode} stream fails closed without raw event diagnostics or a session receipt`, async t => {
+    const f = fixture(t, mode), result = await start(t, f).done;
+    assert.notEqual(result.code, 0);
+    if (mode === 'failed') assert.equal(result.code, 7, 'preserve the CLI exit code');
+    assert.equal(result.stdout.length, 0);
+    assert.doesNotMatch(result.stderr, /credential-canary/);
+    assert.equal(existsSync(join(f.runDir, 'session/claude-session-id')), false);
+  });
+}
+
+test('Pi extension marks real streamed thinking/tools/text, never local activity or user/tool messages', async t => {
+  const runDir = scratch(t);
+  writeFileSync(join(runDir, 'policy'), JSON.stringify({ harness: 'pi', tools: [], typed: [] }));
+  writeFileSync(join(runDir, 'tools'), '[]');
+  const values = { MOSAIC_POLICY_FILE: join(runDir, 'policy'), MOSAIC_TOOLS_FILE: join(runDir, 'tools'),
+    MOSAIC_TOOL_SOCKET: join(runDir, 'sock'), MOSAIC_TURN_MARKER: join(runDir, 'done'), MOSAIC_MODEL_ACTIVITY_DIR: runDir };
+  const prior = Object.fromEntries(Object.keys(values).map(k => [k, process.env[k]]));
+  Object.assign(process.env, values);
+  t.after(() => { for (const [k, v] of Object.entries(prior)) { if (v === undefined) delete process.env[k]; else process.env[k] = v; } });
+  for (const type of ['thinking_delta', 'toolcall_start', 'toolcall_delta', 'text_delta']) {
+    const handlers = new Map();
+    piExtension({ on: (name, fn) => handlers.set(name, fn), registerTool() {} });
+    const update = handlers.get('message_update');
+    for (const role of ['user', 'toolResult']) await update({ message: { role }, assistantMessageEvent: { type } });
+    await update({ message: { role: 'assistant' }, assistantMessageEvent: { type: 'start' } });
+    assert.equal(existsSync(join(runDir, 'proxy-first-output')), false);
+    await update({ message: { role: 'assistant' }, assistantMessageEvent: { type } });
+    assert.equal(readFileSync(join(runDir, 'proxy-first-output'), 'utf8'), 'output\n');
+    // A new producer on each case proves each event type on its own.
+    unlinkSync(join(runDir, 'proxy-first-output'));
+  }
+});

packages/harness/tests/skills.test.mjs

@@ -174,7 +174,7 @@ test("each adapter passes the skills only in the prompt file, with skill discove
     mkdirSync(workspace);
     mkdirSync(join(dir, "bin"));
     const cli = join(dir, "bin", harness === "pi" ? "pi" : "claude");
-    writeFileSync(cli, '#!/bin/sh\nprintf "%s\\0" "$@" > "$ARGV_OUT"\necho ok\n');
+    writeFileSync(cli, '#!/bin/sh\nprintf "%s\\0" "$@" > "$ARGV_OUT"\necho \'{"type":"result","subtype":"success","result":"ok"}\'\n');
     chmodSync(cli, 0o755);
     const resolved = resolvedFor(t, instance, (d) => (d.roles[instance].vars = { harness, model: "claude-opus-5-5", thinking: "high" }));
     assert.equal(resolved.skills.length > 0, instance === "reviewer", instance);

packages/model-proxy/README.md

@@ -91,9 +91,12 @@ before reading config or touching the network/disk.
 The trusted launcher requests `modelPreflight` with business/role and receives
 exactly `{ baseUrl, key }`. That answer never passes through an agent socket or
 status API. A 401 or other non-success HTTP status refuses `proxy-key-refused`;
-a failed connection refuses `proxy-down`. No stdout by the configured deadline
+a failed connection refuses `proxy-down`. No model response activity by the configured deadline
 requests a bounded stop and records `proxy-stall` with key IDs. The runner writes
-only a fixed token to `<run>/proxy-first-output`, never stdout contents. A stop
+only a fixed token to `<run>/proxy-first-output`, never stdout contents. Managed
+adapters also publish this token on trusted streamed model response events
+(including thinking and tool-use), before their answer-only stdout. Startup
+status, tool results and workspace file writes do not count. A stop
 failure records/refuses `proxy-stop-failed` and retains the key.
 
 P5 adds the bus-only `readUsageQueue()` management operation. Its fixed

packages/plugins/harness/claude-code/index.mjs

@@ -16,6 +16,7 @@ export const interfaceVersion = 1;
 const REPO = fileURLToPath(new URL("../../../../", import.meta.url));
 const GATE = fileURLToPath(new URL("../../../harness/src/claude-gate.mjs", import.meta.url));
 const MCP_SERVER = fileURLToPath(new URL("../../../harness/src/mcp-server.mjs", import.meta.url));
+const STREAM = join(REPO, "adapters", "claude", "stream.mjs");
 // Row 66 (#1559): the unslop check's hook script, which the Stop and UserPromptSubmit hooks run. It
 // logs to MOSAIC_UNSLOP_LOG, which the runner sets, and the runner tells the operator.
 const UNSLOP = Object.freeze({
@@ -80,6 +81,7 @@ export const countsAccount = () => true;
 // hook's files.
 export function mounts({ env }) {
   const unslop = Object.values(UNSLOP).map((src) => ({ src, why: "the unslop check the session's Stop and UserPromptSubmit hooks run" }));
+  unslop.push({ src: STREAM, why: "the adapter's private model stream reader" });
   if (!env.MOSAIC_CLAUDE_BIN) return unslop;
   const real = realpathSync(env.MOSAIC_CLAUDE_BIN);
   return [...(real.startsWith("/usr/") ? [] : [{ src: real, dest: env.MOSAIC_CLAUDE_BIN, why: "the pinned Claude Code binary (MOSAIC_CLAUDE_BIN)" }]), ...unslop];
@@ -113,7 +115,7 @@ export function bundle({ dir, files: core, session, write, quote }) {
       mcpServers: { mosaic: { type: "stdio", command: node, args: [MCP_SERVER, core.tools, session.toolSocket] } },
     }, null, 2) + "\n"),
   ];
-  return { files, digests, sources: { claudeGate: GATE, mcpServer: MCP_SERVER, ...UNSLOP } };
+  return { files, digests, sources: { claudeGate: GATE, mcpServer: MCP_SERVER, claudeStream: STREAM, ...UNSLOP } };
 }
 
 export function env({ files, policy, session, proxy = null }) {

packages/plugins/tests/harness.test.mjs

@@ -74,8 +74,8 @@ test("row 65 claude-code preflight: a missing credential, then a missing or rela
   // The pinned binary outside /usr is mounted where the adapter looks for it, and the unslop hook's
   // files where the hook commands name them (row 66).
   const unslop = ["claude-hook.js", "unslop-check.js", "lists.json"].map(f => [join(REPO, "tools", "unslop-hook", f), undefined]);
-  assert.deepEqual(claude.mounts({ env: { MOSAIC_CLAUDE_BIN: bin } }).map(m => [m.src, m.dest]), [[bin, bin], ...unslop]);
-  assert.deepEqual(claude.mounts({ env: {} }).map(m => [m.src, m.dest]), unslop);
+  assert.deepEqual(claude.mounts({ env: { MOSAIC_CLAUDE_BIN: bin } }).map(m => [m.src, m.dest]), [[bin, bin], ...unslop, [join(REPO, "adapters/claude/stream.mjs"), undefined]]);
+  assert.deepEqual(claude.mounts({ env: {} }).map(m => [m.src, m.dest]), [...unslop, [join(REPO, "adapters/claude/stream.mjs"), undefined]]);
 });
 
 // Row 68's claudeVersion precedence, moved here with the version (row 65): the manifest records
@@ -137,7 +137,7 @@ test("row 65 a route's thinking reaches pi as --thinking and claude as --effort,
   const dir = scratch(t);
   mkdirSync(join(dir, "bin"));
   for (const name of ["pi", "claude"]) {
-    writeFileSync(join(dir, "bin", name), '#!/bin/sh\nprintf "%s\\n" "$@" > "$ARGV_OUT"\n');
+    writeFileSync(join(dir, "bin", name), '#!/bin/sh\nprintf "%s\\n" "$@" > "$ARGV_OUT"\necho \'{"type":"result","subtype":"success","result":"ok"}\'\n');
     chmodSync(join(dir, "bin", name), 0o755);
   }
   for (const f of ["prompt.md", "settings.json", "mcp.json"]) writeFileSync(join(dir, f), "{}");