From 719584e2ce5b132a560335792459388cb8c6b59c Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Tue, 2 Jun 2026 07:09:12 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=9D=E7=95=99=20completed=20agentMe?= =?UTF-8?q?ssage=20=E4=BA=8B=E4=BB=B6=20(#56)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Codex --- docs/reference/spec-v01-agentrun-mgr.md | 2 +- docs/reference/spec-v01-backend-adapter.md | 4 +-- src/backend/codex-stdio.ts | 42 ++++++++++++++++++---- src/mgr/result.ts | 11 +++--- src/selftest/cases/30-codex-stdio.ts | 9 +++-- 5 files changed, 52 insertions(+), 16 deletions(-) diff --git a/docs/reference/spec-v01-agentrun-mgr.md b/docs/reference/spec-v01-agentrun-mgr.md index 783f68e..eb11ea8 100644 --- a/docs/reference/spec-v01-agentrun-mgr.md +++ b/docs/reference/spec-v01-agentrun-mgr.md @@ -119,7 +119,7 @@ Manager 只承接 HWLAB v0.2 Code Agent 的通用执行事实，不承接 HWLAB | `status` | run/command 当前聚合状态，只能由 command state 和 terminal_status 推导。 | | `terminalStatus` | `completed`、`failed`、`blocked` 或 `cancelled`；没有 terminal event 时为 `null` 或 equivalent running 状态。 | | `completed` / `terminalSource` | `completed=true` 只能来自 terminal completed；`terminalSource` 标明来自 `terminal_status` event、run record 或暂无 terminal。 | -| `reply` | 从 `assistant_message` 聚合的最终用户可见文本；没有 terminal completed 时不得伪造 completed reply。 | +| `reply` | 从 `assistant_message` 聚合的最终用户可见文本；若存在 `replyAuthority=true` 或 `final=true` 的 `assistant_message`，必须取最后一条作为 reply；没有这些标记时才按 legacy chunk 聚合。没有 terminal completed 时不得伪造 completed reply。 | | `failureKind` / `blocker` | 结构化失败分类和摘要；必须 redacted。 | | `lastSeq` / `eventCount` | 支持调用方增量轮询和 result/trace reconciliation。 | | `runId` / `commandId` / `attemptId` | 支持调用方持久关联和问题定位。 | diff --git a/docs/reference/spec-v01-backend-adapter.md b/docs/reference/spec-v01-backend-adapter.md index f01ed33..148d46f 100644 --- a/docs/reference/spec-v01-backend-adapter.md +++ b/docs/reference/spec-v01-backend-adapter.md @@ -38,7 +38,7 @@ Backend adapter 的第一阶段实现应吸收 HWLAB v0.2 已验证的 Codex std | --- | --- | --- | | Codex app-server JSON-RPC stdio | `internal/cloud/codex-stdio-session.ts`、`internal/cloud/codex-stdio-session-turn-state.ts` | 支持 `initialize`、`thread/start`、`thread/resume`、`turn/start`，并处理 app-server client request；未知请求要记录 unsupported error，不能静默等待。 | | completed 判定 | `docs/reference/code-agent-chat-readiness.md` | 只有 Codex turn terminal completed 且 assistant reply 可聚合时才输出 completed；assistant delta、item completed、stdout 或 transport close 不能单独完成。 | -| assistant stream 和 trace | `internal/cloud/code-agent-trace-store.ts`、`internal/cloud/codex-stdio-session-turn-state.ts` | assistant delta 只能作为 stream/progress 证据；最终 `assistant_message` 必须优先来自最后一个 completed `agentMessage` item，不能把 commentary/progress delta 与 final response 直接串接。event 必须保留 `threadId`、`turnId`、session 摘要和 redacted backend metadata。 | +| assistant stream 和 trace | `internal/cloud/code-agent-trace-store.ts`、`internal/cloud/codex-stdio-session-turn-state.ts` | assistant delta 只能作为 stream/progress 证据；每个非空 completed `agentMessage` item 必须输出一个 `assistant_message` event，保留 `itemId` 和顺序；最终 result reply 必须优先来自最后一个 completed `agentMessage` item，不能把 commentary/progress delta 与 final response 直接串接。event 必须保留 `threadId`、`turnId`、session 摘要和 redacted backend metadata。 | | command/tool output bounded | `docs/reference/code-agent-chat-readiness.md`、`web/hwlab-cloud-web/app-trace.ts` | `tool_call` 和 `command_output` 必须记录状态、摘要、字节数、截断标记；完整大输出只能通过后续 log/artifact 引用。 | | provider/profile 隔离 | `internal/cloud/code-agent-contract.ts` | `codex` 与 `deepseek` 共享同一 backend kind，但必须使用 profile-scoped SecretRef、model/base-url/config 和 writable runtime home。 | | Secret redaction | `internal/cloud/code-agent-trace-store.ts` | `OPENAI_API_KEY`、auth/config、token、password、kubeconfig、URL credential 不得进入 event、result、log 或 health。 | @@ -59,7 +59,7 @@ Registry 只表达能力和选择边界，不读取 Secret 值。Manager 负责 Adapter 输出给 runner 的 event 类型至少包括： - `backend_status`：backend 启动、模型/profile、能力和阶段状态，不包含 Secret 值。 -- `assistant_message`：模型回复的最终用户可见文本。Codex app-server 的 `item/agentMessage/delta` 只能作为流式过程证据或缺少 completed item 时的兜底；一旦收到 completed `agentMessage` item，adapter 必须以最后一个非空 completed item 文本作为 reply authority，避免把 commentary/status/progress 堆入 final response。 +- `assistant_message`：模型输出的用户可见 assistant 文本。Codex app-server 的 `item/agentMessage/delta` 只能作为流式过程证据或缺少 completed item 时的兜底；一旦收到 completed `agentMessage` item，adapter 必须为每个非空 completed item 输出一条 `assistant_message`，并用 `itemId`、`messageIndex`、`messageCount`、`replyAuthority` 和 `final` 标明顺序与最终 reply authority。最终 result reply 必须以最后一个 `replyAuthority=true` / `final=true` 的 `assistant_message` 为准，避免把 commentary/status/progress 堆入 final response。 - `tool_call`：工具调用摘要和 redacted 参数。 - `command_output`：stdout/stderr 或命令输出摘要。 - `diff`：代码变更摘要或 patch 片段；必须受长度限制。 diff --git a/src/backend/codex-stdio.ts b/src/backend/codex-stdio.ts index b1cd54e..60f855e 100644 --- a/src/backend/codex-stdio.ts +++ b/src/backend/codex-stdio.ts @@ -57,6 +57,11 @@ interface PendingRequest { reject: (error: Error) => void; } +interface CompletedAssistantMessage { + itemId: string | null; + text: string; +} + interface CodexStdioCloseInfo extends JsonRecord { code: number | null; signal: string | null; @@ -376,7 +381,7 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess return { terminalStatus: cancelled.status, failureKind: cancelled.failureKind, failureMessage: cancelled.message, events: events.map((event) => ({ ...event, payload: redactJson(event.payload) })) }; } let assistantText = ""; - let finalAssistantText = ""; + const completedAssistantMessages: CompletedAssistantMessage[] = []; let threadId: string | undefined = options.threadId; let turnId: string | undefined; let terminal: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } | null = null; @@ -404,7 +409,7 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess if (normalized.threadId) threadId = normalized.threadId; if (normalized.turnId) turnId = normalized.turnId; if (normalized.assistantDelta) assistantText += normalized.assistantDelta; - if (typeof normalized.assistantFinal === "string" && normalized.assistantFinal.trim().length > 0) finalAssistantText = normalized.assistantFinal; + if (normalized.completedAssistantMessage) completedAssistantMessages.push(normalized.completedAssistantMessage); emitEvents(normalized.events); if (normalized.terminal && !terminal) { terminal = normalized.terminal; @@ -471,9 +476,8 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess } if (!terminal) terminal = { status: "failed", failureKind: "backend-response-invalid", message: "codex app-server finished without terminal status" }; if (terminal.status !== "completed") emitEvents(await session.close()); - const reply = finalAssistantText.trim().length > 0 ? finalAssistantText : assistantText; - if (reply.trim().length > 0) events.push({ type: "assistant_message", payload: { text: reply } }); - events.push({ type: "terminal_status", payload: { terminalStatus: terminal.status, failureKind: terminal.failureKind, message: terminal.message } }); + emitEvents(assistantMessageEventsForTurn(completedAssistantMessages, assistantText, terminal.status === "completed")); + emitEvent({ type: "terminal_status", payload: { terminalStatus: terminal.status, failureKind: terminal.failureKind, message: terminal.message } }); await liveEventWrite; return { terminalStatus: terminal.status, failureKind: terminal.failureKind, failureMessage: terminal.message, events: events.map((event) => ({ ...event, payload: redactJson(event.payload) })), ...(threadId ? { threadId } : {}), ...(turnId ? { turnId } : {}) }; } @@ -534,7 +538,7 @@ function codexHomeReadiness(codexHome: string): BackendTurnResult | null { }; } -function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent[]; assistantDelta?: string; assistantFinal?: string; threadId?: string; turnId?: string; terminal?: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } } { +function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent[]; assistantDelta?: string; completedAssistantMessage?: CompletedAssistantMessage; threadId?: string; turnId?: string; terminal?: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } } { const method = typeof message.method === "string" ? message.method : "unknown"; const params = asRecordAt(message, "params"); if (method === "thread/started") { @@ -551,9 +555,10 @@ function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent const item = asRecordAt(params, "item"); const itemId = stringAt(item, "id") ?? stringAt(params, "itemId"); const text = method === "item/completed" ? agentMessageText(item) : ""; + const completedAssistantMessage = text.trim().length > 0 ? { itemId: itemId ?? null, text } : undefined; return { events: [{ type: "backend_status", payload: { phase: method === "item/completed" ? "item/agentMessage:completed" : "item/agentMessage:started", itemId, textBytes: Buffer.byteLength(text, "utf8") } }], - ...(text.trim().length > 0 ? { assistantFinal: text } : {}), + ...(completedAssistantMessage ? { completedAssistantMessage } : {}), }; } if (method === "item/started" || method === "item/completed") return { events: [{ type: "tool_call", payload: toolCallPayload(method, asRecordAt(params, "item")) }] }; @@ -580,6 +585,29 @@ function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent return { events: [{ type: "backend_status", payload: { phase: method } }] }; } +function assistantMessageEventsForTurn(completedMessages: CompletedAssistantMessage[], assistantDeltaText: string, completed: boolean): BackendEvent[] { + const messages = completedMessages.length > 0 + ? completedMessages.map((message) => ({ ...message, source: "completed-agent-message" })) + : assistantDeltaText.trim().length > 0 + ? [{ itemId: null, text: assistantDeltaText, source: "agent-message-delta-fallback" }] + : []; + return messages.map((message, index) => { + const replyAuthority = completed && index === messages.length - 1; + return { + type: "assistant_message", + payload: { + text: message.text, + itemId: message.itemId, + source: message.source, + messageIndex: index + 1, + messageCount: messages.length, + replyAuthority, + final: replyAuthority, + }, + }; + }); +} + function terminalStatusFromValue(value: unknown): TerminalStatus { if (value === "completed") return "completed"; if (value === "cancelled" || value === "canceled" || value === "interrupted") return "cancelled"; diff --git a/src/mgr/result.ts b/src/mgr/result.ts index adeb79f..e1d61a0 100644 --- a/src/mgr/result.ts +++ b/src/mgr/result.ts @@ -90,11 +90,14 @@ function messageFromEvents(events: RunEvent[]): string | null { } function assistantReply(events: RunEvent[]): string { - return events - .filter((event) => event.type === "assistant_message") + const assistantEvents = events.filter((event) => event.type === "assistant_message"); + const authoritative = assistantEvents + .filter((event) => event.payload.replyAuthority === true || event.payload.final === true) .map((event) => textPayload(event.payload)) - .filter((text) => text.length > 0) - .join(""); + .filter((text) => text.length > 0); + const latestAuthoritative = authoritative.at(-1); + if (latestAuthoritative) return latestAuthoritative; + return assistantEvents.map((event) => textPayload(event.payload)).filter((text) => text.length > 0).join(""); } function textPayload(payload: JsonRecord): string { diff --git a/src/selftest/cases/30-codex-stdio.ts b/src/selftest/cases/30-codex-stdio.ts index 1918e1f..41face3 100644 --- a/src/selftest/cases/30-codex-stdio.ts +++ b/src/selftest/cases/30-codex-stdio.ts @@ -65,8 +65,13 @@ const selfTest: SelfTestCase = async (context) => { assert.equal(finalMessageEnvelope.reply, "Final answer only.", "result reply should use the final completed agentMessage instead of concatenating progress deltas"); const finalMessageEvents = await client.get(`/api/v1/runs/${finalMessage.runId}/events?afterSeq=0&limit=100`) as { items?: Array<{ type: string; payload: unknown }> }; const assistantEvents = finalMessageEvents.items?.filter((event) => event.type === "assistant_message") ?? []; - assert.equal(assistantEvents.length, 1, "backend should emit one final assistant_message event"); - assert.equal(eventPayload(assistantEvents[0] ?? { payload: {} }).text, "Final answer only."); + assert.equal(assistantEvents.length, 2, "backend should preserve each completed agentMessage as assistant_message event"); + assert.equal(eventPayload(assistantEvents[0] ?? { payload: {} }).text, "I am checking the workspace."); + assert.equal(eventPayload(assistantEvents[0] ?? { payload: {} }).itemId, "msg_progress"); + assert.equal(eventPayload(assistantEvents[0] ?? { payload: {} }).replyAuthority, false); + assert.equal(eventPayload(assistantEvents[1] ?? { payload: {} }).text, "Final answer only."); + assert.equal(eventPayload(assistantEvents[1] ?? { payload: {} }).itemId, "msg_final"); + assert.equal(eventPayload(assistantEvents[1] ?? { payload: {} }).replyAuthority, true); const staleResume = await createStaleThreadRun(client, context); const staleResumeResult = await runOnce({