diff --git a/src/backend/codex-stdio.ts b/src/backend/codex-stdio.ts index 7c81a9b..23a980a 100644 --- a/src/backend/codex-stdio.ts +++ b/src/backend/codex-stdio.ts @@ -1228,6 +1228,12 @@ function normalizeCodexNotification(message: JsonRecord, suppressed: SuppressedN recordSuppressedNotification(suppressed, method, "reasoning"); return { events: [] }; } + if (method === "turn/diff/updated") { + return { events: [{ type: "diff", payload: turnDiffPayload(params) }] }; + } + if (method === "item/fileChange/patchUpdated") { + return { events: [{ type: "tool_call", payload: fileChangePatchUpdatedPayload(params) }] }; + } if ((method === "item/started" || method === "item/completed") && asRecordAt(params, "item").type === "agentMessage") { const item = asRecordAt(params, "item"); const itemId = stringAt(item, "id") ?? stringAt(params, "itemId"); @@ -1306,7 +1312,7 @@ function isSuppressedCodexStatusNotification(method: string): boolean { } function isVisibleCodexToolItemType(itemType: string): boolean { - return itemType === "commandExecution" || itemType === "webSearch" || itemType === "mcpToolCall" || itemType === "dynamicToolCall"; + return itemType === "commandExecution" || itemType === "fileChange" || itemType === "webSearch" || itemType === "mcpToolCall" || itemType === "dynamicToolCall"; } function assistantMessageEventForCompleted(message: CompletedAssistantMessage, messageIndex: number): BackendEvent { @@ -1453,6 +1459,7 @@ function terminalStatusFromValue(value: unknown): TerminalStatus { } function toolCallPayload(method: string, item: JsonRecord): JsonRecord { + if (item.type === "fileChange") return fileChangeToolCallPayload(method, item); const redacted = redactJson(item); const itemId = typeof redacted.id === "string" ? redacted.id : null; const itemType = typeof redacted.type === "string" ? redacted.type : "unknown"; @@ -1480,6 +1487,50 @@ function toolCallPayload(method: string, item: JsonRecord): JsonRecord { }; } +function fileChangeToolCallPayload(method: string, item: JsonRecord): JsonRecord { + const visible = redactJson(item); + const changes = Array.isArray(visible.changes) ? visible.changes : []; + return { + method, + itemId: stringAt(visible, "id"), + type: "fileChange", + toolName: "fileChange", + status: toolCallStatus(method, visible), + changes, + fileCount: changes.length, + valuesPrinted: true, + }; +} + +function fileChangePatchUpdatedPayload(params: JsonRecord): JsonRecord { + const visible = redactJson(params); + const changes = Array.isArray(visible.changes) ? visible.changes : []; + return { + method: "item/fileChange/patchUpdated", + itemId: stringAt(visible, "itemId"), + threadId: stringAt(visible, "threadId"), + turnId: stringAt(visible, "turnId"), + type: "fileChange", + toolName: "fileChange", + status: "started", + changes, + fileCount: changes.length, + valuesPrinted: true, + }; +} + +function turnDiffPayload(params: JsonRecord): JsonRecord { + const visible = redactJson(params); + return { + phase: "turn/diff/updated", + threadId: stringAt(visible, "threadId"), + turnId: stringAt(visible, "turnId"), + diff: typeof visible.diff === "string" ? visible.diff : "", + status: "updated", + valuesPrinted: true, + }; +} + function toolCallStatus(method: string, item: JsonRecord): string | null { const status = typeof item.status === "string" ? item.status.trim().toLowerCase() : ""; if (status === "inprogress" || status === "in_progress" || status === "running" || status === "pending") return "started"; diff --git a/src/selftest/cases/30-codex-stdio.ts b/src/selftest/cases/30-codex-stdio.ts index b5a1477..a2593a2 100644 --- a/src/selftest/cases/30-codex-stdio.ts +++ b/src/selftest/cases/30-codex-stdio.ts @@ -197,6 +197,24 @@ const selfTest: SelfTestCase = async (context) => { assert.equal(mcpToolItems.some((event) => event.type === "backend_status" && JSON.stringify(eventPayload(event).itemTypes ?? []).includes("mcpToolCall")), false, "mcpToolCall must not be counted as a suppressed notification"); assertNoSecretLeak(mcpToolEvents); + const fileChange = await createRunWithCommand(client, context, "edit files", "selftest-file-change-events", 15_000); + const fileChangeResult = await runOnce({ managerUrl: server.baseUrl, runId: fileChange.runId, codexCommand: context.fakeCodexCommand, codexArgs: context.fakeCodexArgs, codexHome: context.codexHome, env: { CODEX_HOME: context.codexHome, AGENTRUN_FAKE_CODEX_MODE: "file-change-events" }, oneShot: true }); + assert.equal(fileChangeResult.terminalStatus, "completed", "file change turn should complete"); + const fileChangeEvents = await client.get(`/api/v1/runs/${fileChange.runId}/events?afterSeq=0&limit=100`) as { items?: Array<{ type: string; payload: JsonRecord }> }; + const fileChangeItems = fileChangeEvents.items ?? []; + const fileChangeLifecycle = fileChangeItems.filter((event) => event.type === "tool_call" && eventPayload(event).type === "fileChange"); + assert.deepEqual(fileChangeLifecycle.map((event) => eventPayload(event).method), ["item/started", "item/fileChange/patchUpdated", "item/completed"]); + assert.deepEqual(fileChangeLifecycle.map((event) => eventPayload(event).status), ["started", "started", "completed"]); + assert.equal(eventPayload(fileChangeLifecycle.at(-1) ?? { payload: {} }).fileCount, 2); + assert.equal(JSON.stringify(eventPayload(fileChangeLifecycle.at(-1) ?? { payload: {} }).changes).includes("src/example.ts"), true); + assert.equal(JSON.stringify(eventPayload(fileChangeLifecycle.at(-1) ?? { payload: {} }).changes).includes("docs/说明.md"), true); + assert.equal(JSON.stringify(eventPayload(fileChangeLifecycle.at(-1) ?? { payload: {} }).changes).includes("+export const value = 2;"), true); + const diffEvent = fileChangeItems.find((event) => event.type === "diff"); + assert.equal(eventPayload(diffEvent ?? { payload: {} }).phase, "turn/diff/updated"); + assert.equal(String(eventPayload(diffEvent ?? { payload: {} }).diff ?? "").includes("+# 文件编辑可见"), true); + assert.equal(fileChangeItems.some((event) => event.type === "backend_status" && JSON.stringify(eventPayload(event).itemTypes ?? []).includes("fileChange")), false, "fileChange must not be counted as suppressed"); + assertNoSecretLeak(fileChangeEvents); + const staleThread = await createStaleThreadRun(client, context); const staleThreadResult = await runOnce({ managerUrl: server.baseUrl, diff --git a/src/selftest/cases/36-stdio-event-reconstruction.ts b/src/selftest/cases/36-stdio-event-reconstruction.ts index a11922f..9323d6a 100644 --- a/src/selftest/cases/36-stdio-event-reconstruction.ts +++ b/src/selftest/cases/36-stdio-event-reconstruction.ts @@ -74,6 +74,22 @@ const selfTest: SelfTestCase = async (context) => { assert.equal(replyRef.textSha256, sha256Hex(finalAssistantText)); assert.equal((reconstructed.records.at(-1)?.value.event as JsonRecord).type, "terminal_status"); + const fileChangeFixture = await readFile(path.join(context.root, "src/selftest/fixtures/stdio-file-change-reconstruction.jsonl"), "utf8"); + const fileChangeFrames = parseStdioKafkaFrameJsonl(fileChangeFixture); + const fileChangeReconstruction = reconstructAgentRunEventsFromStdioFrames(fileChangeFrames, { + sessionId: "ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2", + traceId: "trc_9352107b5e8a4125", + }); + assert.equal(fileChangeReconstruction.selectedFrameCount, 8); + assert.deepEqual(fileChangeReconstruction.byEventType, { backend_status: 4, tool_call: 3, diff: 1, assistant_message: 1, terminal_status: 1 }); + const fileChangeEvents = fileChangeReconstruction.records.map((record) => record.value.event as JsonRecord); + const completedFileChange = fileChangeEvents.find((event) => event.type === "tool_call" && (event.payload as JsonRecord).method === "item/completed"); + assert.equal(((completedFileChange?.payload as JsonRecord).changes as JsonRecord[])[0]?.path, "src/example.ts"); + assert.equal(((completedFileChange?.payload as JsonRecord).changes as JsonRecord[])[1]?.path, "docs/说明.md"); + assert.match(String(((completedFileChange?.payload as JsonRecord).changes as JsonRecord[])[0]?.diff), /\+export const value = 2;/u); + const aggregateDiff = fileChangeEvents.find((event) => event.type === "diff"); + assert.match(String((aggregateDiff?.payload as JsonRecord).diff), /\+# 文件编辑可见/u); + let readGroupId = ""; const published: ReconstructedAgentRunKafkaRecord[] = []; const result = await regenerateAgentRunEventsFromStdio({ diff --git a/src/selftest/fake-codex-app-server.ts b/src/selftest/fake-codex-app-server.ts index 1b47963..0ce61b3 100644 --- a/src/selftest/fake-codex-app-server.ts +++ b/src/selftest/fake-codex-app-server.ts @@ -245,6 +245,24 @@ for await (const line of rl) { respond(message.id, { turn }); continue; } + if (mode === "file-change-events") { + turnCounter += 1; + const turn = { id: `turn_selftest_${turnCounter}`, status: "completed" }; + const startedChanges = [{ path: "src/example.ts", kind: "update", diff: "@@ -1 +1 @@\n-export const value = 1;\n+export const value = 2;" }]; + const completedChanges = [ + ...startedChanges, + { path: "docs/说明.md", kind: "add", diff: "@@ -0,0 +1 @@\n+# 文件编辑可见" }, + ]; + respond(message.id, { turn: { id: turn.id, status: "running" } }); + notify("turn/started", { turn: { id: turn.id, status: "running" } }); + notify("item/started", { item: { id: "item_file_change", type: "fileChange", changes: startedChanges, status: "inProgress" } }); + notify("item/fileChange/patchUpdated", { threadId: "thread_selftest_1", turnId: turn.id, itemId: "item_file_change", changes: completedChanges }); + notify("turn/diff/updated", { threadId: "thread_selftest_1", turnId: turn.id, diff: completedChanges.map((change) => change.diff).join("\n") }); + notify("item/completed", { item: { id: "item_file_change", type: "fileChange", changes: completedChanges, status: "completed" } }); + notify("item/completed", { item: { id: "msg_file_change", type: "agentMessage", text: "文件编辑完成。" } }); + notify("turn/completed", { turn }); + continue; + } if (mode === "multi-agent-message-terminal-before-final") { turnCounter += 1; const turn = { id: `turn_selftest_${turnCounter}`, status: "completed" }; diff --git a/src/selftest/fixtures/stdio-file-change-reconstruction.jsonl b/src/selftest/fixtures/stdio-file-change-reconstruction.jsonl new file mode 100644 index 0000000..88c9252 --- /dev/null +++ b/src/selftest/fixtures/stdio-file-change-reconstruction.jsonl @@ -0,0 +1,8 @@ +{"topic":"codex-stdio.raw.v1","partition":0,"offset":"1","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","frameSeq":1,"producedAt":"2026-07-12T15:00:00.001Z","context":{"runId":"run_cc1bf3a430b14de1b1e147ea95f7773e","commandId":"cmd_dffb545477d145f6bbb66d5500dff7ed","sessionId":"ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2","traceId":"trc_9352107b5e8a4125"},"stdio":{"direction":"stdout","rawJson":{"method":"thread/started","params":{"thread":{"id":"thread_file_change_replay"}}}}}} +{"topic":"codex-stdio.raw.v1","partition":0,"offset":"2","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","frameSeq":2,"producedAt":"2026-07-12T15:00:00.002Z","context":{"runId":"run_cc1bf3a430b14de1b1e147ea95f7773e","commandId":"cmd_dffb545477d145f6bbb66d5500dff7ed","sessionId":"ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2","traceId":"trc_9352107b5e8a4125"},"stdio":{"direction":"stdout","rawJson":{"method":"turn/started","params":{"turn":{"id":"turn_file_change_replay"}}}}}} +{"topic":"codex-stdio.raw.v1","partition":0,"offset":"3","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","frameSeq":3,"producedAt":"2026-07-12T15:00:00.003Z","context":{"runId":"run_cc1bf3a430b14de1b1e147ea95f7773e","commandId":"cmd_dffb545477d145f6bbb66d5500dff7ed","sessionId":"ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2","traceId":"trc_9352107b5e8a4125"},"stdio":{"direction":"stdout","rawJson":{"method":"item/started","params":{"item":{"id":"item_file_change_replay","type":"fileChange","status":"inProgress","changes":[{"path":"src/example.ts","kind":"update","diff":"@@ -1 +1 @@\n-export const value = 1;\n+export const value = 2;"}]}}}}}} +{"topic":"codex-stdio.raw.v1","partition":0,"offset":"4","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","frameSeq":4,"producedAt":"2026-07-12T15:00:00.004Z","context":{"runId":"run_cc1bf3a430b14de1b1e147ea95f7773e","commandId":"cmd_dffb545477d145f6bbb66d5500dff7ed","sessionId":"ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2","traceId":"trc_9352107b5e8a4125"},"stdio":{"direction":"stdout","rawJson":{"method":"item/fileChange/patchUpdated","params":{"threadId":"thread_file_change_replay","turnId":"turn_file_change_replay","itemId":"item_file_change_replay","changes":[{"path":"src/example.ts","kind":"update","diff":"@@ -1 +1 @@\n-export const value = 1;\n+export const value = 2;"},{"path":"docs/说明.md","kind":"add","diff":"@@ -0,0 +1 @@\n+# 文件编辑可见"}]}}}}} +{"topic":"codex-stdio.raw.v1","partition":0,"offset":"5","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","frameSeq":5,"producedAt":"2026-07-12T15:00:00.005Z","context":{"runId":"run_cc1bf3a430b14de1b1e147ea95f7773e","commandId":"cmd_dffb545477d145f6bbb66d5500dff7ed","sessionId":"ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2","traceId":"trc_9352107b5e8a4125"},"stdio":{"direction":"stdout","rawJson":{"method":"turn/diff/updated","params":{"threadId":"thread_file_change_replay","turnId":"turn_file_change_replay","diff":"@@ -1 +1 @@\n-export const value = 1;\n+export const value = 2;\n@@ -0,0 +1 @@\n+# 文件编辑可见"}}}}} +{"topic":"codex-stdio.raw.v1","partition":0,"offset":"6","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","frameSeq":6,"producedAt":"2026-07-12T15:00:00.006Z","context":{"runId":"run_cc1bf3a430b14de1b1e147ea95f7773e","commandId":"cmd_dffb545477d145f6bbb66d5500dff7ed","sessionId":"ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2","traceId":"trc_9352107b5e8a4125"},"stdio":{"direction":"stdout","rawJson":{"method":"item/completed","params":{"item":{"id":"item_file_change_replay","type":"fileChange","status":"completed","changes":[{"path":"src/example.ts","kind":"update","diff":"@@ -1 +1 @@\n-export const value = 1;\n+export const value = 2;"},{"path":"docs/说明.md","kind":"add","diff":"@@ -0,0 +1 @@\n+# 文件编辑可见"}]}}}}}} +{"topic":"codex-stdio.raw.v1","partition":0,"offset":"7","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","frameSeq":7,"producedAt":"2026-07-12T15:00:00.007Z","context":{"runId":"run_cc1bf3a430b14de1b1e147ea95f7773e","commandId":"cmd_dffb545477d145f6bbb66d5500dff7ed","sessionId":"ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2","traceId":"trc_9352107b5e8a4125"},"stdio":{"direction":"stdout","rawJson":{"method":"item/completed","params":{"item":{"id":"msg_file_change_replay","type":"agentMessage","text":"文件编辑完成。"}}}}}} +{"topic":"codex-stdio.raw.v1","partition":0,"offset":"8","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","frameSeq":8,"producedAt":"2026-07-12T15:00:00.008Z","context":{"runId":"run_cc1bf3a430b14de1b1e147ea95f7773e","commandId":"cmd_dffb545477d145f6bbb66d5500dff7ed","sessionId":"ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2","traceId":"trc_9352107b5e8a4125"},"stdio":{"direction":"stdout","rawJson":{"method":"turn/completed","params":{"turn":{"id":"turn_file_change_replay","status":"completed"}}}}}}