fix(cli): name a hung claude turn stuck, not started on retries (TASK-710)

cli/__tests__/adapters.claude.test.mjs

@@ -120,7 +120,7 @@ describe('claude adapter — spawn()', () => {
     expect(calls[0].args[0]).toBe('-p');
     expect(calls[0].args[1]).toContain('hello world');
     expect(calls[0].args.slice(2)).toEqual(
-      ['--output-format', 'text', '--resume', 'sid-123', '--model', 'test-model'],
+      ['--output-format', 'stream-json', '--verbose', '--include-partial-messages', '--resume', 'sid-123', '--model', 'test-model'],
     );
     // Explicit: must NOT use --session-id on resume path — the whole point.
     expect(calls[0].args).not.toContain('--session-id');
@@ -177,6 +177,28 @@ describe('claude adapter — spawn()', () => {
     expect(res.text).toBe('recovered');
   });
 
+  test('a first turn whose result is an error is a failure, not a reply, even at exit 0', async () => {
+    const errorResult = '{"type":"result","subtype":"success","is_error":true,"result":"API Error: 500 upstream down"}\n';
+    const { impl, calls } = makeSpawnImpl({ stdout: errorResult, code: 0 });
+    await expect(
+      claude.spawn('x', { environment: { model: 'test-model' }, sessionId: 'sid-err', _spawnImpl: impl }),
+    ).rejects.toThrow(/API Error: 500 upstream down/);
+    expect(calls).toHaveLength(1);
+  });
+
+  test('a recovery turn whose result is an error is a failure, not a reply', async () => {
+    let call = 0;
+    const impl = () => {
+      call += 1;
+      if (call === 1) return fakeChild({ stderr: 'Error: Session ID abc-123 is already in use.', code: 1 });
+      return fakeChild({ stdout: '{"type":"result","is_error":true,"result":"API Error: 529 overloaded"}\n', code: 0 });
+    };
+    await expect(
+      claude.spawn('x', { environment: { model: 'test-model' }, sessionId: 'abc-123', _spawnImpl: impl }),
+    ).rejects.toThrow(/API Error: 529 overloaded/);
+    expect(call).toBe(2);
+  });
+
   test('does NOT self-heal on generic spawn errors — only session-ID poison-pills', async () => {
     const { impl, calls } = makeSpawnImpl({ stderr: 'some other error', code: 1 });
     await expect(

cli/__tests__/first-output-watchdog.test.mjs

@@ -38,8 +38,10 @@ const stubAdapter = (await import('../src/lib/adapters/stub.js')).default;
 const claude = (await import('../src/lib/adapters/claude.js')).default;
 const {
   consoleLocked, spawnStalledError, watchFirstOutput, firstOutputWindowMs, DEFAULT_FIRST_OUTPUT_WINDOW_MS, physicalPath,
-  killStallSurvivors,
+  killStallSurvivors, isModelProgressLine,
 } = await import('../src/lib/first-output-watchdog.js');
+const { claudeResultText, claudeFailureTail } = await import('../src/lib/adapters/claude.js');
+const watchStream = (args) => watchFirstOutput({ ...args, modelStream: true });
 const { classifySpawnFailure, spawnRetryPolicy, SPAWN_FAILURE_CLASS } = await import('../src/lib/spawn-retry.js');
 
 afterAll(() => fs.rmSync(sessionsTmpDir, { recursive: true, force: true }));
@@ -86,10 +88,11 @@ describe('the claude adapter under the first-output watchdog', () => {
     expect(classifySpawnFailure(error)).toBe(SPAWN_FAILURE_CLASS.STALLED);
   });
 
-  test('a missing transcript alone does not kill: the path may be wrong, so it takes a locked console', async () => {
+  test('a model reply is not killed when the transcript path is wrong', async () => {
     const { value, error, child } = await spawnClaude((proc) => {
+      proc.stdout.emit('data', Buffer.from('{"type":"assistant","message":{}}\n'));
       setTimeout(() => {
-        proc.stdout.emit('data', Buffer.from('done quietly'));
+        proc.stdout.emit('data', Buffer.from('{"type":"result","result":"done quietly"}\n'));
         proc.emit('close', 0);
       }, 150);
     }, { windowMs: 50, locked: false });
@@ -247,8 +250,9 @@ describe('the claude adapter under the first-output watchdog', () => {
   test('a slow spawn that is talking is not killed', async () => {
     const { value, error, child } = await spawnClaude((proc) => {
       proc.stderr.emit('data', Buffer.from('warming up'));
+      proc.stdout.emit('data', Buffer.from('{"type":"assistant","message":{}}\n'));
       setTimeout(() => {
-        proc.stdout.emit('data', Buffer.from('the answer'));
+        proc.stdout.emit('data', Buffer.from('{"type":"result","result":"the answer"}\n'));
         proc.emit('close', 0);
       }, 150);
     }, { windowMs: 50 });
@@ -457,3 +461,139 @@ describe('performRun — a stalled turn', () => {
     }
   });
 });
+
+describe('the model-stream watchdog (stream-json, TASK-710)', () => {
+  const transcriptWithFirstRow = () => {
+    const file = path.join(sessionsTmpDir, `transcript-${Math.random().toString(36).slice(2)}.jsonl`);
+    fs.writeFileSync(file, '{"type":"user"}\n');
+    return file;
+  };
+  const streamProc = () => {
+    const proc = new EventEmitter();
+    proc.stdout = new EventEmitter();
+    proc.stderr = new EventEmitter();
+    return proc;
+  };
+
+  test('isModelProgressLine counts model events and not system events or non-JSON', () => {
+    expect(isModelProgressLine('{"type":"assistant","message":{}}')).toBe(true);
+    expect(isModelProgressLine('{"type":"stream_event","event":{}}')).toBe(true);
+    expect(isModelProgressLine('{"type":"system","subtype":"init"}')).toBe(false);
+    expect(isModelProgressLine('{"type":"system","subtype":"api_retry","attempt":1}')).toBe(false);
+    expect(isModelProgressLine('not json')).toBe(false);
+  });
+
+  test('a provider that only retries is named stuck after the first transcript row, not left to the run timeout', async () => {
+    const proc = streamProc();
+    const transcriptPath = transcriptWithFirstRow();
+    const onStall = jest.fn();
+    watchStream({ proc, windowMs: 50, transcriptPath, modelStream: true, onStall, lockProbe: async () => false });
+    proc.stdout.emit('data', Buffer.from('{"type":"system","subtype":"init"}\n'));
+    proc.stdout.emit('data', Buffer.from('{"type":"system","subtype":"api_retry","attempt":1}\n'));
+    // Transcript growth alone must not count as life in stream mode.
+    fs.appendFileSync(transcriptPath, '{"type":"user","retry":true}\n');
+    await new Promise((resolve) => setTimeout(resolve, 120));
+    expect(onStall).toHaveBeenCalledTimes(1);
+    expect(onStall.mock.calls[0][0]).toBe('stuck: no model output inside the window');
+  });
+
+  test('a stderr warning at spawn is not life in stream mode', async () => {
+    const proc = streamProc();
+    const transcriptPath = transcriptWithFirstRow();
+    const onStall = jest.fn();
+    watchStream({ proc, windowMs: 50, transcriptPath, onStall, lockProbe: async () => false });
+    proc.stderr.emit('data', Buffer.from('[claude-code:unrecognized_model] {"model":"x"}\n'));
+    await new Promise((resolve) => setTimeout(resolve, 120));
+    expect(onStall).toHaveBeenCalledWith('stuck: no model output inside the window');
+  });
+
+  test('a model event split across chunks still counts as life', async () => {
+    const proc = streamProc();
+    const transcriptPath = transcriptWithFirstRow();
+    const onStall = jest.fn();
+    const w = watchStream({ proc, windowMs: 50, transcriptPath, modelStream: true, onStall, lockProbe: async () => false });
+    proc.stdout.emit('data', Buffer.from('{"type":"assi'));
+    proc.stdout.emit('data', Buffer.from('stant","message":{}}\n'));
+    await new Promise((resolve) => setTimeout(resolve, 120));
+    w.stop();
+    expect(onStall).not.toHaveBeenCalled();
+  });
+
+  test('a locked console is still named as such in stream mode', async () => {
+    const proc = streamProc();
+    const transcriptPath = transcriptWithFirstRow();
+    const onStall = jest.fn();
+    watchStream({ proc, windowMs: 50, transcriptPath, modelStream: true, onStall, lockProbe: async () => true });
+    await new Promise((resolve) => setTimeout(resolve, 120));
+    expect(onStall).toHaveBeenCalledWith('console locked');
+  });
+
+  test('a missing transcript does not hide a silent provider in stream mode', async () => {
+    const proc = streamProc();
+    const onStall = jest.fn();
+    watchStream({ proc, windowMs: 50, transcriptPath: path.join(sessionsTmpDir, 'absent.jsonl'), onStall, lockProbe: async () => false });
+    await new Promise((resolve) => setTimeout(resolve, 120));
+    expect(onStall).toHaveBeenCalledWith('stuck: no model output inside the window');
+  });
+
+  test('model output that arrives while the lock probe runs still counts as life', async () => {
+    const proc = streamProc();
+    const transcriptPath = transcriptWithFirstRow();
+    const onStall = jest.fn();
+    let releaseProbe;
+    const lockProbe = () => new Promise((resolve) => { releaseProbe = () => resolve(false); });
+    watchStream({ proc, windowMs: 50, transcriptPath, onStall, lockProbe });
+    await new Promise((resolve) => setTimeout(resolve, 60));
+    proc.stdout.emit('data', Buffer.from('{"type":"assistant","message":{}}\n'));
+    releaseProbe();
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    expect(onStall).not.toHaveBeenCalled();
+  });
+});
+
+describe('claudeResultText', () => {
+  test('returns the result event text from stream-json output', () => {
+    const stdout = [
+      '{"type":"system","subtype":"init"}',
+      '{"type":"assistant","message":{"content":[{"type":"text","text":"partial"}]}}',
+      '{"type":"result","subtype":"success","result":"final answer\\n","is_error":false}',
+    ].join('\n');
+    expect(claudeResultText(stdout)).toBe('final answer');
+  });
+
+  test('a result flagged is_error throws instead of returning its text as the reply', () => {
+    const stdout = '{"type":"result","subtype":"success","is_error":true,"result":"API Error: 500"}';
+    expect(() => claudeResultText(stdout)).toThrow(/API Error: 500/);
+  });
+
+  test('returns plain text output as it arrived when there is no JSON at all', () => {
+    expect(claudeResultText('  done quietly \n')).toBe('done quietly');
+  });
+
+  test('a turn with no result event returns its assistant text, never the raw NDJSON', () => {
+    const stdout = [
+      '{"type":"system","subtype":"init"}',
+      '{"type":"assistant","message":{"content":[{"type":"text","text":"half "}]}}',
+      '{"type":"assistant","message":{"content":[{"type":"text","text":"an answer"}]}}',
+    ].join('\n');
+    expect(claudeResultText(stdout)).toBe('half an answer');
+    expect(claudeResultText('{"type":"system","subtype":"init"}')).toBe('');
+  });
+});
+
+describe('claudeFailureTail', () => {
+  test('keeps the terminal error past the init events that precede it', () => {
+    const init = `{"type":"system","subtype":"init","tools":["${'x'.repeat(5000)}"]}`;
+    const stdout = [init, '{"type":"stream_event","event":{}}', '{"type":"result","is_error":true,"result":"Claude AI usage limit reached"}'].join('\n');
+    expect(claudeFailureTail(stdout)).toBe('Claude AI usage limit reached');
+  });
+
+  test('without a result event, keeps the non-system lines and drops init and stream deltas', () => {
+    const stdout = [
+      '{"type":"system","subtype":"init"}',
+      '{"type":"stream_event","event":{}}',
+      'Error: usage limit reached',
+    ].join('\n');
+    expect(claudeFailureTail(stdout)).toBe('Error: usage limit reached');
+  });
+});

cli/src/lib/adapters/claude.js

@@ -494,10 +494,65 @@ const publicSkillPromptArgs = async (skillSources) => {
 
 // `ownGroup` is set only under the run supervisor, which kills the child's
 // whole group; unsupervised, the spawn options are exactly what they were.
+// Stream-json so each model event is a sign of life for the watchdog. The
+// partial messages put thinking deltas on the stream too, so a long think is
+// visible while it runs. A hung provider writes only `system` retry events,
+// and the watchdog does not count those.
+const STREAM_ARGS = ['--output-format', 'stream-json', '--verbose', '--include-partial-messages'];
+
+const parseEvent = (line) => {
+  try {
+    const event = JSON.parse(line);
+    return event && typeof event === 'object' ? event : null;
+  } catch {
+    return null;
+  }
+};
+
+// The reply is the `result` event's text. A turn that reaches no result event
+// (killed mid-stream, exit 0) gets the text of its assistant events, never the
+// raw NDJSON. Output with no JSON lines at all is plain text and returned as it
+// arrived. A result flagged `is_error` is a failed turn, whatever its exit code,
+// so it is raised here, outside the parse guard, and never returned as a reply.
+export const claudeResultText = (stdout, { credentials = [] } = {}) => {
+  const text = String(stdout || '');
+  const events = text.split('\n').map(parseEvent);
+  if (!events.some(Boolean)) return text.trim();
+  for (let i = events.length - 1; i >= 0; i -= 1) {
+    if (events[i]?.type !== 'result') continue;
+    const reply = String(events[i].result ?? '').trim();
+    if (events[i].is_error === true) throw adapterFailure('claude', reply, { credentials });
+    return reply;
+  }
+  return events
+    .filter((event) => event?.type === 'assistant')
+    .flatMap((event) => event.message?.content || [])
+    .filter((block) => block?.type === 'text')
+    .map((block) => block.text)
+    .join('')
+    .trim();
+};
+
+// The failure tail for a non-zero exit. Stream events are mostly system init
+// and status lines, which would push the terminal error past the 2000-char cut,
+// so keep the result text, or the lines that are not system or stream events.
+export const claudeFailureTail = (stdout) => {
+  const text = String(stdout || '');
+  const lines = text.split('\n');
+  const events = lines.map(parseEvent);
+  if (!events.some(Boolean)) return text.trim();
+  const result = [...events].reverse().find((event) => event?.type === 'result');
+  if (String(result?.result ?? '').trim()) return String(result.result).trim();
+  return lines
+    .filter((_, i) => !['system', 'stream_event'].includes(events[i]?.type))
+    .join('\n')
+    .trim();
+};
+
 const runClaude = ({
   cmd, args, cwd, env, timeoutMs, credentials = [], spawnImpl = childSpawn, onSpawn, controls = {}, ownGroup = false,
   transcriptPath = null, firstOutputMs = firstOutputWindowMs(), lockProbe, stallKillGraceMs = STALL_KILL_GRACE_MS,
-  processTable = readProcessTable, survivorIo,
+  processTable = readProcessTable, survivorIo, modelStream = true,
 }) => new Promise((resolve, reject) => {
   const proc = spawnInOwnGroup(spawnImpl, cmd, args, {
     cwd,
@@ -529,6 +584,7 @@ const runClaude = ({
     proc,
     windowMs: firstOutputMs,
     transcriptPath,
+    modelStream,
     lockProbe,
     onStall: (reason) => {
       stalled = spawnStalledError({ label: 'claude', windowMs: firstOutputMs, reason });
@@ -584,7 +640,7 @@ const runClaude = ({
       // reduced to one named field. The status, when the tail names one, is
       // attached as `error.status` so a refusal classifies by field rather than
       // by regex.
-      const detail = [stderr.trim(), stdout.trim()].filter(Boolean).join(' | ');
+      const detail = [stderr.trim(), claudeFailureTail(stdout)].filter(Boolean).join(' | ');
       return reject(adapterFailure('claude', detail, {
         credentials,
         exitCode: code,
@@ -1161,7 +1217,7 @@ export default {
       ...(seatModel ? ['--model', seatModel] : []),
       ...(ctx.environment?.effort ? ['--effort', String(ctx.environment.effort)] : []),
     ];
-    const baseArgs = ['-p', fullPrompt, '--output-format', 'text', sessionFlag, sessionId, ...modelArgs];
+    const baseArgs = ['-p', fullPrompt, ...STREAM_ARGS, sessionFlag, sessionId, ...modelArgs];
 
     let mountedSkills = [];
     let skillMount = { mounted: [], conflicted: [] };
@@ -1274,7 +1330,7 @@ export default {
       const contextTokens = claudeContextTokens({
         cwd: physicalPath(ctx.cwd), sessionId, configDir: claudeConfigDirFor({ ...env, HOME: childHome || env?.HOME }),
       });
-      return { text: stdout.trim(), newSessionId: sessionId, contextTokens };
+      return { text: claudeResultText(stdout, { credentials: spawnCredentials(spawnCtx) }), newSessionId: sessionId, contextTokens };
     } catch (err) {
       // A public seat's refusal is not a session-store failure, whatever its
       // message happens to contain: it must reach the run loop, never the
@@ -1295,7 +1351,7 @@ export default {
           memoryWithheld: ctx.memoryWithheld === true,
         });
         const retryBase = ['-p', freshPrompt,
-        '--output-format', 'text', '--session-id', freshId, ...modelArgs];
+        ...STREAM_ARGS, '--session-id', freshId, ...modelArgs];
         const retryCtx = {
           ...ctx,
           mcpConfigPath: mcpConfig?.file || null,
@@ -1337,7 +1393,7 @@ export default {
           cwd: physicalPath(ctx.cwd), sessionId: freshId,
           configDir: claudeConfigDirFor({ ...retry.env, HOME: retry.childHome || retry.env?.HOME }),
         });
-        return { text: stdout.trim(), newSessionId: freshId, contextTokens };
+        return { text: claudeResultText(stdout, { credentials: spawnCredentials(retryCtx) }), newSessionId: freshId, contextTokens };
       }
       throw err;
     } finally {