From 05e30ce2ba24b967622379d7b65291f147ea2c80 Mon Sep 17 00:00:00 2001 From: lyon Date: Fri, 19 Jun 2026 04:48:14 +0800 Subject: [PATCH] fix: backfill AgentRun terminal projection on read --- internal/cloud/code-agent-agentrun-adapter.ts | 20 ++++-- internal/cloud/server-agent-chat.test.ts | 33 ++++++--- internal/cloud/server-code-agent-http.ts | 67 ++++++++++++++++++- 3 files changed, 101 insertions(+), 19 deletions(-) diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index f52fcc54..9e383697 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -1589,16 +1589,25 @@ function agentRunLongLivedSessionGate(base = {}) { } function agentRunToolCalls(result = {}, status = "completed") { + const replyText = agentRunReplyText(result); return [{ name: "agentrun.v01.command.turn", status, runId: result?.runId ?? null, commandId: result?.commandId ?? null, - outputSummary: result?.reply ? `assistantChars=${String(result.reply).length}` : null, + outputSummary: replyText ? `assistantChars=${replyText.length}` : null, valuesPrinted: false }]; } +function agentRunReplyText(result = {}) { + for (const value of [result?.reply, result?.assistantMessage, result?.finalResponse, result?.content, result?.text, result?.message]) { + const text = promptTextValue(value); + if (text) return text; + } + return ""; +} + function agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, appendResultEvent = true }) { const terminalStatus = String(result?.terminalStatus ?? ""); const terminal = terminalStatus === "completed" || terminalStatus === "failed" || terminalStatus === "blocked" || terminalStatus === "cancelled"; @@ -1609,8 +1618,9 @@ function agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, a const runnerTrace = traceStore.snapshot(traceId, agentRunTraceMeta({}, {})); const terminalEventCreatedAt = agentRunResultTraceCreatedAt(runnerTrace, now); const providerTrace = agentRunProviderTrace({ base, result, terminalStatus }); - if (terminalStatus === "completed" && String(result?.reply ?? "").trim()) { - const finalResponse = agentRunCompletedFinalResponse({ base, result, traceId, now }); + const replyText = agentRunReplyText(result); + if (terminalStatus === "completed" && replyText) { + const finalResponse = agentRunCompletedFinalResponse({ base, result, traceId, now, replyText }); const traceSummary = agentRunCompletedTraceSummary({ base, runnerTrace, finalResponse, traceId }); if (appendResultEvent) { traceStore.append(traceId, agentRunTraceEvent({ @@ -1724,8 +1734,8 @@ function agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, a }; } -function agentRunCompletedFinalResponse({ base, result, traceId, now }) { - const textValue = String(result?.reply ?? "").trim(); +function agentRunCompletedFinalResponse({ base, result, traceId, now, replyText = null }) { + const textValue = promptTextValue(replyText) || agentRunReplyText(result); return { text: textValue, textChars: textValue.length, diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 7301ada8..9478d444 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/internal/cloud/server-agent-chat.test.ts @@ -2308,14 +2308,14 @@ test("cloud api running trace reports current projection without AgentRun event } }); -test("cloud api turn status reads Workbench projection without AgentRun refresh (#1519)", async () => { +test("cloud api turn status backfills terminal AgentRun projection on read (#1555)", async () => { const calls = []; const traceId = "trc_issue1422_turn_status_fast_path"; const runId = "run_issue1422_turn_status_fast_path"; const commandId = "cmd_issue1422_turn_status_fast_path"; const events = [ { id: "evt_issue1422_created", runId, seq: 1, type: "backend_status", payload: { phase: "runner-job-created", commandId }, createdAt: "2026-06-18T04:30:00.000Z" }, - { id: "evt_issue1422_message", runId, seq: 2, type: "assistant_message", payload: { commandId, itemId: "msg_issue1422_running", text: "正在处理请求。" }, createdAt: "2026-06-18T04:30:01.000Z" } + { id: "evt_issue1422_terminal", runId, seq: 2, type: "terminal_status", payload: { commandId, terminalStatus: "completed" }, createdAt: "2026-06-18T04:30:01.000Z" } ]; const agentRunServer = createHttpServer(async (request, response) => { const url = new URL(request.url || "/", "http://127.0.0.1"); @@ -2325,9 +2325,18 @@ test("cloud api turn status reads Workbench projection without AgentRun refresh response.end(`${JSON.stringify({ ok: true, data, traceId: "trc_fake_issue1422_turn" })}\n`); }; if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/events`) return send({ items: events }); - if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${commandId}/result`) { - return send({ runId, commandId, status: "running", runStatus: "running", commandState: "running", terminalStatus: null, completed: false, reply: null, lastSeq: 2, eventCount: 2 }); - } + if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${commandId}/result`) return send({ + runId, + commandId, + status: "completed", + runStatus: "claimed", + commandState: "completed", + terminalStatus: "completed", + completed: true, + reply: { content: "issue 1555 read-side backfill completed" }, + lastSeq: 2, + eventCount: 2 + }); response.writeHead(404, { "content-type": "application/json" }); response.end(`${JSON.stringify({ ok: false, message: `unexpected ${request.method} ${url.pathname}` })}\n`); }); @@ -2374,12 +2383,14 @@ test("cloud api turn status reads Workbench projection without AgentRun refresh }); assert.equal(response.status, 200); const body = await response.json(); - assert.equal(body.status, "running"); - assert.equal(body.projectionStatus, "projecting"); - assert.equal(body.eventCount, 0); - assert.equal(body.runnerTrace, null); - assert.equal(calls.filter((call) => call.path === `/api/v1/runs/${runId}/events`).length, 0); - assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), false); + assert.equal(body.status, "completed"); + assert.equal(body.terminal, true); + assert.equal(body.agentRun.runStatus, "claimed"); + assert.equal(body.agentRun.commandState, "completed"); + assert.equal(body.agentRun.terminalStatus, "completed"); + assert.match(body.finalResponse.text, /issue 1555 read-side backfill/u); + assert.equal(calls.filter((call) => call.path === `/api/v1/runs/${runId}/events`).length, 1); + assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), true); } finally { await new Promise((resolve, reject) => { server.close((error) => (error ? reject(error) : resolve())); diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index bd6acac7..ae189476 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -1235,6 +1235,62 @@ function codeAgentTurnStatusRefreshOptions(options = {}) { }; } +async function refreshAgentRunProjectionForCompatRead({ traceId, result = null, trace = null, options = {} } = {}) { + if (!safeTraceId(traceId) || !shouldRefreshAgentRunProjectionOnRead(result)) return { result, runnerTrace: trace, refreshError: null }; + const traceStore = options.traceStore ?? defaultCodeAgentTraceStore; + const refreshOptions = codeAgentTurnStatusRefreshOptions(options); + try { + const synced = await syncAgentRunChatResult({ + traceId, + currentResult: result, + options: refreshOptions, + traceStore, + forceResultSync: true + }); + const payload = synced.result ?? result; + if (payload && isTraceCommandTerminalStatus(payload.status)) { + await recordCodeAgentTerminalTurnStatusEffects({ + payload, + params: codeAgentReadSideTerminalParams(payload, options), + options: refreshOptions + }); + } + return { result: payload, runnerTrace: synced.runnerTrace ?? traceStore.snapshot(traceId), refreshError: null }; + } catch (error) { + traceStore.append(traceId, { + type: "projection_refresh", + status: "degraded", + label: "agentrun:read_side_sync:failed", + errorCode: error?.code ?? "agentrun_read_side_sync_failed", + message: error?.message ?? "AgentRun read-side projection refresh failed", + terminal: false, + valuesPrinted: false + }); + return { result, runnerTrace: traceStore.snapshot(traceId), refreshError: error }; + } +} + +function shouldRefreshAgentRunProjectionOnRead(result = {}) { + const agentRun = result?.agentRun && typeof result.agentRun === "object" ? result.agentRun : null; + if (!agentRun?.runId || !agentRun?.commandId) return false; + if (isTraceCommandTerminalStatus(result?.status) && agentRunTerminalResultRefreshSatisfied(result, result?.traceId)) return false; + return true; +} + +function codeAgentReadSideTerminalParams(payload = {}, options = {}) { + const session = payload.session && typeof payload.session === "object" ? payload.session : null; + const sessionReuse = payload.sessionReuse && typeof payload.sessionReuse === "object" ? payload.sessionReuse : null; + return { + traceId: payload.traceId, + conversationId: safeConversationId(payload.conversationId ?? session?.conversationId ?? sessionReuse?.conversationId) || null, + sessionId: safeSessionId(payload.sessionId ?? session?.sessionId ?? sessionReuse?.sessionId) || null, + threadId: safeOpaqueId(payload.threadId ?? session?.threadId ?? sessionReuse?.threadId) || null, + projectId: payload.projectId ?? null, + ownerUserId: payload.ownerUserId ?? options.actor?.id, + ownerRole: payload.ownerRole ?? options.actor?.role + }; +} + function codeAgentRequestScopedOptions(options = {}) { return { ...options, @@ -1267,14 +1323,19 @@ function forbiddenTurnSnapshot(traceId) { async function readCodeAgentCompatProjection(traceId, options = {}) { const readModel = createWorkbenchReadModel(options, options.actor ?? null); const session = await readModel.getSessionByTraceId(traceId); - const result = readModel.resultForTrace(traceId) ?? projectedSessionTraceResult(session, traceId); + let result = readModel.resultForTrace(traceId) ?? projectedSessionTraceResult(session, traceId); if (result && !canAccessOwnedResult(result, options.actor)) return forbiddenTurnSnapshot(traceId); if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) return forbiddenTurnSnapshot(traceId); const projectionTrace = await readModel.traceSnapshot(traceId); - const trace = traceSnapshotWithTerminalEvidence(projectionTrace, result, traceId, null); + let trace = traceSnapshotWithTerminalEvidence(projectionTrace, result, traceId, null); + const refreshed = await refreshAgentRunProjectionForCompatRead({ traceId, result, trace, options }); + if (refreshed.result && refreshed.result !== result) { + result = refreshed.result; + trace = traceSnapshotWithTerminalEvidence(refreshed.runnerTrace ?? trace, result, traceId, refreshed.refreshError ?? null); + } const projection = readModel.projectionDiagnostics({ traceId, result, trace }); const found = Boolean(result || session || (trace && trace.status !== "missing") || trace?.persisted === true); - return { result, session, trace, projection, found }; + return { result, session, trace, projection, found, refreshError: refreshed.refreshError ?? null }; } function projectedSessionTraceResult(session, traceId) {