From 33c7b415f51a12972f9a6a7129bab4f70f5e25fb Mon Sep 17 00:00:00 2001 From: Nick <209239467+surlymochan@users.noreply.github.com> Date: Thu, 12 Nov 2026 07:30:14 +0000 Subject: [PATCH] test(stream): pairing invariant + real-stream fixtures + kill coverage (M14 slice 1, #185) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Enforce the pairing invariant at the sink: every id-bearing tool_start must be followed by exactly one tool_end with the same tool_call_id before agent_end. Breaches (unpaired/duplicate end, duplicate start, reused id, call still open at agent_end) are counted in stats.pairing_violations, listed on sink.pairingViolations, and written to the execution log as raw records (reason: pairing_violation) — never silently dropped, never fatal. checkPairing() exposes the same state machine as a pure function for replay checks. Fixtures: claude-error-killed + codex-killed streams (faithfully reconstructed — no live CLI credentials in this environment; see test/fixtures/agent-stream/README.md), covering error results, concurrent calls with out-of-order results, and killed runs. Killed calls close as abandoned with tool_call_id + agent_run_id and count in stats.events. --- docs/workflow-manual.md | 2 +- src/core/agent-stream.js | 140 ++++++++++-- test/agent-stream.test.js | 212 ++++++++++++++++++ test/fixtures/agent-stream/README.md | 26 +++ .../agent-stream/claude-error-killed.ndjson | 6 + .../fixtures/agent-stream/codex-killed.ndjson | 4 + 6 files changed, 374 insertions(+), 16 deletions(-) create mode 100644 test/fixtures/agent-stream/README.md create mode 100644 test/fixtures/agent-stream/claude-error-killed.ndjson create mode 100644 test/fixtures/agent-stream/codex-killed.ndjson diff --git a/docs/workflow-manual.md b/docs/workflow-manual.md index 08f07d2..4082a82 100644 --- a/docs/workflow-manual.md +++ b/docs/workflow-manual.md @@ -1139,7 +1139,7 @@ GET /api/events event: log, id: 未知的 CLI 事件类型一律降级为 `raw` 或被忽略，不抛错：CLI 升级不会让整条 workflow 失败。 -工具调用遵循一条生命周期约定：`tool_start` 带 `tool_call_id`（claude `tool_use.id`、codex `item.id`），配对的 `tool_end` 带同一个 id；没有 id 的调用不进入配对表——没有 id 就没有可对齐的对象。`agent_run_id` 给一次 agent 进程划范围，两个 atom 并发、各自的 CLI 都从 `toolu_1` 编号时靠它区分。`sink.close()` 会把仍未结束的调用补成 `status: "abandoned"`：被杀掉的那次运行（默认 300 秒超时、Ctrl-C、CLI 崩溃）永远发不出最后一条 `tool_result`，而没有终态的日志等于没有证据。 +工具调用遵循一条生命周期约定：`tool_start` 带 `tool_call_id`（claude `tool_use.id`、codex `item.id`），配对的 `tool_end` 带同一个 id；没有 id 的调用不进入配对表——没有 id 就没有可对齐的对象。`agent_run_id` 给一次 agent 进程划范围，两个 atom 并发、各自的 CLI 都从 `toolu_1` 编号时靠它区分。`sink.close()` 会把仍未结束的调用补成 `status: "abandoned"`：被杀掉的那次运行（默认 300 秒超时、Ctrl-C、CLI 崩溃）永远发不出最后一条 `tool_result`，而没有终态的日志等于没有证据。配对违规（无配对的 `tool_end`、重复的 `tool_end`、重复的 `tool_start`、同一 run 内复用 id、`agent_end` 到达时仍有调用未结束）会被显性报出：计入 `stats.pairing_violations`、列在 `sink.pairingViolations`、并以 `reason: "pairing_violation"` 的 `raw` 记录写进执行日志；违规事件本身仍保留，不静默丢弃、不中断运行。 ### 订阅 diff --git a/src/core/agent-stream.js b/src/core/agent-stream.js index 8efdd63..ed36591 100644 --- a/src/core/agent-stream.js +++ b/src/core/agent-stream.js @@ -12,6 +12,14 @@ * * Unknown line shapes degrade to `raw` instead of throwing: a CLI version bump must never be * able to break a workflow run. + * + * Pairing invariant (M14 slice 1, #185): inside one `agent_run_id` scope, every id-bearing + * `tool_start` must be followed by exactly one `tool_end` with the same `tool_call_id` before + * `agent_end`. The sink enforces it while the stream flows: a breach is counted in + * `stats.pairing_violations`, listed on `sink.pairingViolations`, and written to the execution + * log as a `raw` record (`reason: 'pairing_violation'`) — never silently dropped, and never + * fatal to the run. The offending event itself is still recorded. `checkPairing(events)` is the + * same state machine as a pure function over already-normalized events, for replay checks. */ import { createHash, randomBytes } from 'node:crypto'; import { StringDecoder } from 'node:string_decoder'; @@ -319,6 +327,91 @@ function compactUsage(raw) { return Object.keys(usage).length ? usage : null; } +export const PAIRING_VIOLATION_KINDS = Object.freeze([ + 'unpaired_tool_end', + 'duplicate_tool_end', + 'duplicate_tool_start', + 'reused_tool_call_id', + 'unpaired_at_agent_end', +]); + +/** + * The pairing state machine: one instance per agent run (`agent_run_id` scope). + * + * `observe(event)` returns the violations that event commits (usually none). Rules, all scoped + * to id-bearing calls only — without an id there is nothing to pair: + * - `tool_start` for an id already open → `duplicate_tool_start`; for an id already closed → + * `reused_tool_call_id` (an id names one call inside a run; reuse makes the log ambiguous). + * - `tool_end` for an id never started → `unpaired_tool_end`; for an id already closed → + * `duplicate_tool_end`. + * - `agent_end` while calls are still open → one `unpaired_at_agent_end` per open call. The + * calls stay open: only the sink's `close()` may end them, as `abandoned`. + * A violation is a fact about the stream, not a reason to stop it: the offending event is still + * recorded, and the tracker keeps going so one breach cannot hide the next. + */ +function createPairingTracker() { + const open = new Map(); + const closed = new Set(); + + return { + observe(event) { + // agent_end carries no id of its own, so it must be checked before the id guard below. + if (event?.event === 'agent_end') { + return [...open.keys()].map((id) => ({ kind: 'unpaired_at_agent_end', tool_call_id: id })); + } + const callId = callIdOf(event?.tool_call_id); + if (!callId) return []; + if (event.event === 'tool_start') { + if (open.has(callId)) { + return [{ kind: 'duplicate_tool_start', tool_call_id: callId }]; + } + if (closed.has(callId)) { + closed.delete(callId); + open.set(callId, { tool: event.tool ?? null, event_type: event.event_type ?? null }); + return [{ kind: 'reused_tool_call_id', tool_call_id: callId }]; + } + open.set(callId, { tool: event.tool ?? null, event_type: event.event_type ?? null }); + return []; + } + if (event.event === 'tool_end') { + if (open.has(callId)) { + open.delete(callId); + closed.add(callId); + return []; + } + if (closed.has(callId)) return [{ kind: 'duplicate_tool_end', tool_call_id: callId }]; + return [{ kind: 'unpaired_tool_end', tool_call_id: callId }]; + } + return []; + }, + /** Calls still open, as [tool_call_id, {tool, event_type}] pairs; clears them as closed. */ + drainOpen() { + const entries = [...open.entries()]; + for (const [callId] of entries) { + open.delete(callId); + closed.add(callId); + } + return entries; + }, + }; +} + +/** + * Pairing violations in an already-normalized event list, in stream order. Same state machine + * the sink runs live; used by replay checks (M14 slice 2 replays recorded streams through it). + * Each violation carries the offending event's `seq` when present. + */ +export function checkPairing(events) { + const tracker = createPairingTracker(); + const violations = []; + for (const event of events ?? []) { + for (const violation of tracker.observe(event)) { + violations.push({ ...violation, seq: event?.seq ?? null }); + } + } + return violations; +} + /** * Collect normalized events from one agent process and append them to the execution log. * @@ -341,10 +434,12 @@ export function createAgentEventSink(options = {}) { : `agent_run_${randomBytes(6).toString('hex')}`; const queue = []; - const stats = { lines: 0, events: 0, written: 0, dropped: 0, truncated: 0, write_errors: 0 }; + const stats = { lines: 0, events: 0, written: 0, dropped: 0, truncated: 0, write_errors: 0, pairing_violations: 0 }; const toolNames = new Map(); - // Only id-bearing calls are tracked: without an id there is nothing to pair a closing event to. - const openCalls = new Map(); + // Pairing state for this run only: two sinks sharing one log never share a tracker, so a CLI + // reusing ids across runs cannot produce a false violation in either. + const pairing = createPairingTracker(); + const pairingViolations = []; let seq = 0; let timer = null; let writing = Promise.resolve(); @@ -357,8 +452,10 @@ export function createAgentEventSink(options = {}) { seq += 1; stats.events += 1; if (normalized.truncated) stats.truncated += 1; - rememberCall(normalized); enqueue(buildRecord(normalized, seq)); + for (const violation of pairing.observe(normalized)) { + recordPairingViolation({ ...violation, seq }); + } } schedule(); }); @@ -371,14 +468,25 @@ export function createAgentEventSink(options = {}) { } } - function rememberCall(event) { - const callId = callIdOf(event.tool_call_id); - if (!callId) return; - if (event.event === 'tool_start') { - openCalls.set(callId, { tool: event.tool ?? null, event_type: event.event_type ?? null }); - } else if (event.event === 'tool_end') { - openCalls.delete(callId); - } + /** + * A pairing breach is surfaced three ways at once: counted in stats, listed for the caller, + * and written to the execution log as a `raw` record so the cockpit / --progress views can + * show it. The offending event was already enqueued above; this record is the flag on it. + * `sink.close()` discards its stats return in agent-invoke, so the log record is the one + * surface that always survives a real run. + */ + function recordPairingViolation(violation) { + stats.pairing_violations += 1; + pairingViolations.push({ ...violation, agent_run_id: agentRunId }); + seq += 1; + stats.events += 1; + enqueue(buildRecord({ + event: 'raw', + reason: 'pairing_violation', + violation: violation.kind, + tool_call_id: violation.tool_call_id, + text: `pairing violation (${violation.kind}) for tool_call_id ${violation.tool_call_id}`, + }, seq)); } function buildRecord(event, eventSeq) { @@ -393,7 +501,7 @@ export function createAgentEventSink(options = {}) { }; for (const key of [ 'event_type', 'tool', 'tool_call_id', 'text', 'args_preview', 'args_digest', 'usage', - 'exit_code', 'status', 'level', 'cost_usd', 'duration_ms', 'reason', 'truncated', + 'exit_code', 'status', 'level', 'cost_usd', 'duration_ms', 'reason', 'truncated', 'violation', ]) { const value = event[key]; if (value === undefined || value === null) continue; @@ -447,7 +555,7 @@ export function createAgentEventSink(options = {}) { */ async close() { splitter.flush(); - for (const [callId, call] of openCalls) { + for (const [callId, call] of pairing.drainOpen()) { seq += 1; stats.events += 1; enqueue(buildRecord({ @@ -458,13 +566,15 @@ export function createAgentEventSink(options = {}) { status: 'abandoned', }, seq)); } - openCalls.clear(); closed = true; await flush(); return stats; }, flush, stats, + get pairingViolations() { + return pairingViolations.map((violation) => ({ ...violation })); + }, get pending() { return queue.length; }, diff --git a/test/agent-stream.test.js b/test/agent-stream.test.js index 71de400..0e637d7 100644 --- a/test/agent-stream.test.js +++ b/test/agent-stream.test.js @@ -5,6 +5,7 @@ import { join, resolve } from 'node:path'; import { tmpdir } from 'node:os'; import { + checkPairing, createAgentEventSink, createLineSplitter, normalizeAgentEvent, @@ -15,6 +16,8 @@ import { readExecutionLog } from '../src/core/execution-log.js'; const FIXTURES = resolve(import.meta.dirname, 'fixtures', 'agent-stream'); const codexFixture = readFileSync(join(FIXTURES, 'codex.ndjson'), 'utf8'); const claudeFixture = readFileSync(join(FIXTURES, 'claude.ndjson'), 'utf8'); +const claudeErrorKilledFixture = readFileSync(join(FIXTURES, 'claude-error-killed.ndjson'), 'utf8'); +const codexKilledFixture = readFileSync(join(FIXTURES, 'codex-killed.ndjson'), 'utf8'); function makeProjectRoot() { const root = mkdtempSync(join(tmpdir(), 'openflow-agent-stream-')); @@ -34,6 +37,29 @@ function collect(fixture, adapter, options = {}) { return events; } +/** Every id-bearing tool_start has exactly one tool_end with the same id. */ +function assertPaired(events) { + const starts = new Map(); + const ends = new Map(); + for (const event of events) { + if (!event.tool_call_id) continue; + if (event.event === 'tool_start') starts.set(event.tool_call_id, (starts.get(event.tool_call_id) || 0) + 1); + if (event.event === 'tool_end') ends.set(event.tool_call_id, (ends.get(event.tool_call_id) || 0) + 1); + } + assert.deepEqual([...ends.entries()].sort(), [...starts.entries()].sort(), 'every tool_start has exactly one tool_end'); +} + +/** Push a whole fixture through a sink, close it, and return the sink, stats and log events. */ +async function replayFixture(projectRoot, fixture, adapter, options = {}) { + const sink = createAgentEventSink({ projectRoot, adapter, flushIntervalMs: 5, ...options }); + sink.push(fixture); + const stats = await sink.close(); + const log = await readExecutionLog(projectRoot, { limit: 50 }); + assert.equal(log.chain.ok, true); + const events = log.entries.filter((entry) => entry.kind === 'agent_event').reverse(); + return { sink, stats, events }; +} + describe('Agent stream', () => { describe('createLineSplitter', () => { test('reassembles a line whose chunk boundary falls inside it', () => { @@ -173,6 +199,53 @@ describe('Agent stream', () => { }); }); + describe('checkPairing', () => { + const start = (id, seq) => ({ seq, event: 'tool_start', tool: 'Bash', tool_call_id: id }); + const end = (id, seq) => ({ seq, event: 'tool_end', tool: 'Bash', tool_call_id: id }); + + test('a clean claude run has no violations', () => { + assert.deepEqual(checkPairing(collect(claudeFixture, 'claude_cli', { toolNames: new Map() })), []); + }); + + test('flags a tool_end whose tool_use was never seen', () => { + assert.deepEqual(checkPairing([end('toolu_ghost', 1)]), [ + { kind: 'unpaired_tool_end', tool_call_id: 'toolu_ghost', seq: 1 }, + ]); + }); + + test('flags a second tool_end for an already closed call', () => { + assert.deepEqual(checkPairing([start('toolu_1', 1), end('toolu_1', 2), end('toolu_1', 3)]), [ + { kind: 'duplicate_tool_end', tool_call_id: 'toolu_1', seq: 3 }, + ]); + }); + + test('flags a second tool_start while the first is still open', () => { + assert.deepEqual(checkPairing([start('toolu_1', 1), start('toolu_1', 2)]), [ + { kind: 'duplicate_tool_start', tool_call_id: 'toolu_1', seq: 2 }, + ]); + }); + + test('flags an id reused after its call already closed', () => { + assert.deepEqual(checkPairing([start('toolu_1', 1), end('toolu_1', 2), start('toolu_1', 3)]), [ + { kind: 'reused_tool_call_id', tool_call_id: 'toolu_1', seq: 3 }, + ]); + }); + + test('flags every call still open when agent_end arrives', () => { + assert.deepEqual(checkPairing([start('toolu_1', 1), start('toolu_2', 2), { seq: 3, event: 'agent_end' }]), [ + { kind: 'unpaired_at_agent_end', tool_call_id: 'toolu_1', seq: 3 }, + { kind: 'unpaired_at_agent_end', tool_call_id: 'toolu_2', seq: 3 }, + ]); + }); + + test('ignores calls without an id: nothing to pair them to', () => { + assert.deepEqual(checkPairing([ + { seq: 1, event: 'tool_start', tool: 'Bash', tool_call_id: null }, + { seq: 2, event: 'tool_end', tool: 'Bash', tool_call_id: null }, + ]), []); + }); + }); + describe('createAgentEventSink', () => { test('batches CLI output into hash-chained execution log entries', async () => { const projectRoot = makeProjectRoot(); @@ -280,6 +353,117 @@ describe('Agent stream', () => { } }); + test('pairs a claude error + killed run end to end, including concurrent calls', async () => { + const projectRoot = makeProjectRoot(); + try { + const { sink, stats, events } = await replayFixture(projectRoot, claudeErrorKilledFixture, 'claude_cli', { agentRunId: 'agent_run_test' }); + assert.equal(stats.pairing_violations, 0, 'clean replay must not raise a violation'); + assert.deepEqual(sink.pairingViolations, []); + assert.equal(stats.events, 10, '9 normalized events + 1 abandoned end'); + assert.equal(stats.written, 10); + assertPaired(events); + assert.deepEqual(events.map((entry) => entry.event), [ + 'agent_start', 'message_end', + 'tool_start', 'tool_end', + 'tool_start', 'tool_start', 'tool_end', 'tool_end', + 'tool_start', 'tool_end', + ]); + // Error result maps to an error end. + assert.equal(events[3].tool_call_id, 'toolu_10'); + assert.equal(events[3].status, 'error'); + // Concurrent calls pair even when their results arrive out of order. + assert.deepEqual(events.slice(4, 8).map((entry) => entry.tool_call_id), ['toolu_11', 'toolu_12', 'toolu_12', 'toolu_11']); + // The killed call is closed by close(), not left running. + assert.equal(events[9].tool_call_id, 'toolu_13'); + assert.equal(events[9].status, 'abandoned'); + assert.equal(events[9].agent_run_id, 'agent_run_test'); + assert.ok(events.every((entry) => entry.agent_run_id === 'agent_run_test')); + } finally { + rmSync(projectRoot, { recursive: true, force: true }); + } + }); + + test('pairs a killed codex run: the open command is abandoned on close', async () => { + const projectRoot = makeProjectRoot(); + try { + const { sink, stats, events } = await replayFixture(projectRoot, codexKilledFixture, 'codex_cli', { agentRunId: 'agent_run_test' }); + assert.equal(stats.pairing_violations, 0); + assert.deepEqual(sink.pairingViolations, []); + assert.equal(stats.events, 3, 'agent_start + tool_start + abandoned end'); + assertPaired(events); + assert.deepEqual(events.map((entry) => entry.event), ['agent_start', 'tool_start', 'tool_end']); + assert.equal(events[1].tool, 'shell'); + assert.equal(events[1].tool_call_id, 'item_1'); + assert.equal(events[2].status, 'abandoned'); + assert.equal(events[2].tool_call_id, 'item_1'); + assert.equal(events[2].agent_run_id, 'agent_run_test'); + } finally { + rmSync(projectRoot, { recursive: true, force: true }); + } + }); + + test('surfaces an orphan tool_end as a violation without dropping it', async () => { + const projectRoot = makeProjectRoot(); + try { + const orphan = '{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_ghost","content":"ok"}]}}\n'; + const { sink, stats, events } = await replayFixture(projectRoot, orphan, 'claude_cli', { agentRunId: 'agent_run_test' }); + assert.equal(stats.pairing_violations, 1); + assert.deepEqual(sink.pairingViolations, [ + { kind: 'unpaired_tool_end', tool_call_id: 'toolu_ghost', seq: 1, agent_run_id: 'agent_run_test' }, + ]); + assert.equal(stats.events, 2, 'the tool_end plus its violation flag'); + // The orphan end stays readable in the log; the violation is a flag on it, not a drop. + assert.equal(events[0].event, 'tool_end'); + assert.equal(events[0].tool_call_id, 'toolu_ghost'); + assert.equal(events[1].event, 'raw'); + assert.equal(events[1].reason, 'pairing_violation'); + assert.equal(events[1].violation, 'unpaired_tool_end'); + assert.equal(events[1].tool_call_id, 'toolu_ghost'); + assert.match(events[1].text, /unpaired_tool_end/); + } finally { + rmSync(projectRoot, { recursive: true, force: true }); + } + }); + + test('surfaces a duplicate tool_end as a violation without dropping either end', async () => { + const projectRoot = makeProjectRoot(); + try { + const toolUse = '{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_5","name":"Bash","input":{"command":"ls"}}]}}\n'; + const result = '{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_5","content":"ok"}]}}\n'; + const { sink, stats, events } = await replayFixture(projectRoot, toolUse + result + result, 'claude_cli', { agentRunId: 'agent_run_test' }); + assert.equal(stats.pairing_violations, 1); + assert.deepEqual(sink.pairingViolations, [ + { kind: 'duplicate_tool_end', tool_call_id: 'toolu_5', seq: 3, agent_run_id: 'agent_run_test' }, + ]); + assert.equal(events.filter((entry) => entry.event === 'tool_end').length, 2, 'both ends stay in the log'); + assert.equal(events.filter((entry) => entry.reason === 'pairing_violation').length, 1); + } finally { + rmSync(projectRoot, { recursive: true, force: true }); + } + }); + + test('flags a call still open at agent_end, then abandons it on close', async () => { + const projectRoot = makeProjectRoot(); + try { + const stream = [ + '{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_20","name":"Bash","input":{"command":"sleep 999"}}]}}', + '{"type":"result","subtype":"success","is_error":false,"duration_ms":1000,"total_cost_usd":0.001,"usage":{"input_tokens":10,"output_tokens":5}}', + ].join('\n') + '\n'; + const { sink, stats, events } = await replayFixture(projectRoot, stream, 'claude_cli', { agentRunId: 'agent_run_test' }); + assert.equal(stats.pairing_violations, 1); + assert.deepEqual(sink.pairingViolations, [ + { kind: 'unpaired_at_agent_end', tool_call_id: 'toolu_20', seq: 2, agent_run_id: 'agent_run_test' }, + ]); + assertPaired(events); + assert.deepEqual(events.map((entry) => entry.event), ['tool_start', 'agent_end', 'raw', 'tool_end']); + assert.equal(events[2].violation, 'unpaired_at_agent_end'); + assert.equal(events[3].status, 'abandoned'); + assert.equal(events[3].tool_call_id, 'toolu_20'); + } finally { + rmSync(projectRoot, { recursive: true, force: true }); + } + }); + test('scopes concurrent agent runs so their tool-call ids cannot collide', async () => { const projectRoot = makeProjectRoot(); try { @@ -303,5 +487,33 @@ describe('Agent stream', () => { rmSync(projectRoot, { recursive: true, force: true }); } }); + + test('replays the same claude ids through two runs without cross-talk or violations', async () => { + const projectRoot = makeProjectRoot(); + try { + const first = createAgentEventSink({ projectRoot, adapter: 'claude_cli', flushIntervalMs: 5, agentRunId: 'agent_run_a' }); + const second = createAgentEventSink({ projectRoot, adapter: 'claude_cli', flushIntervalMs: 5, agentRunId: 'agent_run_b' }); + first.push(claudeFixture); + second.push(claudeFixture); + const firstStats = await first.close(); + const secondStats = await second.close(); + + assert.equal(firstStats.pairing_violations, 0); + assert.equal(secondStats.pairing_violations, 0); + assert.deepEqual(first.pairingViolations, []); + assert.deepEqual(second.pairingViolations, []); + + const log = await readExecutionLog(projectRoot, { limit: 30 }); + assert.equal(log.chain.ok, true); + const events = log.entries.filter((entry) => entry.kind === 'agent_event'); + for (const runId of ['agent_run_a', 'agent_run_b']) { + const perRun = events.filter((entry) => entry.agent_run_id === runId); + assert.equal(perRun.length, 7, `${runId} keeps its own 7 events`); + assertPaired(perRun); + } + } finally { + rmSync(projectRoot, { recursive: true, force: true }); + } + }); }); }); diff --git a/test/fixtures/agent-stream/README.md b/test/fixtures/agent-stream/README.md new file mode 100644 index 0000000..0db05d7 --- /dev/null +++ b/test/fixtures/agent-stream/README.md @@ -0,0 +1,26 @@ +# agent-stream fixtures — provenance + +`claude.ndjson` / `codex.ndjson` (pre-existing, #183): minimal happy-path streams +in real CLI shapes — claude `--output-format stream-json` +(`system`/`assistant`/`user`/`result` lines) and codex `--json` +(`thread.started`/`item.started`/`item.completed`/`turn.completed`). + +`claude-error-killed.ndjson` / `codex-killed.ndjson` (M14 slice 1, #185): +**faithfully reconstructed, not recorded from a live CLI run.** This environment +has no usable live CLI to record from (the installed codex CLI reports +"Not logged in"; no claude CLI is installed and no CLI credentials are +available to this worker). Line shapes, field names and orderings were copied +from the real stream formats above and from the CLIs' documented stream-json / +`--json` output: + +- claude: an `assistant` line can carry text plus one or more `tool_use` + blocks; each `tool_result` arrives later on a `user` line and carries only + `tool_use_id` (plus `is_error` on failure); results for concurrent calls + may arrive in any order. A killed run ends mid-stream: no `result` line. +- codex: each tool call is an `item.started` / `item.completed` pair keyed by + `item.id`; a killed run ends after `item.started` with no `item.completed` + and no `turn.completed`. + +Slice 2 (#184 acceptance 1) replaces / extends these with ≥3 streams recorded +from real CLI runs; until then no test in this directory claims a fixture is a +verbatim recording. diff --git a/test/fixtures/agent-stream/claude-error-killed.ndjson b/test/fixtures/agent-stream/claude-error-killed.ndjson new file mode 100644 index 0000000..d75ab50 --- /dev/null +++ b/test/fixtures/agent-stream/claude-error-killed.ndjson @@ -0,0 +1,6 @@ +{"type":"system","subtype":"init","session_id":"sess_2","tools":["Bash","Read","Grep"]} +{"type":"assistant","message":{"id":"msg_10","content":[{"type":"text","text":"Running the failing suite first."},{"type":"tool_use","id":"toolu_10","name":"Bash","input":{"command":"npm test"}}]}} +{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_10","is_error":true,"content":"exit 1: 3 failing"}]}} +{"type":"assistant","message":{"id":"msg_11","content":[{"type":"tool_use","id":"toolu_11","name":"Read","input":{"file_path":"test/agent-stream.test.js"}},{"type":"tool_use","id":"toolu_12","name":"Grep","input":{"pattern":"pairing","path":"src/core"}}]}} +{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_12","content":"src/core/agent-stream.js:3 matches"},{"type":"tool_result","tool_use_id":"toolu_11","content":"// agent-stream tests"}]}} +{"type":"assistant","message":{"id":"msg_12","content":[{"type":"tool_use","id":"toolu_13","name":"Bash","input":{"command":"sleep 999"}}]}} diff --git a/test/fixtures/agent-stream/codex-killed.ndjson b/test/fixtures/agent-stream/codex-killed.ndjson new file mode 100644 index 0000000..5cdecf0 --- /dev/null +++ b/test/fixtures/agent-stream/codex-killed.ndjson @@ -0,0 +1,4 @@ +{"type":"thread.started","thread_id":"thread_9d1a"} +{"type":"item.started","item":{"id":"item_0","type":"reasoning","text":"Planning the build"}} +{"type":"item.completed","item":{"id":"item_0","type":"reasoning","text":"Planning the build"}} +{"type":"item.started","item":{"id":"item_1","type":"command_execution","command":"npm run build"}}