Merge pull request #1005 from pikasTech/fix/issue1000-trace-refresh-latency
fix: reduce Code Agent trace refresh latency
This commit is contained in:
Lyon
@@ -428,7 +428,7 @@ function isAgentRunCommandAlreadyClaimed(error) {
|
||||
return statusCode === 409 && /command\s+[^\s]+\s+is not pending:/u.test(message);
|
||||
}
|
||||
|
||||
export async function syncAgentRunChatResult({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore, appendResultEvent = true }) {
|
||||
export async function syncAgentRunChatResult({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore, appendResultEvent = true, refreshEvents = true }) {
|
||||
const initial = currentResult ?? await loadPersistedAgentRunResult(traceId, options);
|
||||
const mapped = initial ? await resolveAgentRunTraceCommandMapping({ traceId, mapped: initial, options }) : initial;
|
||||
if (!mapped?.agentRun?.runId || !mapped?.agentRun?.commandId) return { result: currentResult, runnerTrace: traceStore.snapshot(traceId), found: Boolean(currentResult) };
|
||||
@@ -436,9 +436,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 fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, mapping: { ...mapped.agentRun, traceSummary: mapped.traceSummary } });
|
||||
const events = eventsResponse.events;
|
||||
appendAgentRunEventsToTrace(traceStore, traceId, events, mapped.agentRun);
|
||||
const eventsResponse = refreshEvents ? await fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, mapping: { ...mapped.agentRun, traceSummary: mapped.traceSummary } }) : null;
|
||||
if (eventsResponse) appendAgentRunEventsToTrace(traceStore, traceId, eventsResponse.events, mapped.agentRun);
|
||||
const result = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/commands/${encodeURIComponent(mapped.agentRun.commandId)}/result`, {
|
||||
method: "GET",
|
||||
timeoutMs
|
||||
@@ -446,7 +445,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
||||
const nextMapping = {
|
||||
...mapped.agentRun,
|
||||
...agentRunResultRefs(result),
|
||||
lastSeq: agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq),
|
||||
lastSeq: eventsResponse ? agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq) : mapped.agentRun.lastSeq,
|
||||
status: result?.status ?? mapped.agentRun.status ?? "running",
|
||||
runStatus: result?.runStatus ?? mapped.agentRun.runStatus ?? null,
|
||||
commandState: result?.commandState ?? mapped.agentRun.commandState ?? null,
|
||||
|
||||
@@ -1863,6 +1863,90 @@ test("cloud api trace replays an earlier AgentRun command after same-run lastSeq
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud api running trace refreshes AgentRun events without waiting for command result (#1000)", async () => {
|
||||
const calls = [];
|
||||
const traceId = "trc_issue1000_running_trace";
|
||||
const runId = "run_issue1000_running_trace";
|
||||
const commandId = "cmd_issue1000_running_trace";
|
||||
const codeAgentChatResults = new Map();
|
||||
const events = [
|
||||
{ id: "evt_issue1000_created", runId, seq: 1, type: "backend_status", payload: { phase: "runner-job-created", commandId, jobName: "agentrun-v01-runner-issue1000", namespace: "agentrun-v01" }, createdAt: "2026-06-06T08:34:30.000Z" },
|
||||
{ id: "evt_issue1000_turn", runId, seq: 2, type: "backend_status", payload: { phase: "turn/started", commandId }, createdAt: "2026-06-06T08:34:31.000Z" },
|
||||
{ id: "evt_issue1000_tool", runId, seq: 3, type: "tool_call", payload: { method: "item/completed", type: "commandExecution", toolName: "commandExecution", itemId: "call_issue1000_hwpod", command: "hwpod list --available", status: "completed", exitCode: 0, outputSummary: "available=1", commandId }, createdAt: "2026-06-06T08:34:32.000Z" },
|
||||
{ id: "evt_issue1000_output", runId, seq: 4, type: "assistant_message", payload: { commandId, itemId: "msg_issue1000_running", text: "正在查看 HWPOD。" }, createdAt: "2026-06-06T08:34:33.000Z" }
|
||||
];
|
||||
const agentRunServer = createHttpServer(async (request, response) => {
|
||||
const url = new URL(request.url || "/", "http://127.0.0.1");
|
||||
calls.push({ method: request.method, path: url.pathname, search: url.search });
|
||||
const send = (data) => {
|
||||
response.writeHead(200, { "content-type": "application/json" });
|
||||
response.end(`${JSON.stringify({ ok: true, data, traceId: "trc_fake_issue1000_running" })}\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: 3, eventCount: 3 });
|
||||
}
|
||||
response.writeHead(404, { "content-type": "application/json" });
|
||||
response.end(`${JSON.stringify({ ok: false, message: `unexpected ${request.method} ${url.pathname}` })}\n`);
|
||||
});
|
||||
await new Promise((resolve) => agentRunServer.listen(0, "127.0.0.1", resolve));
|
||||
|
||||
const agentRunPort = agentRunServer.address().port;
|
||||
codeAgentChatResults.set(traceId, {
|
||||
accepted: true,
|
||||
status: "running",
|
||||
shortConnection: true,
|
||||
traceId,
|
||||
conversationId: "cnv_issue1000_running_trace",
|
||||
sessionId: "ses_issue1000_running_trace",
|
||||
threadId: "thread-issue1000-running-trace",
|
||||
ownerUserId: TEST_AGENT_ACTOR.id,
|
||||
ownerRole: TEST_AGENT_ACTOR.role,
|
||||
agentRun: { adapter: "agentrun-v01", managerUrl: `http://127.0.0.1:${agentRunPort}`, runId, commandId, traceId, providerTrace: { traceId }, lastSeq: 0, status: "running", commandState: "running", valuesPrinted: false },
|
||||
valuesPrinted: false
|
||||
});
|
||||
const server = createCloudApiServer({
|
||||
traceStore: createCodeAgentTraceStore(),
|
||||
codeAgentChatResults,
|
||||
env: {
|
||||
HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01",
|
||||
AGENTRUN_MGR_URL: `http://127.0.0.1:${agentRunPort}`,
|
||||
HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1",
|
||||
HWLAB_ENVIRONMENT: "v02",
|
||||
HWLAB_GITOPS_PROFILE: "v02"
|
||||
},
|
||||
accessController: {
|
||||
required: false,
|
||||
async authenticate() {
|
||||
return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION };
|
||||
}
|
||||
}
|
||||
});
|
||||
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
|
||||
try {
|
||||
const { port } = server.address();
|
||||
const response = await fetch(`http://127.0.0.1:${port}/v1/agent/chat/trace/${traceId}`, {
|
||||
headers: { cookie: "hwlab_session=test-stub-session" }
|
||||
});
|
||||
assert.equal(response.status, 200);
|
||||
const body = await response.json();
|
||||
assert.equal(body.status, "running");
|
||||
assert.equal(body.eventCount, 4);
|
||||
assert.ok(body.events.some((event) => event.label === "item/commandExecution:completed"));
|
||||
assert.ok(body.events.some((event) => event.label === "agentrun:assistant:message"));
|
||||
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/events`));
|
||||
assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), false);
|
||||
} finally {
|
||||
await new Promise((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
await new Promise((resolve, reject) => {
|
||||
agentRunServer.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud api repairs historical same-session AgentRun trace after lastTraceId advances (#955)", async () => {
|
||||
const calls = [];
|
||||
const ownerSessions = new Map();
|
||||
@@ -1984,7 +2068,8 @@ test("cloud api repairs historical same-session AgentRun trace after lastTraceId
|
||||
assert.equal(body.status, "completed");
|
||||
assert.equal(body.finalResponse.text, firstFinalText);
|
||||
assert.equal(body.agentRun.commandId, firstCommandId);
|
||||
assert.equal(body.traceSummary.sourceEventCount, 4);
|
||||
assert.equal(body.eventCount, 4);
|
||||
assert.equal(body.traceSummary.sourceEventCount, body.eventCount);
|
||||
assert.ok(body.events.some((event) => event.commandId === firstCommandId));
|
||||
assert.equal(body.events.some((event) => event.commandId === secondCommandId), false);
|
||||
assert.match(text, /d601-f103-v2/u);
|
||||
|
||||
@@ -33,6 +33,7 @@ import {
|
||||
const DEFAULT_CODE_AGENT_RESULT_TRACE_EVENT_LIMIT = 120;
|
||||
const DEFAULT_CODE_AGENT_PROJECT_ID = "prj_hwpod_workbench";
|
||||
const DEFAULT_CODE_AGENT_PROVIDER_PROFILE = "deepseek";
|
||||
const CODE_AGENT_TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled"]);
|
||||
const CODE_AGENT_PROVIDER_PROFILE_IDS = Object.freeze(["deepseek", "codex-api", "minimax-m3", "runtime-default"]);
|
||||
const CODE_AGENT_PROVIDER_PROFILE_LABELS = Object.freeze({
|
||||
"deepseek": "DeepSeek",
|
||||
@@ -1863,9 +1864,13 @@ export async function handleCodeAgentTraceHttp(request, response, url, options)
|
||||
}
|
||||
if (agentRunResult?.agentRun) {
|
||||
try {
|
||||
const synced = await syncAgentRunChatResult({ traceId, currentResult: agentRunResult, options, traceStore, appendResultEvent: false });
|
||||
agentRunResult = synced.result ?? agentRunResult;
|
||||
if (agentRunResult?.status && agentRunResult.status !== "running") {
|
||||
const snapshot = await refreshAgentRunTrace({ traceId, result: agentRunResult, options, traceStore });
|
||||
agentRunResult = options.codeAgentChatResults?.get?.(traceId) ?? agentRunResult;
|
||||
if (traceNeedsCommandResultSync(agentRunResult, snapshot)) {
|
||||
const synced = await syncAgentRunChatResult({ traceId, currentResult: agentRunResult, options, traceStore, appendResultEvent: false, refreshEvents: false });
|
||||
agentRunResult = synced.result ?? agentRunResult;
|
||||
}
|
||||
if (isTraceCommandTerminalStatus(agentRunResult?.status)) {
|
||||
recordCodeAgentConversationFact(agentRunResult, options);
|
||||
await recordCodeAgentSessionOwner({ payload: agentRunResult, params: agentRunResult, options, status: codeAgentOwnerStatusForResult(agentRunResult), preserveLastTraceId: true });
|
||||
}
|
||||
@@ -1880,11 +1885,6 @@ export async function handleCodeAgentTraceHttp(request, response, url, options)
|
||||
});
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await refreshAgentRunTrace({ traceId, result: agentRunResult, options, traceStore });
|
||||
} catch (error) {
|
||||
refreshError = error;
|
||||
}
|
||||
}
|
||||
|
||||
if (parts[5] === "stream") {
|
||||
@@ -1896,6 +1896,23 @@ export async function handleCodeAgentTraceHttp(request, response, url, options)
|
||||
sendJson(response, 200, traceSnapshotWithTerminalEvidence(snapshot, agentRunResult, traceId, refreshError));
|
||||
}
|
||||
|
||||
function traceNeedsCommandResultSync(result, snapshot) {
|
||||
if (!result?.agentRun) return false;
|
||||
const status = String(result.status ?? result.agentRun.status ?? result.agentRun.commandState ?? result.agentRun.terminalStatus ?? "running");
|
||||
if (isTraceCommandTerminalStatus(status)) return true;
|
||||
if (snapshotHasTerminalEvent(snapshot)) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
function isTraceCommandTerminalStatus(status) {
|
||||
return CODE_AGENT_TERMINAL_STATUSES.has(String(status ?? "").trim().toLowerCase().replace(/_/gu, "-"));
|
||||
}
|
||||
|
||||
function snapshotHasTerminalEvent(snapshot) {
|
||||
const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
|
||||
return events.some((event) => event?.terminal === true || (event?.type === "result" && isTraceCommandTerminalStatus(event?.status)));
|
||||
}
|
||||
|
||||
function traceSnapshotWithTerminalEvidence(snapshot, result, traceId, refreshError = null) {
|
||||
const evidence = agentRunTerminalTraceEvidence(result, traceId);
|
||||
if (snapshot?.status !== "missing") return traceSnapshotWithAttachedTerminalEvidence(snapshot, evidence, refreshError);
|
||||
|
||||
@@ -389,6 +389,71 @@ test("subscribeToTrace merges terminal result without text with trace assistant
|
||||
}
|
||||
});
|
||||
|
||||
test("subscribeToTrace starts result and trace polls in the same refresh tick (#1000)", async () => {
|
||||
const originalFetch = globalThis.fetch;
|
||||
const originalSetTimeout = globalThis.window?.setTimeout;
|
||||
const originalClearTimeout = globalThis.window?.clearTimeout;
|
||||
const fetches: string[] = [];
|
||||
let traceStartedBeforeResultResolved = false;
|
||||
let resolveResult: (() => void) | null = null;
|
||||
const resultGate = new Promise<void>((resolve) => { resolveResult = resolve; });
|
||||
globalThis.window = {
|
||||
...(globalThis.window ?? {}),
|
||||
setTimeout: ((callback: () => void) => { callback(); return 1; }) as typeof window.setTimeout,
|
||||
clearTimeout: (() => undefined) as typeof window.clearTimeout
|
||||
} as typeof window;
|
||||
globalThis.fetch = (async (input: RequestInfo | URL) => {
|
||||
const url = String(input);
|
||||
fetches.push(url);
|
||||
if (url.includes("/result/")) {
|
||||
await resultGate;
|
||||
return jsonResponse({ status: "completed", traceId: "trc_issue1000_parallel", assistantText: "最终回复" });
|
||||
}
|
||||
if (url.includes("/trace/")) {
|
||||
traceStartedBeforeResultResolved = true;
|
||||
resolveResult?.();
|
||||
return jsonResponse({
|
||||
status: "running",
|
||||
traceId: "trc_issue1000_parallel",
|
||||
events: [{ seq: 1, label: "agentrun:assistant:message", type: "assistant", status: "running", message: "trace 已更新" }],
|
||||
eventCount: 1
|
||||
});
|
||||
}
|
||||
return jsonResponse({});
|
||||
}) as typeof fetch;
|
||||
|
||||
try {
|
||||
const snapshots: TraceSnapshot[] = [];
|
||||
const completed = await new Promise<AgentChatResultResponse>((resolve, reject) => {
|
||||
void subscribeToTrace({
|
||||
traceId: "trc_issue1000_parallel",
|
||||
initial: { status: "running", traceId: "trc_issue1000_parallel", resultUrl: "/v1/agent/chat/result/trc_issue1000_parallel" },
|
||||
onActivity: () => undefined,
|
||||
onSnapshot: (snapshot) => snapshots.push(snapshot),
|
||||
onComplete: resolve,
|
||||
onInfrastructureError: (error) => reject(new Error(error)),
|
||||
signal: new AbortController().signal,
|
||||
inactivityTimeoutMs: 1000
|
||||
}).catch(reject);
|
||||
});
|
||||
assert.equal(traceStartedBeforeResultResolved, true);
|
||||
assert.equal(snapshots.length, 1);
|
||||
assert.equal(snapshots[0].eventCount, 1);
|
||||
assert.equal(completed.assistantText, "最终回复");
|
||||
assert.deepEqual(fetches, [
|
||||
"/v1/agent/chat/result/trc_issue1000_parallel",
|
||||
"/v1/agent/chat/trace/trc_issue1000_parallel?projectId=prj_hwpod_workbench"
|
||||
]);
|
||||
} finally {
|
||||
resolveResult?.();
|
||||
globalThis.fetch = originalFetch;
|
||||
if (globalThis.window) {
|
||||
globalThis.window.setTimeout = originalSetTimeout ?? globalThis.window.setTimeout;
|
||||
globalThis.window.clearTimeout = originalClearTimeout ?? globalThis.window.clearTimeout;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
test("replayFullTrace retries transient trace API failures before returning an empty replay", async () => {
|
||||
const originalFetch = globalThis.fetch;
|
||||
const originalSetTimeout = globalThis.window?.setTimeout;
|
||||
|
||||
@@ -305,8 +305,10 @@ export async function subscribeToTrace(config: TraceSubscriptionConfig): Promise
|
||||
await new Promise<void>((resolve) => window.setTimeout(resolve, TRACE_POLL_INTERVAL_MS));
|
||||
if (signal.aborted) return;
|
||||
|
||||
// Step 1: poll the result endpoint for terminal status.
|
||||
const resultPolled = await api.getAgentChatResult(resultUrl, inactivityTimeoutMs, null);
|
||||
const [resultPolled, tracePolled] = await Promise.all([
|
||||
api.getAgentChatResult(resultUrl, inactivityTimeoutMs, null),
|
||||
api.getAgentChatResult(traceUrl, inactivityTimeoutMs, null)
|
||||
]);
|
||||
if (signal.aborted) return;
|
||||
|
||||
if (resultPolled.ok && resultPolled.data) {
|
||||
@@ -330,11 +332,6 @@ export async function subscribeToTrace(config: TraceSubscriptionConfig): Promise
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 2: poll the trace endpoint for live events.
|
||||
if (signal.aborted) return;
|
||||
const tracePolled = await api.getAgentChatResult(traceUrl, inactivityTimeoutMs, null);
|
||||
if (signal.aborted) return;
|
||||
|
||||
if (tracePolled.ok && tracePolled.data) {
|
||||
onActivity();
|
||||
const events = Array.isArray(tracePolled.data.events) ? tracePolled.data.events : [];
|
||||
|
||||
Reference in New Issue
Block a user