Merge pull request #2348 from pikasTech/fix/1445-lane-c-agentrun-backoff
fix: add AgentRun projection polling backoff
This commit is contained in:
Lyon
@@ -20,6 +20,25 @@ import {
|
||||
isAgentRunManagerInternalServiceHost,
|
||||
resolveAgentRunManagerUrl
|
||||
} from "./code-agent-agentrun-runtime.ts";
|
||||
import {
|
||||
agentRunErrorWithPolling,
|
||||
agentRunEventCommandId,
|
||||
agentRunMetricStatusForError,
|
||||
agentRunPollingRetryableError,
|
||||
agentRunPollingRootCause,
|
||||
agentRunProjectionPollCount,
|
||||
agentRunProjectionPollingState,
|
||||
agentRunTerminalConsistencyGap,
|
||||
agentRunTerminalResultRefreshSatisfied,
|
||||
agentRunTerminalStatusFromEvents,
|
||||
agentRunTraceCursorSeq,
|
||||
emitAgentRunProjectionPollingSpan,
|
||||
fetchAgentRunEventsForTrace,
|
||||
measuredAgentRunEventsFetch,
|
||||
measuredAgentRunResultFetch,
|
||||
nonNegativeNumber,
|
||||
recordAgentRunEventsProjectionMetrics
|
||||
} from "./code-agent-agentrun-projection-polling.ts";
|
||||
import {
|
||||
agentRunAlreadyTerminalCancelPayload,
|
||||
agentRunCancelBlockedPayload,
|
||||
@@ -699,25 +718,47 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
||||
const runnerTrace = mapped.runnerTrace ?? traceStore.snapshot(traceId);
|
||||
return { result: mapped, runnerTrace, found: true, eventsRefreshed: false, resultSynced: false, terminalRefreshSkipped: true, localCancelSealed: true };
|
||||
}
|
||||
const env = options.env ?? process.env;
|
||||
const performanceStore = options.backendPerformanceStore ?? currentBackendPerformanceContext();
|
||||
if (!forceResultSync && options.skipAgentRunTerminalRefreshIfComplete === true && agentRunTerminalResultRefreshSatisfied(mapped, traceId)) {
|
||||
if (!forceResultSync && agentRunTerminalResultRefreshSatisfied(mapped, traceId)) {
|
||||
const runnerTrace = mapped.runnerTrace ?? traceStore.snapshot(traceId);
|
||||
return { result: mapped, runnerTrace, found: true, eventsRefreshed: false, resultSynced: false, terminalRefreshSkipped: true };
|
||||
const polling = agentRunProjectionPollingState({ env, terminalFromEvents: "completed", rootCause: "terminal_result_sealed", terminalConsistencyGap: 0 });
|
||||
emitAgentRunProjectionPollingSpan({ traceId, env, agentRun: mapped?.agentRun, polling, terminalRefreshSkipped: true });
|
||||
return { result: mapped, runnerTrace, found: true, eventsRefreshed: false, resultSynced: false, terminalRefreshSkipped: true, polling };
|
||||
}
|
||||
if (!mapped?.agentRun?.runId || !mapped?.agentRun?.commandId) return { result: currentResult, runnerTrace: traceStore.snapshot(traceId), found: Boolean(currentResult), eventsRefreshed: false, resultSynced: false };
|
||||
const env = options.env ?? process.env;
|
||||
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 projectionState = await loadWorkbenchProjectionStateForAgentRun(options.runtimeStore, { traceId, agentRun: mapped.agentRun });
|
||||
const projectionRetryParams = retryParams ?? projectionState?.retryParams ?? null;
|
||||
const projectionPollCount = agentRunProjectionPollCount(projectionState);
|
||||
const fetchPlan = buildAgentRunProjectionEventsFetchPlan({
|
||||
projectionState,
|
||||
agentRun: mapped.agentRun,
|
||||
pageLimit: env.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT
|
||||
});
|
||||
const previousLastSeq = Number(fetchPlan.afterSeq ?? mapped.agentRun.lastSeq ?? 0);
|
||||
const eventsResponse = refreshEvents ? await measuredAgentRunEventsFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapping: { ...mapped.agentRun, lastSeq: fetchPlan.afterSeq, afterSeqOverride: fetchPlan.afterSeq, eventsPageLimit: fetchPlan.limit, traceSummary: mapped.traceSummary } }) : null;
|
||||
let eventsResponse = null;
|
||||
try {
|
||||
eventsResponse = refreshEvents ? await measuredAgentRunEventsFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapping: { ...mapped.agentRun, lastSeq: fetchPlan.afterSeq, afterSeqOverride: fetchPlan.afterSeq, eventsPageLimit: fetchPlan.limit, traceSummary: mapped.traceSummary, pollCount: projectionPollCount, backoffMs: nonNegativeNumber(projectionState?.backoffMs), rootCause: projectionState?.rootCause ?? null, terminalConsistencyGap: nonNegativeNumber(projectionState?.terminalConsistencyGap) } }) : null;
|
||||
} catch (error) {
|
||||
const polling = agentRunProjectionPollingState({ projectionState, env, error, rootCause: agentRunPollingRootCause(error), terminalConsistencyGap: nonNegativeNumber(projectionState?.terminalConsistencyGap) });
|
||||
await writeWorkbenchProjectionStateForAgentRun(options.runtimeStore, {
|
||||
currentState: projectionState,
|
||||
result: mapped,
|
||||
agentRun: { ...mapped.agentRun, lastSeq: previousLastSeq },
|
||||
eventsResponse: null,
|
||||
resultSyncState: projectionState?.resultSyncState ?? null,
|
||||
projectionStatus: "degraded",
|
||||
projectionHealth: "degraded",
|
||||
retryParams: projectionRetryParams,
|
||||
polling,
|
||||
error,
|
||||
now: options.now
|
||||
});
|
||||
throw agentRunErrorWithPolling(error, polling);
|
||||
}
|
||||
if (eventsResponse) appendAgentRunEventsToTrace(traceStore, traceId, eventsResponse.events, mapped.agentRun);
|
||||
const nextLastSeq = eventsResponse ? agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq) : mapped.agentRun.lastSeq;
|
||||
const backendProfile = resolveAgentRunBackendProfile(env, {
|
||||
@@ -730,6 +771,17 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
||||
});
|
||||
recordAgentRunEventsProjectionMetrics(performanceStore, eventsResponse, previousLastSeq, nextLastSeq);
|
||||
const terminalFromEvents = agentRunTerminalStatusFromEvents(eventsResponse?.events);
|
||||
const cursorAdvance = Math.max(0, Number(nextLastSeq ?? 0) - Number(previousLastSeq ?? 0));
|
||||
const terminalConsistencyGap = agentRunTerminalConsistencyGap({ mapped, eventsResponse });
|
||||
const projectionPolling = agentRunProjectionPollingState({
|
||||
projectionState,
|
||||
env,
|
||||
pollCount: projectionPollCount,
|
||||
cursorAdvance,
|
||||
terminalFromEvents,
|
||||
rootCause: terminalFromEvents ? "terminal_from_events" : cursorAdvance > 0 ? "cursor_advanced" : "projection_no_progress",
|
||||
terminalConsistencyGap
|
||||
});
|
||||
await writeWorkbenchProjectionStateForAgentRun(options.runtimeStore, {
|
||||
currentState: projectionState,
|
||||
result: mapped,
|
||||
@@ -737,6 +789,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
||||
eventsResponse,
|
||||
resultSyncState: terminalFromEvents ? "pending" : null,
|
||||
projectionStatus: terminalFromEvents ? "terminal" : null,
|
||||
polling: projectionPolling,
|
||||
retryParams: projectionRetryParams,
|
||||
now: options.now
|
||||
});
|
||||
@@ -755,24 +808,28 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
||||
});
|
||||
const payload = decorateAgentRunRunningResult({ base: { ...mapped, status: "running", agentRun: nextMapping, updatedAt: nowIso(options.now) }, mapping: nextMapping, traceStore, traceId });
|
||||
options.codeAgentChatResults?.set?.(traceId, payload);
|
||||
return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: false };
|
||||
return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: false, polling: projectionPolling };
|
||||
}
|
||||
let result;
|
||||
try {
|
||||
result = await measuredAgentRunResultFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapped, eventsResponse });
|
||||
} catch (error) {
|
||||
const retryable = agentRunPollingRetryableError(error);
|
||||
const polling = agentRunProjectionPollingState({ projectionState, env, pollCount: projectionPollCount, cursorAdvance, terminalFromEvents, error, rootCause: agentRunPollingRootCause(error), terminalConsistencyGap });
|
||||
await writeWorkbenchProjectionStateForAgentRun(options.runtimeStore, {
|
||||
currentState: projectionState,
|
||||
result: mapped,
|
||||
agentRun: { ...mapped.agentRun, lastSeq: nextLastSeq },
|
||||
eventsResponse,
|
||||
resultSyncState: agentRunMetricStatusForError(error) === "timeout" ? "timed_out" : "failed",
|
||||
projectionStatus: terminalFromEvents ? "terminal" : null,
|
||||
resultSyncState: retryable ? "pending" : agentRunMetricStatusForError(error) === "timeout" ? "timed_out" : "failed",
|
||||
projectionStatus: retryable ? "degraded" : terminalFromEvents ? "terminal" : null,
|
||||
projectionHealth: "degraded",
|
||||
retryParams: projectionRetryParams,
|
||||
polling,
|
||||
error,
|
||||
now: options.now
|
||||
});
|
||||
throw error;
|
||||
throw agentRunErrorWithPolling(error, polling);
|
||||
}
|
||||
const nextMapping = withAgentRunRunnerJobCount({
|
||||
...mapped.agentRun,
|
||||
@@ -834,6 +891,15 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
||||
const base = { ...mapped, agentRun: nextMapping, updatedAt: nowIso(options.now) };
|
||||
const payload = agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, appendResultEvent });
|
||||
options.codeAgentChatResults?.set?.(traceId, payload);
|
||||
const resultPolling = agentRunProjectionPollingState({
|
||||
projectionState,
|
||||
env,
|
||||
pollCount: projectionPollCount,
|
||||
cursorAdvance,
|
||||
terminalFromEvents: isAgentRunTerminalStatus(payload?.status ?? nextMapping.terminalStatus ?? nextMapping.commandState) ? (nextMapping.terminalStatus ?? payload?.status) : terminalFromEvents,
|
||||
rootCause: isAgentRunTerminalStatus(payload?.status ?? nextMapping.terminalStatus ?? nextMapping.commandState) ? "terminal_result_synced" : "result_synced",
|
||||
terminalConsistencyGap: agentRunTerminalConsistencyGap({ mapped: payload, eventsResponse, result })
|
||||
});
|
||||
await writeWorkbenchProjectionStateForAgentRun(options.runtimeStore, {
|
||||
currentState: projectionState,
|
||||
result: payload,
|
||||
@@ -841,10 +907,11 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op
|
||||
eventsResponse,
|
||||
resultSyncState: "synced",
|
||||
projectionStatus: isAgentRunTerminalStatus(payload?.status) ? "terminal" : null,
|
||||
polling: resultPolling,
|
||||
retryParams: projectionRetryParams,
|
||||
now: options.now
|
||||
});
|
||||
return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: true };
|
||||
return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: true, polling: resultPolling };
|
||||
}
|
||||
|
||||
async function loadWorkbenchProjectionStateForAgentRun(runtimeStore, { traceId, agentRun = {} } = {}) {
|
||||
@@ -875,53 +942,6 @@ function agentRunCommandResultSyncRequired(mapped = {}, eventsResponse = null) {
|
||||
return Boolean(agentRunTerminalStatusFromEvents(eventsResponse?.events));
|
||||
}
|
||||
|
||||
export function agentRunTerminalResultRefreshSatisfied(mapped = {}, traceId = null) {
|
||||
const safeId = safeTraceId(traceId ?? mapped?.traceId);
|
||||
const agentRun = mapped?.agentRun && typeof mapped.agentRun === "object" ? mapped.agentRun : null;
|
||||
const traceSummary = mapped?.traceSummary && typeof mapped.traceSummary === "object" ? mapped.traceSummary : null;
|
||||
const finalResponse = mapped?.finalResponse && typeof mapped.finalResponse === "object" ? mapped.finalResponse : null;
|
||||
const providerTrace = agentRun?.providerTrace && typeof agentRun.providerTrace === "object"
|
||||
? agentRun.providerTrace
|
||||
: mapped?.providerTrace && typeof mapped.providerTrace === "object"
|
||||
? mapped.providerTrace
|
||||
: null;
|
||||
const commandId = text(agentRun?.commandId);
|
||||
const runId = text(agentRun?.runId);
|
||||
if (!safeId || !runId || !commandId) return false;
|
||||
if (!isAgentRunTerminalStatus(mapped?.status ?? agentRun?.terminalStatus ?? agentRun?.commandState ?? agentRun?.status)) return false;
|
||||
if (safeTraceId(mapped?.traceId ?? agentRun?.traceId) !== safeId) return false;
|
||||
if (safeTraceId(providerTrace?.traceId) !== safeId) return false;
|
||||
const finalTraceId = safeTraceId(finalResponse?.traceId ?? safeId);
|
||||
if (finalTraceId && finalTraceId !== safeId) return false;
|
||||
if (!String(finalResponse?.text ?? "").trim()) return false;
|
||||
if (traceSummary?.source !== "agentrun-command-result") return false;
|
||||
if (safeTraceId(traceSummary.traceId ?? safeId) !== safeId) return false;
|
||||
const summaryAgentRun = traceSummary.agentRun && typeof traceSummary.agentRun === "object" ? traceSummary.agentRun : null;
|
||||
if (text(summaryAgentRun?.runId) !== runId) return false;
|
||||
if (text(summaryAgentRun?.commandId) !== commandId) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
function agentRunTerminalStatusFromEvents(events = []) {
|
||||
if (!Array.isArray(events)) return null;
|
||||
for (const event of [...events].reverse()) {
|
||||
const payload = event?.payload && typeof event.payload === "object" ? event.payload : {};
|
||||
const phase = payload.phase ?? event?.phase;
|
||||
const candidates = [
|
||||
event?.type === "terminal_status" ? payload.terminalStatus ?? event?.terminalStatus ?? payload.status ?? event?.status : null,
|
||||
phase === "command-terminal" ? payload.terminalStatus ?? event?.terminalStatus ?? payload.status ?? event?.status : null,
|
||||
phase === "turn:completed" || phase === "turn/steer:completed" ? "completed" : null,
|
||||
event?.terminal === true ? payload.terminalStatus ?? event?.terminalStatus ?? payload.status ?? event?.status ?? "completed" : null,
|
||||
payload.terminal === true ? payload.terminalStatus ?? event?.terminalStatus ?? payload.status ?? event?.status ?? "completed" : null
|
||||
];
|
||||
for (const value of candidates) {
|
||||
const status = normalizeAgentRunStatus(value);
|
||||
if (isAgentRunTerminalStatus(status)) return status;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function isAgentRunTerminalStatus(value) {
|
||||
const status = normalizeAgentRunStatus(value);
|
||||
return status === "timeout" || TERMINAL_RUN_STATUSES.has(status);
|
||||
@@ -949,7 +969,7 @@ async function resolveAgentRunTraceCommandMapping({ traceId, mapped, options = {
|
||||
const traceSummaryCommandId = text(mapped?.traceSummary?.agentRun?.commandId);
|
||||
const commandId = text(agentRun.commandId);
|
||||
const traceFieldsMatch = agentRunTraceId === safeId && providerTraceId === safeId && (!traceSummaryCommandId || traceSummaryCommandId === commandId);
|
||||
if (traceFieldsMatch && (mapped.status === "running" || (options.skipAgentRunTerminalRefreshIfComplete === true && agentRunTerminalResultRefreshSatisfied(mapped, safeId)))) return mapped;
|
||||
if (traceFieldsMatch && (mapped.status === "running" || agentRunTerminalResultRefreshSatisfied(mapped, safeId))) return mapped;
|
||||
const env = options.env ?? process.env;
|
||||
const fetchImpl = options.fetchImpl ?? globalThis.fetch;
|
||||
const managerUrl = resolveAgentRunManagerUrl(env, agentRun.managerUrl);
|
||||
@@ -2388,180 +2408,6 @@ function appendAgentRunEventsToTrace(traceStore, traceId, events, mapping = {})
|
||||
}
|
||||
}
|
||||
|
||||
async function fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, env, mapping = {} }) {
|
||||
const runId = requiredString(mapping.runId, "runId");
|
||||
const currentCommandId = typeof mapping.commandId === "string" ? mapping.commandId : "";
|
||||
const replayWindow = agentRunTraceReplayWindow(mapping);
|
||||
const afterSeqOverride = Number(mapping.afterSeqOverride);
|
||||
const afterSeq = Number.isFinite(afterSeqOverride) && afterSeqOverride >= 0 ? Math.floor(afterSeqOverride) : replayWindow.afterSeq;
|
||||
const endSeq = Number.isFinite(afterSeqOverride) && afterSeqOverride >= 0 ? 0 : replayWindow.endSeq;
|
||||
const limit = parsePositiveInteger(mapping.eventsPageLimit ?? env?.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT, 500);
|
||||
const path = `/api/v1/runs/${encodeURIComponent(runId)}/events?afterSeq=${encodeURIComponent(String(afterSeq))}&limit=${encodeURIComponent(String(limit))}`;
|
||||
const response = await agentRunJson(fetchImpl, managerUrl, path, { method: "GET", timeoutMs, env });
|
||||
const rawEvents = Array.isArray(response?.items) ? response.items : [];
|
||||
const events = rawEvents.filter((event) => agentRunEventBelongsToTrace(event, { currentCommandId, afterSeq, endSeq }));
|
||||
const maxSeq = Math.max(afterSeq, ...rawEvents.map((event) => Number(event?.seq ?? 0)).filter(Number.isFinite));
|
||||
const traceLastSeq = Math.max(afterSeq, ...events.map((event) => Number(event?.seq ?? 0)).filter(Number.isFinite));
|
||||
void emitCodeAgentOtelSpan("projection_sync", mapping.traceId ?? mapping.hwlabTraceId ?? mapping.businessTraceId, env, {
|
||||
attributes: {
|
||||
runId,
|
||||
commandId: currentCommandId || null,
|
||||
afterSeq,
|
||||
endSeq,
|
||||
limit,
|
||||
rawEventCount: rawEvents.length,
|
||||
eventCount: events.length,
|
||||
commandFiltered: Boolean(currentCommandId),
|
||||
maxSeq,
|
||||
traceLastSeq,
|
||||
terminalFromRawEvents: agentRunTerminalStatusFromEvents(rawEvents),
|
||||
terminalFromEvents: agentRunTerminalStatusFromEvents(events)
|
||||
}
|
||||
});
|
||||
return {
|
||||
events,
|
||||
afterSeq,
|
||||
endSeq,
|
||||
rawEventCount: rawEvents.length,
|
||||
commandFiltered: Boolean(currentCommandId),
|
||||
maxSeq,
|
||||
traceLastSeq
|
||||
};
|
||||
}
|
||||
|
||||
async function measuredAgentRunEventsFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapping = {} }) {
|
||||
const startedAt = nowMs();
|
||||
try {
|
||||
const response = await fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, env, mapping });
|
||||
performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "agentrun_events_fetch", status: "ok", durationMs: nowMs() - startedAt });
|
||||
return response;
|
||||
} catch (error) {
|
||||
performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "agentrun_events_fetch", status: agentRunMetricStatusForError(error), durationMs: nowMs() - startedAt });
|
||||
performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "agentrun_events_fetch_failed" });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function measuredAgentRunResultFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapped = {}, eventsResponse = null }) {
|
||||
const startedAt = nowMs();
|
||||
const path = `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/commands/${encodeURIComponent(mapped.agentRun.commandId)}/result`;
|
||||
try {
|
||||
const result = await agentRunJson(fetchImpl, managerUrl, path, { method: "GET", timeoutMs, env });
|
||||
const eventCount = agentRunResultEventCount(result, eventsResponse);
|
||||
performanceStore?.recordAgentRunResult?.({
|
||||
durationMs: nowMs() - startedAt,
|
||||
eventCount,
|
||||
eventsScanned: eventCount,
|
||||
pagesScanned: estimateAgentRunResultPages(eventCount),
|
||||
status: result?.terminalStatus ?? result?.status ?? "ok",
|
||||
budget: timeoutMs
|
||||
});
|
||||
return result;
|
||||
} catch (error) {
|
||||
const status = agentRunMetricStatusForError(error);
|
||||
const eventCount = agentRunResultEventCount(null, eventsResponse);
|
||||
performanceStore?.recordAgentRunResult?.({
|
||||
durationMs: nowMs() - startedAt,
|
||||
eventCount,
|
||||
eventsScanned: eventCount,
|
||||
pagesScanned: estimateAgentRunResultPages(eventCount),
|
||||
status,
|
||||
budget: timeoutMs,
|
||||
timedOut: status === "timeout"
|
||||
});
|
||||
performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "agentrun_result_failed" });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
function recordAgentRunEventsProjectionMetrics(performanceStore, eventsResponse = null, previousLastSeq = 0, nextLastSeq = 0) {
|
||||
if (!performanceStore || !eventsResponse) return;
|
||||
const cursorAdvance = Math.max(0, Number(nextLastSeq ?? 0) - Number(previousLastSeq ?? 0));
|
||||
if (cursorAdvance > 0) {
|
||||
performanceStore.recordWorkbenchProjectionCursorAdvance?.({
|
||||
status: "advanced",
|
||||
reason: eventsResponse.commandFiltered ? "command_filtered_events" : "run_events",
|
||||
projectionStatus: "projecting",
|
||||
source: "agentrun_v01",
|
||||
count: cursorAdvance
|
||||
});
|
||||
}
|
||||
const counts = new Map();
|
||||
for (const event of eventsResponse.events ?? []) {
|
||||
const kind = String(event?.type ?? event?.payload?.phase ?? "unknown").trim() || "unknown";
|
||||
counts.set(kind, (counts.get(kind) ?? 0) + 1);
|
||||
}
|
||||
if (counts.size === 0 && Number(eventsResponse.rawEventCount ?? 0) > 0) counts.set("filtered", Number(eventsResponse.rawEventCount));
|
||||
for (const [eventKind, count] of counts.entries()) {
|
||||
performanceStore.recordWorkbenchProjectorEvents?.({ eventKind, status: "ok", count });
|
||||
}
|
||||
}
|
||||
|
||||
function agentRunResultEventCount(result = null, eventsResponse = null) {
|
||||
const candidates = [
|
||||
result?.traceSummary?.sourceEventCount,
|
||||
result?.traceSummary?.eventCount,
|
||||
result?.sourceEventCount,
|
||||
result?.eventCount,
|
||||
result?.providerTrace?.eventCount,
|
||||
eventsResponse?.maxSeq,
|
||||
eventsResponse?.rawEventCount,
|
||||
eventsResponse?.events?.length
|
||||
];
|
||||
for (const value of candidates) {
|
||||
const count = Number(value);
|
||||
if (Number.isFinite(count) && count >= 0) return Math.floor(count);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
function estimateAgentRunResultPages(eventCount) {
|
||||
const count = Number(eventCount);
|
||||
if (!Number.isFinite(count) || count <= 0) return 0;
|
||||
return Math.max(1, Math.ceil(count / 500));
|
||||
}
|
||||
|
||||
function agentRunMetricStatusForError(error) {
|
||||
return error?.code === "agentrun_timeout" || error?.name === "AbortError" || Number(error?.statusCode) === 504 ? "timeout" : "error";
|
||||
}
|
||||
|
||||
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 : {};
|
||||
if (typeof event?.commandId === "string") return event.commandId;
|
||||
return typeof payload.commandId === "string" ? payload.commandId : "";
|
||||
}
|
||||
|
||||
function isForeignAgentRunCommandEvent(event, mapping = {}) {
|
||||
const currentCommandId = typeof mapping.commandId === "string" ? mapping.commandId : "";
|
||||
if (!currentCommandId) return false;
|
||||
|
||||
@@ -0,0 +1,350 @@
|
||||
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
|
||||
// Responsibility: AgentRun projection polling/backoff I/O, durable polling state, and projection_sync observability attributes.
|
||||
|
||||
import { agentRunJson } from "./code-agent-agentrun-runtime.ts";
|
||||
import { emitCodeAgentOtelSpan } from "./otel-trace.ts";
|
||||
import { parsePositiveInteger, safeTraceId, text } from "./server-http-utils.ts";
|
||||
|
||||
const DEFAULT_AGENTRUN_PROJECTION_SYNC_BACKOFF_INITIAL_MS = 1_000;
|
||||
const DEFAULT_AGENTRUN_PROJECTION_SYNC_BACKOFF_MAX_MS = 30_000;
|
||||
const AGENTRUN_TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "cancelled", "canceled", "timeout"]);
|
||||
|
||||
export function agentRunProjectionPollCount(projectionState = null) {
|
||||
return nonNegativeNumber(projectionState?.pollCount) + 1;
|
||||
}
|
||||
|
||||
export function agentRunProjectionPollingState({ projectionState = null, env = process.env, pollCount = null, cursorAdvance = 0, terminalFromEvents = null, error = null, rootCause = null, terminalConsistencyGap = 0 } = {}) {
|
||||
const retryable = agentRunPollingRetryableError(error);
|
||||
const noProgress = !error && !terminalFromEvents && Number(cursorAdvance ?? 0) <= 0;
|
||||
const previousNoProgress = nonNegativeNumber(projectionState?.noProgressPollCount);
|
||||
const noProgressPollCount = retryable || noProgress ? previousNoProgress + 1 : 0;
|
||||
const backoffReason = retryable ? "provider_retry_or_rate_limit" : noProgress ? "projection_no_progress" : null;
|
||||
const backoffMs = backoffReason ? agentRunProjectionBackoffMs(noProgressPollCount, env) : 0;
|
||||
return {
|
||||
pollCount: Number.isInteger(Number(pollCount)) && Number(pollCount) > 0 ? Math.floor(Number(pollCount)) : agentRunProjectionPollCount(projectionState),
|
||||
noProgressPollCount,
|
||||
backoffMs,
|
||||
backoffReason,
|
||||
rootCause: boundedAgentRunRootCause(rootCause ?? agentRunPollingRootCause(error) ?? backoffReason ?? "projection_sync"),
|
||||
terminalConsistencyGap: nonNegativeNumber(terminalConsistencyGap),
|
||||
nextRetryAt: backoffMs > 0 ? new Date(Date.now() + backoffMs).toISOString() : null,
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
|
||||
export function agentRunPollingRetryableError(error = null) {
|
||||
const statusCode = Number(error?.statusCode ?? error?.agentRunError?.statusCode ?? 0);
|
||||
if ([408, 425, 429, 500, 502, 503, 504].includes(statusCode)) return true;
|
||||
const code = String(error?.code ?? error?.agentRunError?.failureKind ?? "").toLowerCase();
|
||||
return error?.retryable === true || code === "agentrun_timeout" || /rate.?limit|retry|timeout|temporar|unavailable|overload/u.test(code);
|
||||
}
|
||||
|
||||
export function agentRunPollingRootCause(error = null) {
|
||||
if (!error) return null;
|
||||
const statusCode = Number(error?.statusCode ?? error?.agentRunError?.statusCode ?? 0);
|
||||
if (statusCode === 429) return "provider_rate_limited";
|
||||
if ([408, 425, 500, 502, 503, 504].includes(statusCode)) return `provider_retry_http_${statusCode}`;
|
||||
return error?.code ?? error?.agentRunError?.failureKind ?? error?.name ?? "agentrun_sync_failed";
|
||||
}
|
||||
|
||||
export function agentRunErrorWithPolling(error, polling = null) {
|
||||
if (!error || !polling) return error;
|
||||
error.polling = polling;
|
||||
error.backoffMs = polling.backoffMs;
|
||||
error.nextRetryAt = polling.nextRetryAt;
|
||||
error.rootCause = polling.rootCause;
|
||||
return error;
|
||||
}
|
||||
|
||||
export function agentRunTerminalConsistencyGap({ mapped = {}, eventsResponse = null, result = null } = {}) {
|
||||
const agentRun = mapped?.agentRun && typeof mapped.agentRun === "object" ? mapped.agentRun : {};
|
||||
const terminal = agentRunPollingTerminalStatus(result?.terminalStatus ?? result?.commandState ?? result?.status ?? mapped?.status ?? agentRun.terminalStatus ?? agentRun.commandState) || Boolean(agentRunTerminalStatusFromEvents(eventsResponse?.events));
|
||||
if (!terminal) return 0;
|
||||
const latest = Math.max(
|
||||
nonNegativeNumber(eventsResponse?.maxSeq),
|
||||
nonNegativeNumber(result?.eventCount),
|
||||
nonNegativeNumber(result?.lastSeq),
|
||||
nonNegativeNumber(agentRun.lastSeq)
|
||||
);
|
||||
const observed = Math.max(
|
||||
nonNegativeNumber(eventsResponse?.traceLastSeq),
|
||||
nonNegativeNumber(result?.lastSeq),
|
||||
nonNegativeNumber(agentRun.lastSeq)
|
||||
);
|
||||
return Math.max(0, latest - observed);
|
||||
}
|
||||
|
||||
export function agentRunTerminalResultRefreshSatisfied(mapped = {}, traceId = null) {
|
||||
const safeId = safeTraceId(traceId ?? mapped?.traceId);
|
||||
const agentRun = mapped?.agentRun && typeof mapped.agentRun === "object" ? mapped.agentRun : null;
|
||||
const traceSummary = mapped?.traceSummary && typeof mapped.traceSummary === "object" ? mapped.traceSummary : null;
|
||||
const finalResponse = mapped?.finalResponse && typeof mapped.finalResponse === "object" ? mapped.finalResponse : null;
|
||||
const providerTrace = agentRun?.providerTrace && typeof agentRun.providerTrace === "object"
|
||||
? agentRun.providerTrace
|
||||
: mapped?.providerTrace && typeof mapped.providerTrace === "object"
|
||||
? mapped.providerTrace
|
||||
: null;
|
||||
const commandId = text(agentRun?.commandId);
|
||||
const runId = text(agentRun?.runId);
|
||||
if (!safeId || !runId || !commandId) return false;
|
||||
if (!agentRunPollingTerminalStatus(mapped?.status ?? agentRun?.terminalStatus ?? agentRun?.commandState ?? agentRun?.status)) return false;
|
||||
if (safeTraceId(mapped?.traceId ?? agentRun?.traceId) !== safeId) return false;
|
||||
if (safeTraceId(providerTrace?.traceId) !== safeId) return false;
|
||||
const finalTraceId = safeTraceId(finalResponse?.traceId ?? safeId);
|
||||
if (finalTraceId && finalTraceId !== safeId) return false;
|
||||
if (!String(finalResponse?.text ?? "").trim()) return false;
|
||||
if (traceSummary?.source !== "agentrun-command-result") return false;
|
||||
if (safeTraceId(traceSummary.traceId ?? safeId) !== safeId) return false;
|
||||
const summaryAgentRun = traceSummary.agentRun && typeof traceSummary.agentRun === "object" ? traceSummary.agentRun : null;
|
||||
if (text(summaryAgentRun?.runId) !== runId) return false;
|
||||
if (text(summaryAgentRun?.commandId) !== commandId) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
export function emitAgentRunProjectionPollingSpan({ traceId, env = process.env, agentRun = {}, polling = {}, terminalRefreshSkipped = false } = {}) {
|
||||
void emitCodeAgentOtelSpan("projection_sync", traceId, env, {
|
||||
attributes: {
|
||||
runId: agentRun?.runId ?? null,
|
||||
commandId: agentRun?.commandId ?? null,
|
||||
terminalRefreshSkipped: terminalRefreshSkipped === true,
|
||||
pollCount: nonNegativeNumber(polling.pollCount),
|
||||
backoffMs: nonNegativeNumber(polling.backoffMs),
|
||||
rootCause: boundedAgentRunRootCause(polling.rootCause),
|
||||
terminalConsistencyGap: nonNegativeNumber(polling.terminalConsistencyGap)
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export async function fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, env, mapping = {} }) {
|
||||
const runId = requiredString(mapping.runId, "runId");
|
||||
const currentCommandId = typeof mapping.commandId === "string" ? mapping.commandId : "";
|
||||
const replayWindow = agentRunTraceReplayWindow(mapping);
|
||||
const afterSeqOverride = Number(mapping.afterSeqOverride);
|
||||
const afterSeq = Number.isFinite(afterSeqOverride) && afterSeqOverride >= 0 ? Math.floor(afterSeqOverride) : replayWindow.afterSeq;
|
||||
const endSeq = Number.isFinite(afterSeqOverride) && afterSeqOverride >= 0 ? 0 : replayWindow.endSeq;
|
||||
const limit = parsePositiveInteger(mapping.eventsPageLimit ?? env?.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT, 500);
|
||||
const path = `/api/v1/runs/${encodeURIComponent(runId)}/events?afterSeq=${encodeURIComponent(String(afterSeq))}&limit=${encodeURIComponent(String(limit))}`;
|
||||
const response = await agentRunJson(fetchImpl, managerUrl, path, { method: "GET", timeoutMs, env });
|
||||
const rawEvents = Array.isArray(response?.items) ? response.items : [];
|
||||
const events = rawEvents.filter((event) => agentRunEventBelongsToTrace(event, { currentCommandId, afterSeq, endSeq }));
|
||||
const maxSeq = Math.max(afterSeq, ...rawEvents.map((event) => Number(event?.seq ?? 0)).filter(Number.isFinite));
|
||||
const traceLastSeq = Math.max(afterSeq, ...events.map((event) => Number(event?.seq ?? 0)).filter(Number.isFinite));
|
||||
void emitCodeAgentOtelSpan("projection_sync", mapping.traceId ?? mapping.hwlabTraceId ?? mapping.businessTraceId, env, {
|
||||
attributes: {
|
||||
runId,
|
||||
commandId: currentCommandId || null,
|
||||
afterSeq,
|
||||
endSeq,
|
||||
limit,
|
||||
rawEventCount: rawEvents.length,
|
||||
eventCount: events.length,
|
||||
commandFiltered: Boolean(currentCommandId),
|
||||
maxSeq,
|
||||
traceLastSeq,
|
||||
terminalFromRawEvents: agentRunTerminalStatusFromEvents(rawEvents),
|
||||
terminalFromEvents: agentRunTerminalStatusFromEvents(events),
|
||||
pollCount: nonNegativeNumber(mapping.pollCount),
|
||||
backoffMs: nonNegativeNumber(mapping.backoffMs),
|
||||
rootCause: boundedAgentRunRootCause(mapping.rootCause),
|
||||
terminalConsistencyGap: nonNegativeNumber(mapping.terminalConsistencyGap)
|
||||
}
|
||||
});
|
||||
return {
|
||||
events,
|
||||
afterSeq,
|
||||
endSeq,
|
||||
rawEventCount: rawEvents.length,
|
||||
commandFiltered: Boolean(currentCommandId),
|
||||
maxSeq,
|
||||
traceLastSeq
|
||||
};
|
||||
}
|
||||
|
||||
export async function measuredAgentRunEventsFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapping = {} }) {
|
||||
const startedAt = nowMs();
|
||||
try {
|
||||
const response = await fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, env, mapping });
|
||||
performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "agentrun_events_fetch", status: "ok", durationMs: nowMs() - startedAt });
|
||||
return response;
|
||||
} catch (error) {
|
||||
performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "agentrun_events_fetch", status: agentRunMetricStatusForError(error), durationMs: nowMs() - startedAt });
|
||||
performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "agentrun_events_fetch_failed" });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export async function measuredAgentRunResultFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapped = {}, eventsResponse = null }) {
|
||||
const startedAt = nowMs();
|
||||
const path = `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/commands/${encodeURIComponent(mapped.agentRun.commandId)}/result`;
|
||||
try {
|
||||
const result = await agentRunJson(fetchImpl, managerUrl, path, { method: "GET", timeoutMs, env });
|
||||
const eventCount = agentRunResultEventCount(result, eventsResponse);
|
||||
performanceStore?.recordAgentRunResult?.({
|
||||
durationMs: nowMs() - startedAt,
|
||||
eventCount,
|
||||
eventsScanned: eventCount,
|
||||
pagesScanned: estimateAgentRunResultPages(eventCount),
|
||||
status: result?.terminalStatus ?? result?.status ?? "ok",
|
||||
budget: timeoutMs
|
||||
});
|
||||
return result;
|
||||
} catch (error) {
|
||||
const status = agentRunMetricStatusForError(error);
|
||||
const eventCount = agentRunResultEventCount(null, eventsResponse);
|
||||
performanceStore?.recordAgentRunResult?.({
|
||||
durationMs: nowMs() - startedAt,
|
||||
eventCount,
|
||||
eventsScanned: eventCount,
|
||||
pagesScanned: estimateAgentRunResultPages(eventCount),
|
||||
status,
|
||||
budget: timeoutMs,
|
||||
timedOut: status === "timeout"
|
||||
});
|
||||
performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "agentrun_result_failed" });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export function recordAgentRunEventsProjectionMetrics(performanceStore, eventsResponse = null, previousLastSeq = 0, nextLastSeq = 0) {
|
||||
if (!performanceStore || !eventsResponse) return;
|
||||
const cursorAdvance = Math.max(0, Number(nextLastSeq ?? 0) - Number(previousLastSeq ?? 0));
|
||||
if (cursorAdvance > 0) {
|
||||
performanceStore.recordWorkbenchProjectionCursorAdvance?.({
|
||||
status: "advanced",
|
||||
reason: eventsResponse.commandFiltered ? "command_filtered_events" : "run_events",
|
||||
projectionStatus: "projecting",
|
||||
source: "agentrun_v01",
|
||||
count: cursorAdvance
|
||||
});
|
||||
}
|
||||
const counts = new Map();
|
||||
for (const event of eventsResponse.events ?? []) {
|
||||
const kind = String(event?.type ?? event?.payload?.phase ?? "unknown").trim() || "unknown";
|
||||
counts.set(kind, (counts.get(kind) ?? 0) + 1);
|
||||
}
|
||||
if (counts.size === 0 && Number(eventsResponse.rawEventCount ?? 0) > 0) counts.set("filtered", Number(eventsResponse.rawEventCount));
|
||||
for (const [eventKind, count] of counts.entries()) {
|
||||
performanceStore.recordWorkbenchProjectorEvents?.({ eventKind, status: "ok", count });
|
||||
}
|
||||
}
|
||||
|
||||
export function agentRunMetricStatusForError(error) {
|
||||
return error?.code === "agentrun_timeout" || error?.name === "AbortError" || Number(error?.statusCode) === 504 ? "timeout" : "error";
|
||||
}
|
||||
|
||||
export 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));
|
||||
}
|
||||
|
||||
export function agentRunTerminalStatusFromEvents(events = []) {
|
||||
if (!Array.isArray(events)) return null;
|
||||
for (const event of [...events].reverse()) {
|
||||
const payload = event?.payload && typeof event.payload === "object" ? event.payload : {};
|
||||
const phase = payload.phase ?? event?.phase;
|
||||
const candidates = [
|
||||
event?.type === "terminal_status" ? payload.terminalStatus ?? event?.terminalStatus ?? payload.status ?? event?.status : null,
|
||||
phase === "command-terminal" ? payload.terminalStatus ?? event?.terminalStatus ?? payload.status ?? event?.status : null,
|
||||
phase === "turn:completed" || phase === "turn/steer:completed" ? "completed" : null,
|
||||
event?.terminal === true ? payload.terminalStatus ?? event?.terminalStatus ?? payload.status ?? event?.status ?? "completed" : null,
|
||||
payload.terminal === true ? payload.terminalStatus ?? event?.terminalStatus ?? payload.status ?? event?.status ?? "completed" : null
|
||||
];
|
||||
for (const value of candidates) {
|
||||
const status = normalizeAgentRunStatus(value);
|
||||
if (agentRunPollingTerminalStatus(status)) return status;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export function agentRunEventCommandId(event) {
|
||||
const payload = event?.payload && typeof event.payload === "object" ? event.payload : {};
|
||||
if (typeof event?.commandId === "string") return event.commandId;
|
||||
return typeof payload.commandId === "string" ? payload.commandId : "";
|
||||
}
|
||||
|
||||
export function nonNegativeNumber(value) {
|
||||
const number = Number(value ?? 0);
|
||||
return Number.isFinite(number) && number > 0 ? Math.floor(number) : 0;
|
||||
}
|
||||
|
||||
function agentRunProjectionBackoffMs(attempt, env = process.env) {
|
||||
const initialMs = parsePositiveInteger(env?.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_SYNC_BACKOFF_INITIAL_MS, DEFAULT_AGENTRUN_PROJECTION_SYNC_BACKOFF_INITIAL_MS);
|
||||
const maxMs = parsePositiveInteger(env?.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_SYNC_BACKOFF_MAX_MS, DEFAULT_AGENTRUN_PROJECTION_SYNC_BACKOFF_MAX_MS);
|
||||
const exponent = Math.max(0, Math.min(10, nonNegativeNumber(attempt) - 1));
|
||||
return Math.min(maxMs, initialMs * 2 ** exponent);
|
||||
}
|
||||
|
||||
function boundedAgentRunRootCause(value) {
|
||||
const raw = String(value ?? "").trim().replace(/[^a-z0-9_.:-]+/giu, "_").replace(/^_+|_+$/gu, "").toLowerCase();
|
||||
return raw ? raw.slice(0, 120) : null;
|
||||
}
|
||||
|
||||
function agentRunResultEventCount(result = null, eventsResponse = null) {
|
||||
const candidates = [
|
||||
result?.traceSummary?.sourceEventCount,
|
||||
result?.traceSummary?.eventCount,
|
||||
result?.sourceEventCount,
|
||||
result?.eventCount,
|
||||
result?.providerTrace?.eventCount,
|
||||
eventsResponse?.maxSeq,
|
||||
eventsResponse?.rawEventCount,
|
||||
eventsResponse?.events?.length
|
||||
];
|
||||
for (const value of candidates) {
|
||||
const count = Number(value);
|
||||
if (Number.isFinite(count) && count >= 0) return Math.floor(count);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
function estimateAgentRunResultPages(eventCount) {
|
||||
const count = Number(eventCount);
|
||||
if (!Number.isFinite(count) || count <= 0) return 0;
|
||||
return Math.max(1, Math.ceil(count / 500));
|
||||
}
|
||||
|
||||
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 agentRunPollingTerminalStatus(value) {
|
||||
return AGENTRUN_TERMINAL_STATUSES.has(normalizeAgentRunStatus(value));
|
||||
}
|
||||
|
||||
function normalizeAgentRunStatus(value) {
|
||||
const status = String(value ?? "").trim().toLowerCase().replace(/_/gu, "-");
|
||||
return status === "cancelled" ? "canceled" : status;
|
||||
}
|
||||
|
||||
function requiredString(value, fieldName) {
|
||||
if (typeof value === "string" && value.trim()) return value.trim();
|
||||
throw Object.assign(new Error(`AgentRun response missing ${fieldName}`), { code: "agentrun_response_invalid", statusCode: 502 });
|
||||
}
|
||||
|
||||
function nowMs() {
|
||||
if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now();
|
||||
return Date.now();
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
|
||||
// Responsibility: Focused AgentRun projection polling/backoff regressions for Cloud API chat result sync.
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { createServer as createHttpServer } from "node:http";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import { syncAgentRunChatResult } from "./code-agent-agentrun-adapter.ts";
|
||||
import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts";
|
||||
|
||||
test("AgentRun sync stops upstream polling when terminal command result is already sealed (#1445)", async () => {
|
||||
const traceId = "trc_1445_sealed_terminal";
|
||||
const runId = "run_1445_sealed_terminal";
|
||||
const commandId = "cmd_1445_sealed_terminal";
|
||||
const traceStore = createCodeAgentTraceStore();
|
||||
let fetched = false;
|
||||
const synced = await syncAgentRunChatResult({
|
||||
traceId,
|
||||
currentResult: {
|
||||
ok: true,
|
||||
status: "completed",
|
||||
traceId,
|
||||
finalResponse: { text: "sealed final response", traceId },
|
||||
traceSummary: { source: "agentrun-command-result", traceId, agentRun: { runId, commandId, lastSeq: 17 } },
|
||||
agentRun: {
|
||||
adapter: "agentrun-v01",
|
||||
runId,
|
||||
commandId,
|
||||
traceId,
|
||||
status: "completed",
|
||||
runStatus: "completed",
|
||||
commandState: "completed",
|
||||
terminalStatus: "completed",
|
||||
providerTrace: { traceId, runId, commandId, valuesPrinted: false },
|
||||
lastSeq: 17,
|
||||
managerUrl: "http://127.0.0.1:1",
|
||||
valuesPrinted: false
|
||||
},
|
||||
valuesPrinted: false
|
||||
},
|
||||
options: {
|
||||
fetchImpl: async () => {
|
||||
fetched = true;
|
||||
throw new Error("sealed terminal result should not fetch AgentRun");
|
||||
},
|
||||
env: { HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1" }
|
||||
},
|
||||
traceStore
|
||||
});
|
||||
|
||||
assert.equal(fetched, false);
|
||||
assert.equal(synced.terminalRefreshSkipped, true);
|
||||
assert.equal(synced.resultSynced, false);
|
||||
assert.equal(synced.polling.rootCause, "terminal_result_sealed");
|
||||
assert.equal(synced.polling.backoffMs, 0);
|
||||
assert.equal(synced.polling.terminalConsistencyGap, 0);
|
||||
});
|
||||
|
||||
test("AgentRun events rate limit writes durable projection backoff (#1445)", async () => {
|
||||
const traceId = "trc_1445_events_rate_limit";
|
||||
const runId = "run_1445_events_rate_limit";
|
||||
const commandId = "cmd_1445_events_rate_limit";
|
||||
const calls = [];
|
||||
const writes = [];
|
||||
const runtimeStore = {
|
||||
async getWorkbenchProjectionState() {
|
||||
return { projectionState: { traceId, sourceRunId: runId, sourceCommandId: commandId, lastSourceSeq: 40, pollCount: 2, noProgressPollCount: 1, resultSyncState: "not_started", createdAt: "2026-07-02T00:00:00.000Z" } };
|
||||
},
|
||||
async writeWorkbenchProjectionState(input = {}) {
|
||||
writes.push(input.projectionState);
|
||||
return { written: true, projectionState: input.projectionState };
|
||||
}
|
||||
};
|
||||
const agentRunServer = createHttpServer(async (request, response) => {
|
||||
const url = new URL(request.url || "/", "http://127.0.0.1");
|
||||
calls.push(`${request.method} ${url.pathname}${url.search}`);
|
||||
if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/events`) {
|
||||
response.writeHead(429, { "content-type": "application/json" });
|
||||
response.end(`${JSON.stringify({ ok: false, failureKind: "provider_rate_limited", message: "provider rate limited; retry later" })}\n`);
|
||||
return;
|
||||
}
|
||||
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;
|
||||
const traceStore = createCodeAgentTraceStore();
|
||||
try {
|
||||
await assert.rejects(
|
||||
syncAgentRunChatResult({
|
||||
traceId,
|
||||
currentResult: {
|
||||
status: "running",
|
||||
traceId,
|
||||
agentRun: { adapter: "agentrun-v01", runId, commandId, traceId, providerTrace: { traceId }, lastSeq: 40, managerUrl: `http://127.0.0.1:${agentRunPort}`, valuesPrinted: false },
|
||||
valuesPrinted: false
|
||||
},
|
||||
options: {
|
||||
runtimeStore,
|
||||
env: {
|
||||
HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1",
|
||||
HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_SYNC_BACKOFF_INITIAL_MS: "1000",
|
||||
HWLAB_CODE_AGENT_AGENTRUN_HTTP_TIMEOUT_MS: "2500"
|
||||
}
|
||||
},
|
||||
traceStore
|
||||
}),
|
||||
(error) => {
|
||||
assert.equal(error.statusCode, 429);
|
||||
assert.equal(error.polling.rootCause, "provider_rate_limited");
|
||||
assert.equal(error.polling.backoffReason, "provider_retry_or_rate_limit");
|
||||
assert.equal(error.polling.backoffMs, 2000);
|
||||
assert.ok(error.polling.nextRetryAt);
|
||||
return true;
|
||||
}
|
||||
);
|
||||
|
||||
assert.equal(calls.length, 1);
|
||||
assert.match(calls[0], /afterSeq=40/u);
|
||||
assert.equal(writes.length, 1);
|
||||
assert.equal(writes[0].projectionStatus, "degraded");
|
||||
assert.equal(writes[0].projectionHealth, "degraded");
|
||||
assert.equal(writes[0].backoffMs, 2000);
|
||||
assert.equal(writes[0].backoffReason, "provider_retry_or_rate_limit");
|
||||
assert.equal(writes[0].rootCause, "provider_rate_limited");
|
||||
assert.ok(writes[0].nextRetryAt);
|
||||
} finally {
|
||||
await new Promise((resolve, reject) => agentRunServer.close((error) => (error ? reject(error) : resolve())));
|
||||
}
|
||||
});
|
||||
@@ -31,6 +31,7 @@ export function buildWorkbenchProjectionStateUpdate({
|
||||
resultSyncState = null,
|
||||
projectionStatus = null,
|
||||
projectionHealth = null,
|
||||
polling = null,
|
||||
retryParams = null,
|
||||
error = null,
|
||||
now = () => new Date().toISOString()
|
||||
@@ -56,9 +57,11 @@ export function buildWorkbenchProjectionStateUpdate({
|
||||
lastSourceSeq
|
||||
);
|
||||
const nextRetryParams = normalizeProjectionRetryParams(retryParams) ?? normalizeProjectionRetryParams(previous.retryParams);
|
||||
const nextPolling = normalizeProjectionPolling(polling);
|
||||
const nextResultSyncState = textValue(resultSyncState ?? previous.resultSyncState) || "not_started";
|
||||
const nextProjectionStatus = textValue(projectionStatus) || inferProjectionStatus({ lastSourceSeq, sourceLatestSeq, resultSyncState: nextResultSyncState, previous });
|
||||
const nextProjectionHealth = textValue(projectionHealth) || (error ? "degraded" : "healthy");
|
||||
const nextRetryAt = nextPolling?.nextRetryAt ?? (error ? previous.nextRetryAt ?? null : null);
|
||||
return {
|
||||
...previous,
|
||||
traceId,
|
||||
@@ -87,7 +90,13 @@ export function buildWorkbenchProjectionStateUpdate({
|
||||
lastErrorCode: error ? textValue(error.code ?? error.errorCode ?? "projection_sync_failed") : null,
|
||||
lastErrorMessage: error ? textValue(error.message ?? "Projection sync failed.") : null,
|
||||
failureCount: error ? maxNonNegativeInteger(previous.failureCount) + 1 : 0,
|
||||
nextRetryAt: error ? previous.nextRetryAt ?? null : null,
|
||||
nextRetryAt,
|
||||
pollCount: nextPolling?.pollCount ?? maxNonNegativeInteger(previous.pollCount),
|
||||
noProgressPollCount: nextPolling?.noProgressPollCount ?? (error ? maxNonNegativeInteger(previous.noProgressPollCount) : 0),
|
||||
backoffMs: nextPolling?.backoffMs ?? 0,
|
||||
backoffReason: nextPolling?.backoffReason ?? null,
|
||||
rootCause: nextPolling?.rootCause ?? null,
|
||||
terminalConsistencyGap: nextPolling?.terminalConsistencyGap ?? null,
|
||||
retryParams: nextRetryParams,
|
||||
createdAt: previous.createdAt ?? timestamp,
|
||||
updatedAt: timestamp,
|
||||
@@ -137,6 +146,20 @@ function normalizeProjectionRetryParams(value) {
|
||||
return result;
|
||||
}
|
||||
|
||||
function normalizeProjectionPolling(value) {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) return null;
|
||||
const result = {
|
||||
pollCount: maxNonNegativeInteger(value.pollCount),
|
||||
noProgressPollCount: maxNonNegativeInteger(value.noProgressPollCount),
|
||||
backoffMs: maxNonNegativeInteger(value.backoffMs),
|
||||
backoffReason: clippedText(value.backoffReason, 80),
|
||||
rootCause: clippedText(value.rootCause, 120),
|
||||
terminalConsistencyGap: maxNonNegativeInteger(value.terminalConsistencyGap),
|
||||
nextRetryAt: timestampText(value.nextRetryAt)
|
||||
};
|
||||
return result;
|
||||
}
|
||||
|
||||
function inferProjectionStatus({ lastSourceSeq, sourceLatestSeq, resultSyncState, previous }) {
|
||||
if (["pending", "timed_out", "failed"].includes(resultSyncState)) return "terminal";
|
||||
if (sourceLatestSeq > lastSourceSeq) return "projecting";
|
||||
@@ -167,3 +190,16 @@ function timestampValue(now) {
|
||||
function textValue(value) {
|
||||
return String(value ?? "").trim();
|
||||
}
|
||||
|
||||
function clippedText(value, limit) {
|
||||
const text = textValue(value);
|
||||
if (!text) return null;
|
||||
return text.length > limit ? text.slice(0, limit) : text;
|
||||
}
|
||||
|
||||
function timestampText(value) {
|
||||
const text = textValue(value);
|
||||
if (!text) return null;
|
||||
const parsed = Date.parse(text);
|
||||
return Number.isFinite(parsed) ? new Date(parsed).toISOString() : null;
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@ export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult =
|
||||
void (async () => {
|
||||
let result = currentResult;
|
||||
let lastError = null;
|
||||
let nextDelayMs = pollIntervalMs;
|
||||
while (true) {
|
||||
const cached = getCachedResult?.(traceId);
|
||||
if (isCanceled?.(cached)) return;
|
||||
@@ -28,6 +29,7 @@ export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult =
|
||||
const synced = await syncResult({ result });
|
||||
performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "candidate_scan", status: "ok", durationMs: Date.now() - syncStartedAt });
|
||||
result = synced?.result ?? result;
|
||||
nextDelayMs = projectionFinalizerDelayMs(synced?.polling, pollIntervalMs);
|
||||
const runnerTrace = synced?.runnerTrace ?? traceStore?.snapshot?.(traceId);
|
||||
if (result && isTerminal?.(result, runnerTrace)) {
|
||||
performanceStore?.recordWorkbenchProjectorCandidate?.({ status: "completed", reason: "synced_terminal" });
|
||||
@@ -44,6 +46,7 @@ export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult =
|
||||
} catch (error) {
|
||||
performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "candidate_scan", status: error?.code === "agentrun_timeout" ? "timeout" : "error", durationMs: Date.now() - syncStartedAt });
|
||||
performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "finalizer_sync_failed" });
|
||||
nextDelayMs = projectionFinalizerDelayMs(error?.polling, pollIntervalMs);
|
||||
lastError = error;
|
||||
}
|
||||
const idleMs = Date.now() - lastActivityAt;
|
||||
@@ -67,7 +70,7 @@ export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult =
|
||||
});
|
||||
idleDegraded = true;
|
||||
}
|
||||
await sleep(pollIntervalMs);
|
||||
await sleep(nextDelayMs);
|
||||
}
|
||||
})().catch((error) => {
|
||||
performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "workbench_projection_finalizer_crashed" });
|
||||
@@ -90,3 +93,14 @@ export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult =
|
||||
function defaultSleep(ms) {
|
||||
return new Promise((resolve) => setTimeout(resolve, Math.max(0, Number(ms) || 0)));
|
||||
}
|
||||
|
||||
function projectionFinalizerDelayMs(polling = null, fallbackMs = 1000) {
|
||||
const fallback = positiveInteger(fallbackMs, 1000);
|
||||
const backoffMs = positiveInteger(polling?.backoffMs, 0);
|
||||
return backoffMs > 0 ? Math.max(fallback, backoffMs) : fallback;
|
||||
}
|
||||
|
||||
function positiveInteger(value, fallback) {
|
||||
const parsed = Number.parseInt(String(value ?? ""), 10);
|
||||
return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user