From 1bd0077391f1925db5e66796b98540e48f13bfc2 Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Thu, 5 Nov 2026 20:45:03 -0600 Subject: [PATCH] fix(harness): mark model activity before final answers (#797) --- BUILD-LOG.md | 9 ++ adapters/README.md | 9 ++ adapters/claude/adapter.sh | 9 +- adapters/claude/stream.mjs | 72 +++++++++ docs/SESSIONS.md | 2 + packages/harness/src/first-output.mjs | 3 +- packages/harness/src/pi-extension.mjs | 11 ++ packages/harness/src/runner.mjs | 2 + packages/harness/tests/bundle.test.mjs | 6 +- .../harness/tests/claude-session.test.mjs | 25 ++- packages/harness/tests/creative.test.mjs | 2 +- .../tests/fixtures/claude-activity.mjs | 42 +++++ .../harness/tests/model-activity.test.mjs | 153 ++++++++++++++++++ packages/harness/tests/skills.test.mjs | 2 +- packages/model-proxy/README.md | 7 +- .../plugins/harness/claude-code/index.mjs | 4 +- packages/plugins/tests/harness.test.mjs | 6 +- 17 files changed, 346 insertions(+), 18 deletions(-) create mode 100644 adapters/claude/stream.mjs create mode 100644 packages/harness/tests/fixtures/claude-activity.mjs create mode 100644 packages/harness/tests/model-activity.test.mjs diff --git a/BUILD-LOG.md b/BUILD-LOG.md index 0fc793057..1de8e9c14 100644 --- a/BUILD-LOG.md +++ b/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. diff --git a/adapters/README.md b/adapters/README.md index 772e7d31a..4bda20b31 100644 --- a/adapters/README.md +++ b/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, `/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 diff --git a/adapters/claude/adapter.sh b/adapters/claude/adapter.sh index 84079b338..963792ea9 100644 --- a/adapters/claude/adapter.sh +++ b/adapters/claude/adapter.sh @@ -11,6 +11,7 @@ # wrapped as `timeout -k 2 10 || 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//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" \ diff --git a/adapters/claude/stream.mjs b/adapters/claude/stream.mjs new file mode 100644 index 000000000..30a06f606 --- /dev/null +++ b/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 !== '' && 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'); + } +}); diff --git a/docs/SESSIONS.md b/docs/SESSIONS.md index 3a717e1c2..84c45f841 100644 --- a/docs/SESSIONS.md +++ b/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. diff --git a/packages/harness/src/first-output.mjs b/packages/harness/src/first-output.mjs index db3b19cee..23cf6f710 100644 --- a/packages/harness/src/first-output.mjs +++ b/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 => { diff --git a/packages/harness/src/pi-extension.mjs b/packages/harness/src/pi-extension.mjs index 5647f0f52..e538b0efb 100644 --- a/packages/harness/src/pi-extension.mjs +++ b/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 }; diff --git a/packages/harness/src/runner.mjs b/packages/harness/src/runner.mjs index 9d870be6b..c827e2cf4 100644 --- a/packages/harness/src/runner.mjs +++ b/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) } : {}), }; diff --git a/packages/harness/tests/bundle.test.mjs b/packages/harness/tests/bundle.test.mjs index f2e26e8c4..af2967406 100644 --- a/packages/harness/tests/bundle.test.mjs +++ b/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); diff --git a/packages/harness/tests/claude-session.test.mjs b/packages/harness/tests/claude-session.test.mjs index d9c45e446..8b24f1c5c 100644 --- a/packages/harness/tests/claude-session.test.mjs +++ b/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)); diff --git a/packages/harness/tests/creative.test.mjs b/packages/harness/tests/creative.test.mjs index c3191f99e..d9ee76af8 100644 --- a/packages/harness/tests/creative.test.mjs +++ b/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" }; diff --git a/packages/harness/tests/fixtures/claude-activity.mjs b/packages/harness/tests/fixtures/claude-activity.mjs new file mode 100644 index 000000000..7b20928e3 --- /dev/null +++ b/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: '', 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)); + } +} diff --git a/packages/harness/tests/model-activity.test.mjs b/packages/harness/tests/model-activity.test.mjs new file mode 100644 index 000000000..96812d558 --- /dev/null +++ b/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')); + } +}); diff --git a/packages/harness/tests/skills.test.mjs b/packages/harness/tests/skills.test.mjs index 4baa3027d..e7581d8b9 100644 --- a/packages/harness/tests/skills.test.mjs +++ b/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); diff --git a/packages/model-proxy/README.md b/packages/model-proxy/README.md index d9b2d0bdd..ac2f636f3 100644 --- a/packages/model-proxy/README.md +++ b/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 `/proxy-first-output`, never stdout contents. A stop +only a fixed token to `/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 diff --git a/packages/plugins/harness/claude-code/index.mjs b/packages/plugins/harness/claude-code/index.mjs index 95f851a49..9f6367f85 100644 --- a/packages/plugins/harness/claude-code/index.mjs +++ b/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 }) { diff --git a/packages/plugins/tests/harness.test.mjs b/packages/plugins/tests/harness.test.mjs index 0cf6c5966..5ff4bb558 100644 --- a/packages/plugins/tests/harness.test.mjs +++ b/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), "{}");