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

fix: stabilize AgentRun trace replay and final rows

Browse Source
This commit is contained in:
Codex
2026-06-06 10:33:14 +08:00
parent e2c33ae3e6
commit dd0121c93c
5 changed files with 303 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/code-agent-agentrun-adapter.ts
+61 -14
View File
@@ -435,11 +435,8 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
const fetchImpl = options.fetchImpl ?? globalThis.fetch;
const managerUrl = resolveAgentRunManagerUrl(env, mapped.agentRun.managerUrl);
const timeoutMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_HTTP_TIMEOUT_MS, 20_000);
const eventsResponse = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/events?afterSeq=${encodeURIComponent(String(mapped.agentRun.lastSeq ?? 0))}&limit=500`, {
method: "GET",
timeoutMs
});
const events = Array.isArray(eventsResponse?.items) ? eventsResponse.items : [];
const eventsResponse = await fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, mapping: { ...mapped.agentRun, traceSummary: mapped.traceSummary } });
const events = eventsResponse.events;
appendAgentRunEventsToTrace(traceStore, traceId, events, mapped.agentRun);
const result = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/commands/${encodeURIComponent(mapped.agentRun.commandId)}/result`, {
method: "GET",
@@ -448,7 +445,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
const nextMapping = {
...mapped.agentRun,
...agentRunResultRefs(result),
lastSeq: Math.max(Number(mapped.agentRun.lastSeq ?? 0), Number(result?.lastSeq ?? 0), ...events.map((event) => Number(event?.seq ?? 0))),
lastSeq: agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq),
status: result?.status ?? mapped.agentRun.status ?? "running",
runStatus: result?.runStatus ?? mapped.agentRun.runStatus ?? null,
commandState: result?.commandState ?? mapped.agentRun.commandState ?? null,
@@ -469,13 +466,10 @@ export async function refreshAgentRunTrace({ traceId, result = null, options = {
const fetchImpl = options.fetchImpl ?? globalThis.fetch;
const managerUrl = resolveAgentRunManagerUrl(env, mapped.agentRun.managerUrl);
const timeoutMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_HTTP_TIMEOUT_MS, 20_000);
const eventsResponse = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/events?afterSeq=${encodeURIComponent(String(mapped.agentRun.lastSeq ?? 0))}&limit=500`, {
method: "GET",
timeoutMs
});
const events = Array.isArray(eventsResponse?.items) ? eventsResponse.items : [];
const eventsResponse = await fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, mapping: { ...mapped.agentRun, traceSummary: mapped.traceSummary } });
const events = eventsResponse.events;
appendAgentRunEventsToTrace(traceStore, traceId, events, mapped.agentRun);
const lastSeq = Math.max(Number(mapped.agentRun.lastSeq ?? 0), ...events.map((event) => Number(event?.seq ?? 0)));
const lastSeq = agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq);
if (lastSeq !== Number(mapped.agentRun.lastSeq ?? 0)) {
options.codeAgentChatResults?.set?.(traceId, { ...mapped, agentRun: { ...mapped.agentRun, lastSeq, updatedAt: nowIso(options.now) } });
}
@@ -1293,11 +1287,64 @@ function appendAgentRunEventsToTrace(traceStore, traceId, events, mapping = {})
}
}
async function fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, mapping = {} }) {
const runId = requiredString(mapping.runId, "runId");
const currentCommandId = typeof mapping.commandId === "string" ? mapping.commandId : "";
const { afterSeq, endSeq } = agentRunTraceReplayWindow(mapping);
const path = `/api/v1/runs/${encodeURIComponent(runId)}/events?afterSeq=${encodeURIComponent(String(afterSeq))}&limit=500`;
const response = await agentRunJson(fetchImpl, managerUrl, path, { method: "GET", timeoutMs });
const rawEvents = Array.isArray(response?.items) ? response.items : [];
const events = rawEvents.filter((event) => agentRunEventBelongsToTrace(event, { currentCommandId, afterSeq, endSeq }));
return {
events,
afterSeq,
endSeq,
commandFiltered: Boolean(currentCommandId),
maxSeq: Math.max(afterSeq, ...rawEvents.map((event) => Number(event?.seq ?? 0))),
traceLastSeq: Math.max(afterSeq, ...events.map((event) => Number(event?.seq ?? 0)).filter(Number.isFinite))
};
}
function agentRunTraceCursorSeq(eventsResponse = {}, previousLastSeq = 0) {
const afterSeq = Number(eventsResponse.afterSeq ?? 0);
const endSeq = Number(eventsResponse.endSeq ?? 0);
const traceLastSeq = Number(eventsResponse.traceLastSeq ?? 0);
if (Number.isFinite(traceLastSeq) && traceLastSeq > afterSeq) return Math.floor(traceLastSeq);
if (Number.isFinite(endSeq) && endSeq > 0) return Math.floor(endSeq);
if (eventsResponse.commandFiltered === true) return Math.max(Number(previousLastSeq ?? 0), afterSeq);
return Math.max(Number(previousLastSeq ?? 0), Number(eventsResponse.maxSeq ?? 0));
}
function agentRunTraceReplayWindow(mapping = {}) {
const eventStartSeq = Number(mapping.eventStartSeq ?? mapping.commandStartSeq ?? mapping.startSeq ?? 0);
const summary = mapping.traceSummary && typeof mapping.traceSummary === "object" ? mapping.traceSummary : null;
const summaryAgentRun = summary?.agentRun && typeof summary.agentRun === "object" ? summary.agentRun : null;
const summaryLastSeq = Number(summaryAgentRun?.lastSeq ?? summary?.lastSeq ?? 0);
const currentLastSeq = Number(mapping.lastSeq ?? 0);
const endSeq = Number.isFinite(summaryLastSeq) && summaryLastSeq > 0 ? Math.floor(summaryLastSeq) : 0;
if (Number.isFinite(eventStartSeq) && eventStartSeq > 0) return { afterSeq: Math.max(0, Math.floor(eventStartSeq) - 1), endSeq };
if (endSeq > 0) return { afterSeq: Math.max(0, endSeq - 500), endSeq };
if (Number.isFinite(currentLastSeq) && currentLastSeq > 0) return { afterSeq: Math.floor(currentLastSeq), endSeq: 0 };
return { afterSeq: 0, endSeq: 0 };
}
function agentRunEventBelongsToTrace(event, { currentCommandId = "", afterSeq = 0, endSeq = 0 } = {}) {
const eventCommandId = agentRunEventCommandId(event);
const seq = Number(event?.seq ?? 0);
if (currentCommandId && eventCommandId && eventCommandId !== currentCommandId) return false;
if (endSeq > 0 && Number.isFinite(seq)) return seq > afterSeq && seq <= endSeq;
return true;
}
function agentRunEventCommandId(event) {
const payload = event?.payload && typeof event.payload === "object" ? event.payload : {};
return typeof payload.commandId === "string" ? payload.commandId : "";
}
function isForeignAgentRunCommandEvent(event, mapping = {}) {
const currentCommandId = typeof mapping.commandId === "string" ? mapping.commandId : "";
if (!currentCommandId) return false;
const payload = event?.payload && typeof event.payload === "object" ? event.payload : {};
const eventCommandId = typeof payload.commandId === "string" ? payload.commandId : "";
const eventCommandId = agentRunEventCommandId(event);
return Boolean(eventCommandId && eventCommandId !== currentCommandId);
}