refactor: split AgentRun projection polling
This commit is contained in:
UniDesk Codex
@@ -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,
|
||||
@@ -56,8 +75,6 @@ const DEFAULT_TENANT_ID = "hwlab";
|
||||
const DEFAULT_PROJECT_ID = "pikasTech/HWLAB";
|
||||
const DEFAULT_RUNNER_NAMESPACE = "agentrun-v01";
|
||||
const DEFAULT_TIMEOUT_MS = 1_200_000;
|
||||
const DEFAULT_AGENTRUN_PROJECTION_SYNC_BACKOFF_INITIAL_MS = 1_000;
|
||||
const DEFAULT_AGENTRUN_PROJECTION_SYNC_BACKOFF_MAX_MS = 30_000;
|
||||
const AGENTRUN_RUNNER_KIND = "agentrun-v01-shared-runner";
|
||||
const AGENTRUN_SESSION_MODE = "agentrun-v01-durable-session";
|
||||
const AGENTRUN_IMPLEMENTATION_TYPE = "agentrun-v01-shared-execution-infra";
|
||||
@@ -918,102 +935,6 @@ async function writeWorkbenchProjectionStateForAgentRun(runtimeStore, params = {
|
||||
return await runtimeStore.writeWorkbenchProjectionState({ projectionState });
|
||||
}
|
||||
|
||||
function agentRunProjectionPollCount(projectionState = null) {
|
||||
return nonNegativeNumber(projectionState?.pollCount) + 1;
|
||||
}
|
||||
|
||||
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
|
||||
};
|
||||
}
|
||||
|
||||
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 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);
|
||||
}
|
||||
|
||||
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";
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
function agentRunTerminalConsistencyGap({ mapped = {}, eventsResponse = null, result = null } = {}) {
|
||||
const agentRun = mapped?.agentRun && typeof mapped.agentRun === "object" ? mapped.agentRun : {};
|
||||
const terminal = isAgentRunTerminalStatus(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);
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function boundedAgentRunRootCause(value) {
|
||||
const raw = String(value ?? "").trim().replace(/[^a-z0-9_.:-]+/giu, "_").replace(/^_+|_+$/gu, "").toLowerCase();
|
||||
return raw ? raw.slice(0, 120) : null;
|
||||
}
|
||||
|
||||
function nonNegativeNumber(value) {
|
||||
const number = Number(value ?? 0);
|
||||
return Number.isFinite(number) && number > 0 ? Math.floor(number) : 0;
|
||||
}
|
||||
|
||||
function agentRunCommandResultSyncRequired(mapped = {}, eventsResponse = null) {
|
||||
const agentRun = mapped?.agentRun && typeof mapped.agentRun === "object" ? mapped.agentRun : {};
|
||||
if (agentRunTerminalResultRefreshSatisfied(mapped, mapped?.traceId)) return true;
|
||||
@@ -1021,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);
|
||||
@@ -2534,184 +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),
|
||||
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
|
||||
};
|
||||
}
|
||||
|
||||
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())));
|
||||
}
|
||||
});
|
||||
@@ -1111,127 +1111,6 @@ test("AgentRun sync converts terminal command result even when run remains claim
|
||||
}
|
||||
});
|
||||
|
||||
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())));
|
||||
}
|
||||
});
|
||||
|
||||
test("AgentRun sync seals completed final response from authoritative terminal assistant trace event (#1629)", async () => {
|
||||
const calls = [];
|
||||
const traceId = "trc_issue1629_terminal_assistant_final";
|
||||
|
||||
Reference in New Issue
Block a user