diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index a7849fc5..e50f0188 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -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; diff --git a/internal/cloud/code-agent-agentrun-projection-polling.ts b/internal/cloud/code-agent-agentrun-projection-polling.ts new file mode 100644 index 00000000..546f82d1 --- /dev/null +++ b/internal/cloud/code-agent-agentrun-projection-polling.ts @@ -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(); +} diff --git a/internal/cloud/server-agent-chat-agentrun-backoff.test.ts b/internal/cloud/server-agent-chat-agentrun-backoff.test.ts new file mode 100644 index 00000000..06ae85eb --- /dev/null +++ b/internal/cloud/server-agent-chat-agentrun-backoff.test.ts @@ -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()))); + } +}); diff --git a/internal/cloud/workbench-projection-cursor.ts b/internal/cloud/workbench-projection-cursor.ts index 28280029..3f286390 100644 --- a/internal/cloud/workbench-projection-cursor.ts +++ b/internal/cloud/workbench-projection-cursor.ts @@ -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; +} diff --git a/internal/cloud/workbench-projection-finalizer.ts b/internal/cloud/workbench-projection-finalizer.ts index 42841a66..c5b72a10 100644 --- a/internal/cloud/workbench-projection-finalizer.ts +++ b/internal/cloud/workbench-projection-finalizer.ts @@ -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; +}