mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: backfill AgentRun terminal projection on read

Browse Source
This commit is contained in:
lyon
2026-06-19 04:48:14 +08:00
parent 77818eb77a
commit 05e30ce2ba
3 changed files with 101 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/code-agent-agentrun-adapter.ts
+15 -5
View File
@@ -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,
internal/cloud/server-agent-chat.test.ts
+22 -11
View File
@@ -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()));
internal/cloud/server-code-agent-http.ts
+64 -3
View File
@@ -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) {