diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index a7849fc5..1935e203 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -56,6 +56,8 @@ 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"; @@ -699,25 +701,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 +754,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 +772,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 +791,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 +874,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 +890,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 = {} } = {}) { @@ -868,6 +918,102 @@ 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; @@ -949,7 +1095,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); @@ -2415,7 +2561,11 @@ async function fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, e maxSeq, traceLastSeq, terminalFromRawEvents: agentRunTerminalStatusFromEvents(rawEvents), - terminalFromEvents: agentRunTerminalStatusFromEvents(events) + terminalFromEvents: agentRunTerminalStatusFromEvents(events), + pollCount: nonNegativeNumber(mapping.pollCount), + backoffMs: nonNegativeNumber(mapping.backoffMs), + rootCause: boundedAgentRunRootCause(mapping.rootCause), + terminalConsistencyGap: nonNegativeNumber(mapping.terminalConsistencyGap) } }); return { diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 0a6e0d8b..187b142c 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/internal/cloud/server-agent-chat.test.ts @@ -1111,6 +1111,127 @@ 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"; 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; +}