diff --git a/docs/workflow-manual.md b/docs/workflow-manual.md index 4b8a06a..08f07d2 100644 --- a/docs/workflow-manual.md +++ b/docs/workflow-manual.md @@ -1107,8 +1107,8 @@ GET /api/events event: log, id: | event | 来源 | 关键字段 | |---|---|---| | `agent_start` | codex `thread.started` / claude `system:init` | `adapter` | -| `tool_start` | codex `item.started`：`command_execution`(→ `tool`=`shell`)、`mcp_tool_call`、`file_change`、`web_search`；claude `tool_use` | `event_type`(CLI 原始 item 类型), `tool`, `args_preview`(≤160 字符), `args_digest`(sha256 前 16 位) | -| `tool_end` | codex `item.completed` 的同四种 item 类型（`event_type` 区分）；claude 侧没有配对事件 | `event_type`, `tool`, `exit_code`, `status` | +| `tool_start` | codex `item.started`：`command_execution`(→ `tool`=`shell`)、`mcp_tool_call`、`file_change`、`web_search`；claude `tool_use` | `event_type`(CLI 原始 item 类型), `tool`, `tool_call_id`, `args_preview`(≤160 字符), `args_digest`(sha256 前 16 位) | +| `tool_end` | codex `item.completed` 的同四种 item 类型（`event_type` 区分）；claude `tool_result`(`event_type`=`tool_result`)，`tool` 用 `tool_use_id` 回查先前那条 `tool_use`；sink 关闭时仍未结束的调用补一条 `abandoned` | `event_type`, `tool`, `tool_call_id`, `exit_code`, `status`(`completed`/`error`/`abandoned`) | | `message_end` | codex `agent_message` / claude assistant `text` | `text`(≤2048 字符，超出以 `…` 结尾) | | `agent_end` | codex `turn.completed` / claude `result` | `usage`{`input`,`output`,`cache_read`,`cache_write`}, `cost_usd`, `duration_ms` | | `raw` | 无法识别、无法解析、或超长被截断的行；CLI 报错（`turn.failed` / `error`） | `reason`(`invalid_json` / `non_object_payload` / `line_truncated`), `event_type`, `level`(`error`), `truncated` | @@ -1123,9 +1123,11 @@ GET /api/events event: log, id: "phase": "implementation", "atom": "I6a.tdd.run", "adapter": "codex_cli", + "agent_run_id": "agent_run_5f0c1a9d2b47", "seq": 3, "event_type": "command_execution", "tool": "shell", + "tool_call_id": "item_1", "args_preview": "npm test", "args_digest": "9f2c1e0b7d4a5c88", "change_id": "chg-124", @@ -1137,6 +1139,8 @@ 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`，而没有终态的日志等于没有证据。 + ### 订阅 ```javascript @@ -1174,7 +1178,7 @@ CHANGE_ID= xflow workflow run workflows/yolo.yaml --progress ### 性能约束 -- 写入：每 200ms 合并一次，单条 `text` ≤2048 字符、`args_preview` ≤160 字符；队列上限 500，超出丢最旧条目并计入 `stats.dropped`。 +- 写入：每 200ms 合并一次，单条 `text` ≤2048 字符、`args_preview` ≤160 字符；队列上限 500，超出丢最旧条目并计入 `stats.dropped`。关闭时补写的 `abandoned` 也走同一条队列、同一个上限，并计入 `stats.events`。 - 读取：只从字节偏移往后读，单次 ≤4MB；未写完的尾行留到下次；游标只停在换行边界。 - 哈希链：追加时从文件尾部 16KB 起求 `prev_hash`；若窗口里没有完整行（单条记录大于 16KB），窗口按 4 倍扩大直到找到为止，否则会写出 `prev_hash: null` 把链断掉。缓存按 `(size, mtimeMs)` 失效，避免同尺寸的轮转或截断命中陈旧哈希。 - 订阅：每个 SSE 连接各自持有一个 200ms 的 tailer，一次 tick 是“stat + 从上次偏移读一小段”，单个看板标签页可以忽略；若将来有大量客户端同时观看同一 run，更省的形状是**一个共享 tailer 广播给所有连接**（当前版本未做）。 diff --git a/src/core/agent-stream.js b/src/core/agent-stream.js index e6fa894..8efdd63 100644 --- a/src/core/agent-stream.js +++ b/src/core/agent-stream.js @@ -13,7 +13,7 @@ * Unknown line shapes degrade to `raw` instead of throwing: a CLI version bump must never be * able to break a workflow run. */ -import { createHash } from 'node:crypto'; +import { createHash, randomBytes } from 'node:crypto'; import { StringDecoder } from 'node:string_decoder'; import { appendExecutionLog } from './execution-log.js'; @@ -127,7 +127,8 @@ export function normalizeAgentEvents(raw, options = {}) { } const family = detectFamily(payload, adapter); - const mapped = (family === 'claude' ? mapClaudePayload : mapCodexPayload)(payload) + const toolNames = options.toolNames instanceof Map ? options.toolNames : null; + const mapped = (family === 'claude' ? mapClaudePayload : mapCodexPayload)(payload, { toolNames }) .filter((event) => event && event.event); if (!mapped.length) return []; return mapped.map((event) => ({ seq, ...event })); @@ -163,6 +164,7 @@ function mapCodexPayload(payload) { function mapCodexItem(item, phase) { if (!item || typeof item !== 'object') return []; const itemType = String(item.type || ''); + const callId = callIdOf(item.id); if (itemType === 'agent_message') { return phase === 'end' ? [{ event: 'message_end', event_type: itemType, text: truncateText(item.text) }] : []; } @@ -173,31 +175,31 @@ function mapCodexItem(item, phase) { if (itemType === 'command_execution') { if (phase === 'start') { const command = Array.isArray(item.command) ? item.command.join(' ') : String(item.command || ''); - return [toolStart('shell', command, itemType)]; + return [toolStart('shell', command, itemType, callId)]; } - return [toolEnd('shell', itemType, item.exit_code, item.status)]; + return [toolEnd('shell', itemType, item.exit_code, item.status, callId)]; } if (itemType === 'mcp_tool_call') { const tool = String(item.name || item.tool || 'mcp_tool'); - if (phase === 'start') return [toolStart(tool, stringifyArgs(item.arguments ?? item.input ?? item.args), itemType)]; - return [toolEnd(tool, itemType, item.exit_code, item.status)]; + if (phase === 'start') return [toolStart(tool, stringifyArgs(item.arguments ?? item.input ?? item.args), itemType, callId)]; + return [toolEnd(tool, itemType, item.exit_code, item.status, callId)]; } if (itemType === 'file_change') { const paths = Array.isArray(item.changes) ? item.changes.map((change) => change && change.path).filter(Boolean).join(', ') : String(item.path || ''); - if (phase === 'start') return [toolStart('file_change', paths, itemType)]; - return [toolEnd('file_change', itemType, item.exit_code, item.status)]; + if (phase === 'start') return [toolStart('file_change', paths, itemType, callId)]; + return [toolEnd('file_change', itemType, item.exit_code, item.status, callId)]; } if (itemType === 'web_search') { return phase === 'start' - ? [toolStart('web_search', String(item.query || ''), itemType)] - : [toolEnd('web_search', itemType, null, item.status)]; + ? [toolStart('web_search', String(item.query || ''), itemType, callId)] + : [toolEnd('web_search', itemType, null, item.status, callId)]; } return []; } -function mapClaudePayload(payload) { +function mapClaudePayload(payload, { toolNames } = {}) { const type = String(payload.type || ''); if (type === 'system') { return payload.subtype === 'init' ? [{ event: 'agent_start', event_type: 'system:init' }] : []; @@ -218,15 +220,28 @@ function mapClaudePayload(payload) { } else if (block.type === 'tool_use') { // Preserve the order the model produced: text first, then the tool call it announced. flushText(); - events.push(toolStart(String(block.name || 'tool'), stringifyArgs(block.input), 'tool_use')); + const tool = String(block.name || 'tool'); + const callId = callIdOf(block.id); + if (toolNames && callId) toolNames.set(callId, tool); + events.push(toolStart(tool, stringifyArgs(block.input), 'tool_use', callId)); } } flushText(); return events; } if (type === 'user') { - // Tool results echo output the tool_start/tool_end pair already accounted for. - return []; + // Claude answers a tool call with a `tool_result` block that only carries `tool_use_id`, so the + // tool name comes from the `tool_use` line seen earlier (the sink owns that map). Codex users + // never reach this branch: its results arrive as `item.completed` instead. + const content = Array.isArray(payload.message?.content) ? payload.message.content : []; + const events = []; + for (const block of content) { + if (!block || block.type !== 'tool_result') continue; + const callId = callIdOf(block.tool_use_id); + const tool = callId && toolNames ? (toolNames.get(callId) ?? null) : null; + events.push(toolEnd(tool, 'tool_result', null, block.is_error ? 'error' : 'completed', callId)); + } + return events; } if (type === 'result') { return [{ @@ -247,27 +262,34 @@ function mapClaudePayload(payload) { return []; } -function toolStart(tool, argsText, eventType) { +function toolStart(tool, argsText, eventType, callId = null) { const text = typeof argsText === 'string' ? argsText : ''; return { event: 'tool_start', event_type: eventType, tool, + tool_call_id: callId, args_preview: text ? truncateText(text, MAX_PREVIEW_CHARS) : null, args_digest: text ? digest(text) : null, }; } -function toolEnd(tool, eventType, exitCode, status) { +function toolEnd(tool, eventType, exitCode, status, callId = null) { return { event: 'tool_end', event_type: eventType, tool, + tool_call_id: callId, exit_code: finiteOrNull(exitCode), status: status ? String(status) : null, }; } +/** The CLI's own id for a tool call, or null when the payload has none. */ +function callIdOf(value) { + return typeof value === 'string' && value ? value : null; +} + function readCodexUsage(usage) { if (!usage || typeof usage !== 'object') return null; return compactUsage({ @@ -312,9 +334,17 @@ export function createAgentEventSink(options = {}) { const adapter = options.adapter || 'unknown'; const flushIntervalMs = Math.max(0, Number(options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS)); const maxQueue = Math.max(1, Number(options.maxQueue) || MAX_QUEUE_LENGTH); + // One id per agent process, so two atoms streaming into the same log stay distinguishable even + // when their CLIs reuse the same tool-call ids. Injectable so tests can pin it. + const agentRunId = typeof options.agentRunId === 'string' && options.agentRunId + ? options.agentRunId + : `agent_run_${randomBytes(6).toString('hex')}`; const queue = []; const stats = { lines: 0, events: 0, written: 0, dropped: 0, truncated: 0, write_errors: 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(); let seq = 0; let timer = null; let writing = Promise.resolve(); @@ -323,19 +353,34 @@ export function createAgentEventSink(options = {}) { const splitter = createLineSplitter((line) => { if (closed || !line) return; stats.lines += 1; - for (const normalized of normalizeAgentEvents(line, { adapter, seq })) { + for (const normalized of normalizeAgentEvents(line, { adapter, seq, toolNames })) { seq += 1; stats.events += 1; if (normalized.truncated) stats.truncated += 1; - queue.push(buildRecord(normalized, seq)); - if (queue.length > maxQueue) { - queue.shift(); - stats.dropped += 1; - } + rememberCall(normalized); + enqueue(buildRecord(normalized, seq)); } schedule(); }); + function enqueue(record) { + queue.push(record); + if (queue.length > maxQueue) { + queue.shift(); + stats.dropped += 1; + } + } + + 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); + } + } + function buildRecord(event, eventSeq) { const record = { kind: 'agent_event', @@ -343,10 +388,11 @@ export function createAgentEventSink(options = {}) { phase, atom, adapter, + agent_run_id: agentRunId, seq: eventSeq, }; for (const key of [ - 'event_type', 'tool', 'text', 'args_preview', 'args_digest', 'usage', + 'event_type', 'tool', 'tool_call_id', 'text', 'args_preview', 'args_digest', 'usage', 'exit_code', 'status', 'level', 'cost_usd', 'duration_ms', 'reason', 'truncated', ]) { const value = event[key]; @@ -392,9 +438,27 @@ export function createAgentEventSink(options = {}) { if (closed) return; splitter.push(chunk); }, - /** Write the trailing partial line, then every queued event. Safe to call twice. */ + /** + * Write the trailing partial line, close any call still open, then every queued event. + * + * A run that is killed (the default 300s agent timeout, a Ctrl-C, a crashed CLI) never emits + * its final `tool_result`, so without this the log would show that tool as running forever. + * Safe to call twice. + */ async close() { splitter.flush(); + for (const [callId, call] of openCalls) { + seq += 1; + stats.events += 1; + enqueue(buildRecord({ + event: 'tool_end', + event_type: call.event_type, + tool: call.tool, + tool_call_id: callId, + status: 'abandoned', + }, seq)); + } + openCalls.clear(); closed = true; await flush(); return stats; diff --git a/test/agent-stream.test.js b/test/agent-stream.test.js index b7c90d9..71de400 100644 --- a/test/agent-stream.test.js +++ b/test/agent-stream.test.js @@ -26,10 +26,10 @@ function linesOf(fixture) { return fixture.trim().split('\n'); } -function collect(fixture, adapter) { +function collect(fixture, adapter, options = {}) { const events = []; for (const line of linesOf(fixture)) { - events.push(...normalizeAgentEvents(line, { adapter, seq: events.length })); + events.push(...normalizeAgentEvents(line, { adapter, seq: events.length, ...options })); } return events; } @@ -113,20 +113,52 @@ describe('Agent stream', () => { }); test('maps a claude stream-json run onto the event vocabulary', () => { - const events = collect(claudeFixture, 'claude_cli'); + const events = collect(claudeFixture, 'claude_cli', { toolNames: new Map() }); assert.deepEqual(events.map((event) => event.event), [ 'agent_start', 'message_end', 'tool_start', + 'tool_end', 'tool_start', + 'tool_end', 'agent_end', ]); assert.equal(events[1].text, 'Reading the phase artifact template.'); assert.equal(events[2].tool, 'Read'); assert.equal(events[2].args_digest.length, 16); - assert.equal(events[4].cost_usd, 0.0412); - assert.equal(events[4].duration_ms, 184000); - assert.deepEqual(events[4].usage, { input: 900, output: 210, cache_read: 400 }); + assert.equal(events[6].cost_usd, 0.0412); + assert.equal(events[6].duration_ms, 184000); + assert.deepEqual(events[6].usage, { input: 900, output: 210, cache_read: 400 }); + }); + + test('pairs every claude tool_result back to the tool_use that started it', () => { + const events = collect(claudeFixture, 'claude_cli', { toolNames: new Map() }); + const starts = events.filter((event) => event.event === 'tool_start'); + const ends = events.filter((event) => event.event === 'tool_end'); + assert.deepEqual(starts.map((event) => event.tool_call_id), ['toolu_1', 'toolu_2']); + assert.deepEqual(ends.map((event) => event.tool_call_id), ['toolu_1', 'toolu_2']); + assert.deepEqual(ends.map((event) => event.tool), ['Read', 'Write']); + assert.deepEqual(ends.map((event) => event.status), ['completed', 'completed']); + }); + + test('marks a failed claude tool_result as an error end', () => { + const toolUse = '{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_9","name":"Bash","input":{"command":"npm test"}}]}}'; + const failed = '{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_9","is_error":true,"content":"exit 1"}]}}'; + const events = collect(toolUse + '\n' + failed, 'claude_cli', { toolNames: new Map() }); + assert.deepEqual(events.map((event) => event.event), ['tool_start', 'tool_end']); + assert.equal(events[1].tool, 'Bash'); + assert.equal(events[1].tool_call_id, 'toolu_9'); + assert.equal(events[1].status, 'error'); + }); + + test('keeps a claude tool_end readable when the matching tool_use was never seen', () => { + const orphan = '{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_ghost","content":"ok"}]}}'; + const events = collect(orphan, 'claude_cli', { toolNames: new Map() }); + assert.equal(events.length, 1); + assert.equal(events[0].event, 'tool_end'); + assert.equal(events[0].tool, null); + assert.equal(events[0].tool_call_id, 'toolu_ghost'); + assert.equal(events[0].status, 'completed'); }); test('degrades unknown and malformed lines to raw events instead of throwing', () => { @@ -220,5 +252,56 @@ describe('Agent stream', () => { rmSync(projectRoot, { recursive: true, force: true }); } }); + + test('closes a call that is still open when the agent is killed', async () => { + const projectRoot = makeProjectRoot(); + try { + const sink = createAgentEventSink({ + projectRoot, + adapter: 'claude_cli', + flushIntervalMs: 5, + agentRunId: 'agent_run_test', + }); + const toolUse = { type: 'assistant', message: { content: [{ type: 'tool_use', id: 'toolu_7', name: 'Bash', input: { command: 'sleep 999' } }] } }; + sink.push(`${JSON.stringify(toolUse)}\n`); + const stats = await sink.close(); + assert.equal(stats.events, 2, 'the abandoned end counts as an event'); + + const log = await readExecutionLog(projectRoot, { limit: 10 }); + assert.equal(log.chain.ok, true); + const events = log.entries.filter((entry) => entry.kind === 'agent_event').reverse(); + assert.deepEqual(events.map((entry) => entry.event), ['tool_start', 'tool_end']); + assert.equal(events[1].status, 'abandoned'); + assert.equal(events[1].tool, 'Bash'); + assert.equal(events[1].tool_call_id, 'toolu_7'); + assert.ok(events.every((entry) => entry.agent_run_id === 'agent_run_test')); + } finally { + rmSync(projectRoot, { recursive: true, force: true }); + } + }); + + test('scopes concurrent agent runs so their tool-call ids cannot collide', async () => { + const projectRoot = makeProjectRoot(); + try { + const start = (sink, command) => sink.push(`${JSON.stringify({ type: 'item.started', item: { id: 'item_1', type: 'command_execution', command } })}\n`); + const first = createAgentEventSink({ projectRoot, adapter: 'codex_cli', flushIntervalMs: 5, agentRunId: 'agent_run_a' }); + const second = createAgentEventSink({ projectRoot, adapter: 'codex_cli', flushIntervalMs: 5, agentRunId: 'agent_run_b' }); + start(first, 'npm test'); + start(second, 'npm run build'); + await first.close(); + await second.close(); + + const log = await readExecutionLog(projectRoot, { limit: 20 }); + assert.equal(log.chain.ok, true); + const events = log.entries.filter((entry) => entry.kind === 'agent_event'); + const perRun = new Map(); + for (const entry of events) perRun.set(entry.agent_run_id, (perRun.get(entry.agent_run_id) || 0) + 1); + assert.deepEqual([...perRun.keys()].sort(), ['agent_run_a', 'agent_run_b']); + assert.deepEqual([...perRun.values()], [2, 2]); + assert.ok(events.every((entry) => entry.tool_call_id === 'item_1')); + } finally { + rmSync(projectRoot, { recursive: true, force: true }); + } + }); }); }); diff --git a/test/fixtures/agent-stream/claude.ndjson b/test/fixtures/agent-stream/claude.ndjson index 5987e0b..fafd3ed 100644 --- a/test/fixtures/agent-stream/claude.ndjson +++ b/test/fixtures/agent-stream/claude.ndjson @@ -2,4 +2,5 @@ {"type":"assistant","message":{"id":"msg_1","content":[{"type":"text","text":"Reading the phase artifact template."},{"type":"tool_use","id":"toolu_1","name":"Read","input":{"file_path":"prompts/phase-artifact.md"}}]}} {"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_1","content":"# Phase artifact"}]}} {"type":"assistant","message":{"id":"msg_2","content":[{"type":"tool_use","id":"toolu_2","name":"Write","input":{"file_path":"specs/changes/chg-1/artifact.json"}}]}} +{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_2","content":"ok"}]}} {"type":"result","subtype":"success","is_error":false,"duration_ms":184000,"total_cost_usd":0.0412,"usage":{"input_tokens":900,"output_tokens":210,"cache_read_input_tokens":400}}