fix(harness): mark first streamed model output (#797)

BUILD-LOG.md

@@ -6206,3 +6206,7 @@ After: personally authored three-file skills/ms-ux-improvement coordinator packa
 2026-11-06 UTC | #797 ui797 support R5 conflicting-ledger correction verified | Before: fce831c05 let a valid pin and a v2 ledger disagree on target and still halted the ledger's binding; the reviewer's unchanged target-conflict probe sent 8 halts to the wrong business/config (business-only corruption and a foreign ledger, check and serve startup). After: checkTarget runs corroborate() on every tick. A ledger target directs a halt only when the owner campaign, the owner pin, or the live binding of the owner campaign names the same target; otherwise the owner pin replaces it only when it matches the owner campaign hash, the live config/marker/root/store and the live config digest (evidence 'ledger-target-changed'); anything else refuses with campaign-instance-changed. Retained-target comparison stays as the second guard. Four new regressions (check, serve startup, running service, no corroboration) fail 4/4 on r4 source and pass after. Probe after fix: 0 wrong halts in all six cases. Sequential whole files budget 93, identity 8, browser 13, request-lifecycle-live 2 = 116 pass / 0 fail / 0 skip. No push, stash, .53, live service or model spend. Evidence: /mnt/storage/scratch/mosaic-lead/mvp3/ui797-support-r5-fix-evidence/.
 
 2026-11-06 UTC | #797 ui797 support R6 matching-damage and ledger-path correction verified | Before: 85e1ec5d1 rule (b) accepted a pin whose binding.business had been damaged to match a damaged ledger, and a foreign pin restored with the invoked ledger path; the reviewer's probe sent 8 halts to business B in check and serve startup. serveBudget re-read opts.ledger every tick, so a programmatic caller that mutated it moved stop/halt receipts to the replacement ledger. After: a pin corroborates a ledger target only when its binding names its own campaign's instance/config/business/owner and, while the owner campaign is readable, the owner's too. serveBudget freezes the startup campaign/ledger paths and retains the ledger path with the startup target; a running service accepts an exactly unchanged startup ledger as its own corroboration. Tradeoff (recorded): with no retained evidence, an owner business/config edit while the pin and ledger are intact now refuses (ambiguous target, indistinguishable from a forged pin+ledger pair) instead of halting the original; a running service still halts its retained startup target with reason campaign-changed; owner cap/epoch edits and sidecar-only damage still stop. Proof: 4 new regressions, 3 fail on 85e1ec5d1 and all pass after (the running matching-damage case already refused on r5); whole files budget-supervisor 97, isolated-identity 8, ui-browser 13, request-lifecycle-live 2, all pass/0 fail; reviewer probes with r6 expectations: matching-damage 6 cases 0 wrong halts, ledger-path 2 halts to the startup binding with stop/halt receipts only in the startup ledger, adversarial 104 cases verified (3 owner-retarget cases refuse in check/startup), r4 target-conflict exit 0. Evidence /mnt/storage/scratch/mosaic-lead/mvp3/ui797-support-r6-evidence/. Local commit only; independent r6 review next.
+
+2026-11-06 UTC | #797 stall-guard implementation started | Pinned 5df42478b6c1ef5e22e2eb6c3fa8605722ab5f77 on feat/ui797-stall-guard. Before: both pi print mode and Claude text mode emit the turn reply only at completion, so a live first turn can be stopped despite model output; a chief can also immediately repeat an instance-running launch refusal. Scope is event-based first output and bounded launch refusal handling with focused and package gates; no configuration, role, schema, proxy initialization or live-instance change. The queue command refused in the linked worktree; no canonical queue access.
+
+2026-11-06 UTC | #797 F6 streamed first output | Managed pi JSON events and Claude stream-json partial events pass through one harness monitor; substantive text, thinking and tool-use events mark a fixed output token while stdout remains the final answer. Six real confined runner/adapter controls use a six-second guard: both streaming turns survived and both headers-only and keepalive-only cases stopped; both streaming tests failed on the pinned base and all four stall controls passed there. Harness suite 250 passed; CLI main partition 909 passed, one failed and one skipped, with its 53-test file rerun green; the failure also reproduced on the base. No guard threshold, proxy initialization, configuration, schema or role changes. Release and lane partitions are retained under /mnt/storage/scratch/mosaic-lead/mvp3/stall-guard-evidence/.

adapters/claude/adapter.sh

@@ -84,8 +84,10 @@ 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 \
+OUTPUT_FORMAT=text
+[ -z "${MOSAIC_MODEL_EVENT_MONITOR:-}" ] || OUTPUT_FORMAT=stream-json
+set -- "$CLAUDE" -p "$MOSAIC_REQUEST" \
+    --output-format "$OUTPUT_FORMAT" \
     --system-prompt-file "$PROMPT_FILE" \
     --model "$MOSAIC_MODEL" \
     --effort "$MOSAIC_THINKING" \
@@ -97,5 +99,12 @@ ALLOWED=mcp__mosaic
     --strict-mcp-config \
     --mcp-config "$MOSAIC_CLAUDE_MCP_CONFIG" \
     --disable-slash-commands \
-    "$SESSION_FLAG" "$SESSION_ID" || exit $?
+    "$SESSION_FLAG" "$SESSION_ID"
+if [ -n "${MOSAIC_MODEL_EVENT_MONITOR:-}" ]; then
+    # Stream partial model events to the shared monitor; its stdout is still answer-only.
+    set -- "$@" --verbose --include-partial-messages
+    node "$MOSAIC_MODEL_EVENT_MONITOR" claude "$MOSAIC_FIRST_OUTPUT_DIR" "$@" || exit $?
+else
+    "$@" || exit $?
+fi
 [ "$SESSION_FLAG" = --resume ] || printf '%s\n' "$SESSION_ID" > "$ID_FILE"

adapters/pi/adapter.sh

@@ -132,4 +132,7 @@ set -- \
 # One-shot mode appends -p and the request (both safely quoted);
 # interactive mode appends nothing - clean TUI.
 [ "$INTERACTIVE" = "1" ] || set -- "$@" -p "$MOSAIC_REQUEST"
+if [ -n "${MOSAIC_MODEL_EVENT_MONITOR:-}" ] && [ "$INTERACTIVE" != "1" ]; then
+    exec node "$MOSAIC_MODEL_EVENT_MONITOR" pi "$MOSAIC_FIRST_OUTPUT_DIR" pi "$@" --mode json
+fi
 exec pi "$@"

packages/cli/tests/model-event-guard.test.mjs

@@ -0,0 +1,48 @@
+// Real confined runner and adapter, real local model stream; only fixture children are stopped.
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import { spawnSync } from 'node:child_process';
+import { join } from 'node:path';
+import { sandboxProblem } from '../../seat/src/sandbox.mjs';
+import { managedLaunchFixture, pause, waitFor } from './model-managed-launch-fixture.mjs';
+
+const skip = sandboxProblem() ?? false;
+const claude = spawnSync('sh', ['-c', 'command -v claude'], {encoding:'utf8'}).stdout.trim();
+for (const harness of ['pi', 'claude-code']) {
+  for (const mode of ['stream', 'headers', 'keepalive']) {
+    test(`797 ${harness} ${mode}: real model stream controls the managed guard`, {skip: skip || (harness === 'claude-code' && !claude ? 'claude is not on PATH' : false), timeout:40000}, async t => {
+      const modelStream = { mode, seconds:6, duration:8500 };
+      const f = await managedLaunchFixture(t, { harness, real:[harness], modelStream,
+        hostEnv: harness === 'claude-code' ? {MOSAIC_CLAUDE_BIN:claude} : {} });
+      const launched = await f.launcher.launchPm();
+      const session = f.sessions().find(s => s.run === launched.run);
+      assert.ok(session?.live);
+      const sent = f.send('Respond to the test.');
+      try { await waitFor(() => f.state.modelRequests, 'model request reached the local server', 10000); }
+      catch (e) {
+        const file = join(session.runDir, 'turns', '0001.stderr');
+        let diagnostic = fs.existsSync(file) ? fs.readFileSync(file, 'utf8') : 'no adapter stderr';
+        for (const key of f.state.keys) diagnostic = diagnostic.replaceAll(key, '[fixture key]');
+        t.diagnostic(`adapter stderr: ${diagnostic}; launcher log: ${f.logs.join('; ')}`);
+        throw e;
+      }
+      if (mode === 'stream') {
+        await waitFor(() => fs.existsSync(join(session.runDir,'proxy-first-output')) || f.trace.stalls.length, 'model delta or stall observed', 10000);
+        assert.equal(f.trace.stalls.length, 0, 'streaming must not be stopped before its final output');
+        assert.equal(fs.existsSync(join(session.runDir,'proxy-first-output')), true, 'real model delta produces marker');
+        assert.equal(fs.readFileSync(join(session.runDir,'proxy-first-output'),'utf8'), 'output\n');
+        await pause(7000);
+        assert.equal(f.trace.stalls.length, 0, 'streaming past the window never requests a stop');
+        assert.equal(f.sessions().some(s => s.run === session.run && s.live), true);
+        const reply = await f.reply(sent);
+        assert.match(reply.body, /first/);
+      } else {
+        await waitFor(() => f.trace.stalls.length === 1, 'silent model triggers trusted stop', 12000);
+        assert.equal(fs.existsSync(join(session.runDir,'proxy-first-output')), false);
+        assert.equal(f.trace.stalls[0].stopped, true);
+      }
+      f.valueFree();
+    });
+  }
+}

packages/cli/tests/model-managed-launch-fixture.mjs

@@ -21,12 +21,12 @@ const id = value => createHash('sha256').update(value).digest('hex').slice(0,12)
 // #774 R2: extra names more harnesses to switch on and load (a fallback route's); real names the
 // harnesses loaded as the real module, so their own preflight runs; hostEnv adds launcher env
 // such as MOSAIC_CLAUDE_BIN.
-export async function managedLaunchFixture(t,{proxy=true,launches=true,refuseProbe=false,harness='pi',bwrap=null,admit=null,extra=[],real=[],hostEnv={}}={}){
+export async function managedLaunchFixture(t,{proxy=true,launches=true,refuseProbe=false,harness='pi',bwrap=null,admit=null,extra=[],real=[],hostEnv={},modelStream=null}={}){
  const root=fs.mkdtempSync(join(tmpdir(),'ml653-')),closers=[],history=new Set(['fx653-manage']),proxyReceipts=[],logs=[];
  let server,bus,launcher,stderr;const trace={preflightAttempts:0,successfulPreflights:0,enableBeforeSpawn:[],bindings:[],stalls:[],ops:[]};
  const checkHistory=entries=>{const text=JSON.stringify(entries);for(const value of history)if(text.includes(value))throw Error('fixture credential leaked');};
  t.after(async()=>{const failures=[];for(const close of [...closers].reverse()){try{await close();}catch{failures.push('owned-close');}}try{server?.closeAllConnections();if(server?.listening)await new Promise((resolve,reject)=>server.close(e=>e?reject(e):resolve()));}catch{failures.push('owned-server-close');}try{checkHistory({logs,proxyReceipts,trace});}catch{failures.push('value-free-receipts');}if(stderr)process.stderr.write=stderr;try{fs.rmSync(root,{recursive:true,force:true});}catch{failures.push('owned-root-close');}if(failures.length)throw new AggregateError(failures.map(code=>Error(code)),'owned managed fixture cleanup failed');});
- const f=fixture(root,'acme',doc=>{doc.plugins={...doc.plugins,harness:[harness,...extra]};for(const role of Object.values(doc.roles))role.vars={...role.vars,harness};doc.roles.chief.vars.model='claude-opus-5-5';return doc;});
+ const f=fixture(root,'acme',doc=>{doc.plugins={...doc.plugins,harness:[harness,...extra]};for(const role of Object.values(doc.roles))role.vars={...role.vars,harness};doc.roles.chief.vars.model='claude-opus-5-5';if(modelStream){doc.launch.models['claude-opus-5-5']={provider:'anthropic',model:'claude-opus-5-5'};}return doc;});
  const env={PATH:process.env.PATH,HOME:process.env.HOME,TMPDIR:process.env.TMPDIR,MOSAIC_CONFIG:f.config,MOSAIC_ROLES_DIR:f.roles,...hostEnv};
  // The nonproxy positive fixture supplies its own fake pi login. Managed
  // proxy fixtures deliberately hold no upstream credential at all.
@@ -34,7 +34,25 @@ export async function managedLaunchFixture(t,{proxy=true,launches=true,refusePro
  const configDir=join(root,'config'),prefix=`msk-${id(configDir)}-`,sentinel=prefix+'sentinel-'+randomBytes(32).toString('base64url');history.add(sentinel);
  const state={keys:[sentinel],calls:[],refuseProbe};
  server=http.createServer(async(req,res)=>{try{
-  const supplied=req.headers.authorization?.slice(7);if(supplied)history.add(supplied);for(const key of state.keys)history.add(key);
+  const supplied=req.headers.authorization?.slice(7) ?? req.headers['x-api-key'];if(supplied)history.add(supplied);for(const key of state.keys)history.add(key);
+  if(modelStream && req.method === 'POST' && /\/messages(?:\?|$)/.test(req.url)) {
+   if (!state.keys.includes(supplied)) { res.writeHead(401);res.end();return; }
+   req.resume(); req.on('end', () => {
+    state.modelRequests = (state.modelRequests ?? 0) + 1;
+    res.writeHead(200, {'content-type':'text/event-stream','cache-control':'no-cache'});res.flushHeaders();
+    const send=(type,data)=>res.write(`event: ${type}\ndata: ${JSON.stringify({type,...data})}\n\n`);
+    let ticker;
+    if(modelStream.mode === 'stream') {
+     send('message_start',{message:{id:'msg_test',type:'message',role:'assistant',model:'claude-opus-5-5',content:[],stop_reason:null,stop_sequence:null,usage:{input_tokens:5,output_tokens:1,cache_creation_input_tokens:0,cache_read_input_tokens:0}}});
+     send('content_block_start',{index:0,content_block:{type:'text',text:''}});
+     send('content_block_delta',{index:0,delta:{type:'text_delta',text:'first'}});
+     ticker=setInterval(()=>send('content_block_delta',{index:0,delta:{type:'text_delta',text:' more'}}),400);
+     const end=setTimeout(()=>{clearInterval(ticker);send('content_block_stop',{index:0});send('message_delta',{delta:{stop_reason:'end_turn',stop_sequence:null},usage:{output_tokens:5}});send('message_stop',{});res.end();},modelStream.duration);
+     res.on('close',()=>clearTimeout(end));
+    } else if(modelStream.mode === 'keepalive') ticker=setInterval(()=>{res.write(': keepalive\n\n');send('ping',{});},300);
+    res.on('close',()=>clearInterval(ticker));
+   });return;
+  }
   if(req.url==='/v8/management/config/access/api-keys'){
    state.calls.push(req.method);if(supplied!=='fx653-manage'){res.writeHead(401);res.end('{}');return;}
    if(req.method==='GET'){res.end(JSON.stringify(state.keys));return;}
@@ -48,7 +66,7 @@ export async function managedLaunchFixture(t,{proxy=true,launches=true,refusePro
  const git=(...args)=>{const result=spawnSync('git',['-c','user.name=test','-c','[email protected]','-C',f.doc.projects.stack.root,...args],{env:gitEnv(env),encoding:'utf8'});assert.equal(result.status,0,'owned local git fixture succeeds');};
  git('init','-q','-b','main');fs.writeFileSync(join(f.doc.projects.stack.root,'file.txt'),'one\n');git('add','file.txt');git('commit','-q','-m','one');
  const adapter=join(root,'adapter.sh');fs.writeFileSync(adapter,'case "$MOSAIC_REQUEST" in *SILENT*) exec /usr/bin/sleep 30;; *) printf %s fx653-private-output; printf "done\\n" > "$MOSAIC_TURN_MARKER";; esac\n',{mode:0o644});
- const system={...loadSystem({env}),...(proxy?{proxy:{port,binary:'/fixture-unused',firstOutputSeconds:1}}:{})};
+ const system={...loadSystem({env}),...(proxy?{proxy:{port,binary:'/fixture-unused',firstOutputSeconds:modelStream?.seconds ?? 1}}:{})};
  const {trackers,...boot}=bootConfig({system,businessId:'acme',env});assert.equal(trackers,undefined);
  stderr=process.stderr.write;process.stderr.write=function(chunk,...args){if(String(chunk).startsWith('model-proxy: ')){proxyReceipts.push(String(chunk));const callback=args.find(a=>typeof a==='function');callback?.();return true;}return stderr.call(this,chunk,...args);};
  bus=await startBroker(boot);closers.push(()=>bus.close());

packages/harness/src/first-output.mjs

@@ -1,5 +1,7 @@
 import { writeFileSync, openSync, closeSync, linkSync, unlinkSync } from 'node:fs';
-import { join } from 'node:path';
+import { join, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { spawn } from 'node:child_process';
 // The host needs an output-exists signal, never the output or a model credential.
 export function firstOutputMarker(runDir) {
   let seen = false;
@@ -30,3 +32,68 @@ export function firstTurnMarker(runDir) {
     if (failed) throw new Error('first-turn marker publication failed');
   };
 }
+
+// Both managed adapters use this monitor to translate their native JSON event
+// streams back into the answer-only adapter contract. Startup, headers and
+// keepalives are not model output; substantive deltas and named tool calls count.
+export function modelEventStream({ harness, runDir }) {
+  if (!['pi', 'claude'].includes(harness)) throw new Error('unknown model event stream');
+  const mark = firstOutputMarker(runDir);
+  let pending = '', answer = null, failed = false;
+  const line = value => {
+    if (!value) return;
+    const event = JSON.parse(value);
+    if (harness === 'pi') {
+      const delta = event.type === 'message_update' && event.assistantMessageEvent;
+      if (delta && ['text_delta', 'thinking_delta', 'toolcall_delta'].includes(delta.type) && typeof delta.delta === 'string' && delta.delta.length) mark(delta.delta);
+      if (delta?.type === 'toolcall_start' && typeof delta.id === 'string' && delta.id.length && typeof delta.toolName === 'string' && delta.toolName.length) mark('output');
+      if (event.type === 'agent_end') {
+        const last = event.messages?.at(-1);
+        if (last?.role === 'assistant') {
+          failed = ['error', 'aborted'].includes(last.stopReason);
+          answer = (last.content ?? []).filter(part => part.type === 'text').map(part => part.text).join('\n');
+        }
+      }
+    } else {
+      const delta = event.type === 'stream_event' && event.event?.type === 'content_block_delta' && event.event.delta;
+      if (delta && [["text_delta", delta.text], ["thinking_delta", delta.thinking], ["input_json_delta", delta.partial_json]].some(([type, value]) => delta.type === type && typeof value === 'string' && value.length > 0)) mark('output');
+      const block = event.type === 'stream_event' && event.event?.type === 'content_block_start' && event.event.content_block;
+      if (block?.type === 'tool_use' && typeof block.id === 'string' && block.id.length && typeof block.name === 'string' && block.name.length) mark('output');
+      if (block && [["text", block.text], ["thinking", block.thinking]].some(([type, value]) => block.type === type && typeof value === 'string' && value.length)) mark('output');
+      if (event.type === 'result') { failed = event.is_error === true || event.subtype !== 'success'; answer = event.result ?? ''; }
+    }
+  };
+  return {
+    push(chunk) {
+      pending += chunk;
+      if (pending.length > 32 * 1024 * 1024) throw new Error('model event too large');
+      let i; while ((i = pending.indexOf('\n')) !== -1) { line(pending.slice(0, i)); pending = pending.slice(i + 1); }
+    },
+    finish() {
+      if (pending) line(pending);
+      if (failed || typeof answer !== 'string') throw new Error('model stream did not complete');
+      return answer;
+    },
+  };
+}
+
+// Invoked only by managed proxy adapter turns; the model CLI keeps its normal
+// process group and stderr, while stdout remains the final answer alone.
+if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
+  const [, , harness, runDir, command, ...args] = process.argv;
+  const stream = modelEventStream({ harness, runDir });
+  const child = spawn(command, args, { stdio: ['ignore', 'pipe', 'inherit'] });
+  let bad = false;
+  child.stdout.setEncoding('utf8');
+  child.stdout.on('data', chunk => {
+    if (bad) return;
+    try { stream.push(chunk); }
+    catch { bad = true; child.kill('SIGTERM'); }
+  });
+  child.on('error', () => { bad = true; });
+  child.on('close', (code, signal) => {
+    if (bad || signal || code !== 0) { process.exitCode = code || 1; return; }
+    try { const answer = stream.finish(); if (answer) process.stdout.write(answer + '\n'); }
+    catch { process.stderr.write('model event stream failed\n'); process.exitCode = 1; }
+  });
+}

packages/harness/src/runner.mjs

@@ -36,7 +36,7 @@ import { spawn } from "node:child_process";
 import { chmodSync, existsSync, mkdirSync, openSync, readFileSync, rmSync, closeSync } from "node:fs";
 import { createServer, connect } from "node:net";
 import { join } from "node:path";
-import { pathToFileURL } from "node:url";
+import { fileURLToPath, pathToFileURL } from "node:url";
 import { Client } from "../../bus/src/client.mjs";
 import { firstOutputMarker, firstTurnMarker } from './first-output.mjs';
 import { GUARD_CODES, FIXED_TEXTS } from "../../library/src/reply-protocol.mjs";
@@ -417,6 +417,7 @@ export function adapterEnv(session, env) {
     ...session.harnessEnv,
     // Row 66: after the module's variables, so the log is always the run's own.
     ...(session.runDir ? { MOSAIC_UNSLOP_LOG: unslopLog(session.runDir) } : {}),
+    ...(session.proxyFirstTurn ? { MOSAIC_MODEL_EVENT_MONITOR: fileURLToPath(new URL('./first-output.mjs', import.meta.url)), MOSAIC_FIRST_OUTPUT_DIR: session.runDir } : {}),
   };
 }
 

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

@@ -0,0 +1,70 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import * as output from '../src/first-output.mjs';
+
+for (const harness of ['pi', 'claude']) {
+  test(`${harness} counts a named tool call even when it has no input delta`, t => {
+    const root = fs.mkdtempSync(join(tmpdir(), 'stream797-tool-'));
+    t.after(() => fs.rmSync(root, { recursive: true, force: true }));
+    const stream = output.modelEventStream({ harness, runDir: root });
+    const event = harness === 'pi'
+      ? {type:'message_update',assistantMessageEvent:{type:'toolcall_start',id:'tool_1',toolName:'read'}}
+      : {type:'stream_event',event:{type:'content_block_start',content_block:{type:'tool_use',id:'tool_1',name:'read',input:{}}}};
+    stream.push(JSON.stringify(event)+'\n');
+    assert.equal(fs.readFileSync(join(root, 'proxy-first-output'), 'utf8'), 'output\n');
+  });
+  test(`${harness} marks real deltas, ignores startup events, and returns only the final answer`, t => {
+    const root = fs.mkdtempSync(join(tmpdir(), 'stream797-'));
+    t.after(() => fs.rmSync(root, { recursive: true, force: true }));
+    assert.equal(typeof output.modelEventStream, 'function', 'shared event monitor exists');
+    const stream = output.modelEventStream({ harness, runDir: root });
+    const send = event => stream.push(JSON.stringify(event) + '\n');
+    const file = join(root, 'proxy-first-output');
+    send(harness === 'pi' ? {type:'message_update', assistantMessageEvent:{type:'start'}} : {type:'system',subtype:'init'});
+    assert.equal(fs.existsSync(file), false);
+    const delta = harness === 'pi' ? {type:'message_update', assistantMessageEvent:{type:'thinking_delta',delta:'secret thinking'}} : {type:'stream_event',event:{type:'content_block_delta',delta:{type:'thinking_delta',thinking:'secret thinking'}}};
+    send(delta);
+    assert.equal(fs.readFileSync(file, 'utf8'), 'output\n');
+    send(harness === 'pi' ? {type:'agent_end',messages:[{role:'assistant',stopReason:'stop',content:[{type:'text',text:'final answer'}]}]} : {type:'result',subtype:'success',result:'final answer'});
+    assert.equal(stream.finish(), 'final answer');
+  });
+  test(`${harness} does not mark empty deltas, pings, or startup events`, t => {
+    const root = fs.mkdtempSync(join(tmpdir(), 'stream797-'));
+    t.after(() => fs.rmSync(root, { recursive: true, force: true }));
+    assert.equal(typeof output.modelEventStream, 'function', 'shared event monitor exists');
+    const stream = output.modelEventStream({ harness, runDir: root });
+    for (const e of [{type:'system'}, {type:'stream_event',event:{type:'ping'}}, {type:'message_update',assistantMessageEvent:{type:'text_delta',delta:''}}, {type:'stream_event',event:{type:'content_block_delta',delta:{type:'text_delta',text:''}}}, {type:'stream_event',event:{type:'content_block_delta',delta:{type:'text_delta',text:{}}}}]) stream.push(JSON.stringify(e)+'\n');
+    assert.equal(fs.existsSync(join(root, 'proxy-first-output')), false);
+  });
+}
+
+test('monitor forwards only the answer and keeps model events private', async t => {
+  const { spawn } = await import('node:child_process');
+  const { fileURLToPath } = await import('node:url');
+  const root = fs.mkdtempSync(join(tmpdir(), 'stream797-'));
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }));
+  const monitor = fileURLToPath(new URL('../src/first-output.mjs', import.meta.url));
+  const producer = `console.log(JSON.stringify({type:'message_update',assistantMessageEvent:{type:'text_delta',delta:'private'}}));console.log(JSON.stringify({type:'agent_end',messages:[{role:'assistant',stopReason:'stop',content:[{type:'text',text:'answer'}]}]}))`;
+  const child = spawn(process.execPath, [monitor, 'pi', root, process.execPath, '-e', producer], { stdio:['ignore','pipe','pipe'] });
+  let stdout='', stderr=''; child.stdout.on('data', c => stdout += c); child.stderr.on('data', c => stderr += c);
+  const code = await new Promise(resolve => child.on('close', resolve));
+  assert.equal(code, 0, stderr); assert.equal(stdout.trim(), 'answer');
+  assert.equal(fs.readFileSync(join(root, 'proxy-first-output'), 'utf8'), 'output\n');
+});
+
+test('an empty final answer does not create synthetic stdout or a marker', async t => {
+  const { spawn } = await import('node:child_process');
+  const { fileURLToPath } = await import('node:url');
+  const root = fs.mkdtempSync(join(tmpdir(), 'stream797-empty-'));
+  t.after(() => fs.rmSync(root, { recursive: true, force: true }));
+  const monitor = fileURLToPath(new URL('../src/first-output.mjs', import.meta.url));
+  const producer = `console.log(JSON.stringify({type:'result',subtype:'success',result:''}))`;
+  const child = spawn(process.execPath, [monitor, 'claude', root, process.execPath, '-e', producer], { stdio:['ignore','pipe','pipe'] });
+  let stdout=''; child.stdout.on('data', c => stdout += c);
+  const code = await new Promise(resolve => child.on('close', resolve));
+  assert.equal(code, 0); assert.equal(stdout, '');
+  assert.equal(fs.existsSync(join(root, 'proxy-first-output')), false);
+});