diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 8ee8ac0e..1c3d6caf 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -696,6 +696,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op 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 fetchPlan = buildAgentRunProjectionEventsFetchPlan({ projectionState, agentRun: mapped.agentRun, @@ -714,6 +715,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op eventsResponse, resultSyncState: terminalFromEvents ? "pending" : null, projectionStatus: terminalFromEvents ? "terminal" : null, + retryParams: projectionRetryParams, now: options.now }); const deferResultSync = agentRunResultSyncDeferred({ forceResultSync, options, env }); @@ -744,6 +746,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op eventsResponse, resultSyncState: agentRunMetricStatusForError(error) === "timeout" ? "timed_out" : "failed", projectionStatus: terminalFromEvents ? "terminal" : null, + retryParams: projectionRetryParams, error, now: options.now }); @@ -761,7 +764,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op valuesPrinted: false }); const freshSessionRetry = await retryAgentRunTurnAfterFreshSessionFailure({ - params: retryParams, + params: projectionRetryParams, options, traceId, traceStore, @@ -783,6 +786,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op eventsResponse: null, resultSyncState: null, projectionStatus: "projecting", + retryParams: projectionRetryParams, now: options.now }); return { result: freshSessionRetry.payload, runnerTrace: freshSessionRetry.payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: false, freshSessionRetried: true }; @@ -797,6 +801,7 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op eventsResponse, resultSyncState: "synced", projectionStatus: isAgentRunTerminalStatus(payload?.status) ? "terminal" : null, + retryParams: projectionRetryParams, now: options.now }); return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: true }; diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index cc00dcb7..3c921756 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/internal/cloud/server-agent-chat.test.ts @@ -3032,6 +3032,200 @@ test("cloud api resumes AgentRun projection from durable state after restart", a } }); +test("cloud api projection resume retries evicted AgentRun session with stored prompt", async () => { + const calls = []; + const durableEvents = []; + const projectionStates = new Map(); + const traceId = "trc_projection_resume_evicted_retry"; + const runId = "run_projection_resume_evicted"; + const commandId = "cmd_projection_resume_evicted"; + const retryRunId = "run_projection_resume_evicted_retry"; + const retryCommandId = "cmd_projection_resume_evicted_retry"; + const sessionId = "ses_projection_resume_evicted"; + const conversationId = "cnv_projection_resume_evicted"; + const promptText = "resume projection evicted session with original prompt"; + + const agentRunServer = createHttpServer(async (request, response) => { + const url = new URL(request.url || "/", "http://127.0.0.1"); + const chunks = []; + for await (const chunk of request) chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + const body = chunks.length ? JSON.parse(Buffer.concat(chunks).toString("utf8")) : null; + calls.push({ method: request.method, path: url.pathname, search: url.search, afterSeq: url.searchParams.get("afterSeq"), body }); + const send = (data, status = 200) => { + response.writeHead(status, { "content-type": "application/json" }); + response.end(`${JSON.stringify({ ok: status < 400, data })}\n`); + }; + if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/events`) { + return send({ items: [ + { id: "evt_projection_evicted_1", runId, seq: 1, type: "error", payload: { commandId, failureKind: "session-store-evicted", message: "codex app-server thread/resume reported no rollout found for PVC-backed session; session storage was likely evicted" }, createdAt: "2026-06-19T12:01:00.000Z" }, + { id: "evt_projection_evicted_2", runId, seq: 2, type: "terminal_status", payload: { commandId, terminalStatus: "failed", failureKind: "session-store-evicted" }, createdAt: "2026-06-19T12:01:01.000Z" } + ] }); + } + if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${commandId}/result`) { + return send({ + runId, + commandId, + status: "failed", + runStatus: "failed", + commandState: "failed", + terminalStatus: "failed", + failureKind: "session-store-evicted", + failureMessage: "codex app-server thread/resume reported no rollout found for PVC-backed session; session storage was likely evicted", + completed: false, + lastSeq: 2, + eventCount: 2, + sessionRef: { sessionId: "ses_agentrun_projection_resume_evicted", conversationId, threadId: "thread_projection_evicted" } + }); + } + if (request.method === "POST" && url.pathname === "/api/v1/sessions") { + assert.match(body.sessionId, /-reset-/u); + return send({ sessionId: body.sessionId, backendProfile: body.backendProfile }); + } + if (request.method === "POST" && url.pathname === "/api/v1/runs") { + assert.match(body.sessionRef?.sessionId, /-reset-/u); + assert.equal(body.sessionRef?.threadId, undefined); + return send({ id: retryRunId, status: "pending", backendProfile: "dsflash-go", sessionRef: body.sessionRef, resourceBundleRef: body.resourceBundleRef }); + } + if (request.method === "POST" && url.pathname === `/api/v1/runs/${retryRunId}/commands`) { + assert.equal(body.type, "turn"); + assert.equal(body.payload.prompt, promptText); + assert.equal(body.payload.message, promptText); + assert.equal(body.payload.threadId, null); + assert.match(body.payload.sessionId, /-reset-/u); + assert.equal(body.payload.providerProfile, "dsflash-go"); + return send({ id: retryCommandId, runId: retryRunId, state: "queued" }); + } + if (request.method === "POST" && url.pathname === `/api/v1/runs/${retryRunId}/runner-jobs`) { + return send({ + action: "create-kubernetes-job", + runId: retryRunId, + commandId: retryCommandId, + attemptId: "attempt_projection_resume_evicted_retry", + runnerId: "runner_projection_resume_evicted_retry", + namespace: "agentrun-v01", + jobName: "agentrun-v01-projection-resume-evicted-retry", + runner: { attemptId: "attempt_projection_resume_evicted_retry", runnerId: "runner_projection_resume_evicted_retry" } + }); + } + return send({ message: `unexpected ${request.method} ${url.pathname}` }, 500); + }); + await new Promise((resolve) => agentRunServer.listen(0, "127.0.0.1", resolve)); + const agentRunPort = agentRunServer.address().port; + + projectionStates.set(traceId, { + traceId, + sessionId, + conversationId, + threadId: "thread_projection_evicted", + ownerUserId: TEST_AGENT_ACTOR.id, + ownerRole: TEST_AGENT_ACTOR.role, + runId, + commandId, + sourceRunId: runId, + sourceCommandId: commandId, + managerUrl: `http://127.0.0.1:${agentRunPort}`, + backendProfile: "dsflash-go", + providerId: "JD01", + lastSourceSeq: 0, + lastAgentRunSeq: 0, + lastProjectedSeq: 0, + sourceLatestSeq: 0, + upstreamLatestSeq: 0, + projectionStatus: "projecting", + projectionHealth: "healthy", + resultSyncState: "not_started", + retryParams: { + message: promptText, + prompt: promptText, + text: promptText, + sessionId, + conversationId, + threadId: "thread_projection_evicted", + ownerUserId: TEST_AGENT_ACTOR.id, + ownerRole: TEST_AGENT_ACTOR.role, + providerProfile: "dsflash-go", + codeAgentProviderProfile: "dsflash-go", + backendProfile: "dsflash-go", + valuesPrinted: false + }, + createdAt: "2026-06-19T12:00:00.000Z", + updatedAt: "2026-06-19T12:00:00.000Z", + valuesPrinted: false + }); + + const runtimeStore = { + async queryAgentTraceEvents(params = {}) { + return { events: durableEvents.filter((event) => event.traceId === params.traceId), count: durableEvents.length }; + }, + async writeAgentTraceEvent(params = {}) { + const event = params.event ?? params.traceEvent ?? params; + durableEvents.push(event); + return { written: true, traceEvent: event }; + }, + async getWorkbenchProjectionState(params = {}) { + const projectionState = projectionStates.get(params.traceId) ?? null; + return { projectionState, found: Boolean(projectionState) }; + }, + async queryWorkbenchProjectionStates() { + return { states: [...projectionStates.values()], count: projectionStates.size }; + }, + async writeWorkbenchProjectionState(params = {}) { + const projectionState = params.projectionState ?? params.state ?? params; + projectionStates.set(projectionState.traceId, projectionState); + return { written: true, projectionState }; + } + }; + const accessController = { + required: false, + store: { + async listAgentSessionsForUser() { return []; }, + async getAgentSessionByTraceId() { return null; } + }, + async authenticate() { return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION }; } + }; + + const server = createCloudApiServer({ + accessController, + runtimeStore, + codeAgentChatResults: new Map(), + env: { + HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01", + HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1", + HWLAB_CODE_AGENT_AGENTRUN_SOURCE_COMMIT: "a".repeat(40), + HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "JD01", + HWLAB_RUNTIME_NAMESPACE: "hwlab-v03", + HWLAB_CODE_AGENT_AGENTRUN_BACKEND_PROFILE: "dsflash-go", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_ENABLED: "1", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS: "1", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS: "0", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS: "0", + HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE: "10" + } + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + for (let index = 0; index < 100 && !calls.some((call) => call.path === `/api/v1/runs/${retryRunId}/commands`); index += 1) await delay(10); + assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${commandId}/result`), true); + assert.equal(calls.some((call) => call.path === `/api/v1/runs/${retryRunId}/commands`), true); + const state = projectionStates.get(traceId); + assert.equal(state.runId, retryRunId); + assert.equal(state.commandId, retryCommandId); + assert.equal(state.sourceRunId, retryRunId); + assert.equal(state.sourceCommandId, retryCommandId); + assert.equal(state.projectionStatus, "projecting"); + assert.equal(state.retryParams.message, promptText); + assert.equal(state.retryParams.providerProfile, "dsflash-go"); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + await new Promise((resolve, reject) => { + agentRunServer.close((error) => (error ? reject(error) : resolve())); + }); + } +}); + test("cloud api trace skips AgentRun refresh when terminal snapshot is already complete (#1422)", async () => { const calls = []; const traceId = "trc_issue1422_trace_terminal_snapshot"; diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 536fc67f..412a3b9a 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -1683,9 +1683,11 @@ function agentRunProjectionResumeCandidate(session = {}, record = {}, { minAgeMs const threadId = safeOpaqueId(record.threadId ?? session.threadId) || null; const ownerUserId = textValue(record.ownerUserId ?? session.ownerUserId); const ownerRole = textValue(record.ownerRole ?? session.ownerRole) || "user"; + const retryParams = agentRunProjectionRetryParams(record.retryParams ?? record.projectionRetryParams ?? session.retryParams); return { traceId, params: { + ...retryParams, traceId, sessionId, conversationId, @@ -1731,9 +1733,11 @@ function agentRunProjectionResumeCandidateFromState(state = {}, { minAgeMs = DEF const ownerUserId = textValue(state.ownerUserId); const ownerRole = textValue(state.ownerRole) || "user"; const lastSeq = Number.isFinite(Number(state.lastSourceSeq ?? state.lastAgentRunSeq)) ? Number(state.lastSourceSeq ?? state.lastAgentRunSeq) : 0; + const retryParams = agentRunProjectionRetryParams(state.retryParams ?? state.projectionRetryParams); return { traceId, params: { + ...retryParams, traceId, sessionId, conversationId, @@ -1772,6 +1776,29 @@ function agentRunProjectionResumeCandidateFromState(state = {}, { minAgeMs = DEF }; } +function agentRunProjectionRetryParams(value = null) { + if (!value || typeof value !== "object" || Array.isArray(value)) return {}; + const message = textValue(value.message ?? value.prompt ?? value.text); + if (!message) return {}; + const result = { + message, + prompt: message, + text: message, + projectId: textValue(value.projectId) || undefined, + ownerUserId: textValue(value.ownerUserId) || undefined, + ownerRole: textValue(value.ownerRole) || undefined, + providerProfile: textValue(value.providerProfile ?? value.codeAgentProviderProfile ?? value.backendProfile) || undefined, + codeAgentProviderProfile: textValue(value.codeAgentProviderProfile ?? value.providerProfile ?? value.backendProfile) || undefined, + backendProfile: textValue(value.backendProfile ?? value.providerProfile ?? value.codeAgentProviderProfile) || undefined, + providerId: textValue(value.providerId) || undefined, + valuesPrinted: false + }; + for (const [key, item] of Object.entries(result)) { + if (item === undefined || item === "") delete result[key]; + } + return result; +} + async function resumeAgentRunProjectionCandidate({ candidate, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) { const { currentResult } = await agentRunProjectionResumeResultWithCursor(candidate, options); const resumeOptions = { ...options, deferAgentRunResultSync: false }; @@ -1781,7 +1808,8 @@ async function resumeAgentRunProjectionCandidate({ candidate, options = {}, trac options: resumeOptions, traceStore, forceResultSync: true, - refreshEvents: true + refreshEvents: true, + retryParams: candidate.params }); const payload = codeAgentPayloadWithObservedTerminalStatus(synced?.result ?? currentResult, synced?.runnerTrace ?? traceStore?.snapshot?.(candidate.traceId)) ?? (synced?.result ?? currentResult); if (isTraceCommandTerminalStatus(payload?.status)) { diff --git a/internal/cloud/workbench-projection-cursor.test.ts b/internal/cloud/workbench-projection-cursor.test.ts index 08633544..98d6f4f2 100644 --- a/internal/cloud/workbench-projection-cursor.test.ts +++ b/internal/cloud/workbench-projection-cursor.test.ts @@ -80,3 +80,32 @@ test("AgentRun projection state advances from incremental events response", () = assert.equal(state.createdAt, "2026-06-19T12:00:00.000Z"); assert.equal(state.updatedAt, "2026-06-19T12:01:00.000Z"); }); + +test("AgentRun projection state preserves retry params for background resume", () => { + const state = buildWorkbenchProjectionStateUpdate({ + currentState: { + traceId: "trc_cursor_retry_params", + sourceRunId: "run_cursor_retry_params", + sourceCommandId: "cmd_cursor_retry_params", + retryParams: { + message: "preserve this prompt for fresh-session retry", + sessionId: "ses_cursor_retry", + providerProfile: "dsflash-go", + valuesPrinted: false + }, + createdAt: "2026-06-19T12:00:00.000Z" + }, + agentRun: { + runId: "run_cursor_retry_params", + commandId: "cmd_cursor_retry_params", + lastSeq: 3 + }, + now: () => "2026-06-19T12:01:00.000Z" + }); + + assert.equal(state.retryParams.message, "preserve this prompt for fresh-session retry"); + assert.equal(state.retryParams.prompt, "preserve this prompt for fresh-session retry"); + assert.equal(state.retryParams.sessionId, "ses_cursor_retry"); + assert.equal(state.retryParams.providerProfile, "dsflash-go"); + assert.equal(state.retryParams.valuesPrinted, false); +}); diff --git a/internal/cloud/workbench-projection-cursor.ts b/internal/cloud/workbench-projection-cursor.ts index 8ca74894..990eb2d8 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, + retryParams = null, error = null, now = () => new Date().toISOString() } = {}) { @@ -38,8 +39,8 @@ export function buildWorkbenchProjectionStateUpdate({ const mapping = normalizeAgentRunMapping(agentRun ?? result?.agentRun); const timestamp = timestampValue(now); const traceId = textValue(previous.traceId ?? result?.traceId ?? mapping.traceId); - const runId = textValue(previous.sourceRunId ?? previous.runId ?? mapping.runId); - const commandId = textValue(previous.sourceCommandId ?? previous.commandId ?? mapping.commandId); + const runId = textValue(mapping.runId) || textValue(previous.sourceRunId ?? previous.runId); + const commandId = textValue(mapping.commandId) || textValue(previous.sourceCommandId ?? previous.commandId); const lastSourceSeq = Math.max( maxNonNegativeInteger(previous.lastSourceSeq, previous.lastAgentRunSeq), maxNonNegativeInteger(mapping.lastSeq), @@ -50,6 +51,7 @@ export function buildWorkbenchProjectionStateUpdate({ maxNonNegativeInteger(eventsResponse?.maxSeq), lastSourceSeq ); + const nextRetryParams = normalizeProjectionRetryParams(retryParams) ?? normalizeProjectionRetryParams(previous.retryParams); 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"); @@ -82,6 +84,7 @@ export function buildWorkbenchProjectionStateUpdate({ lastErrorMessage: error ? textValue(error.message ?? "Projection sync failed.") : null, failureCount: error ? maxNonNegativeInteger(previous.failureCount) + 1 : 0, nextRetryAt: error ? previous.nextRetryAt ?? null : null, + retryParams: nextRetryParams, createdAt: previous.createdAt ?? timestamp, updatedAt: timestamp, valuesPrinted: false @@ -104,6 +107,32 @@ function normalizeAgentRunMapping(value) { return value && typeof value === "object" && !Array.isArray(value) ? value : {}; } +function normalizeProjectionRetryParams(value) { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + const message = textValue(value.message ?? value.prompt ?? value.text); + if (!message) return null; + const result = { + message, + prompt: message, + text: message, + sessionId: textValue(value.sessionId) || null, + conversationId: textValue(value.conversationId) || null, + threadId: textValue(value.threadId) || null, + projectId: textValue(value.projectId) || null, + ownerUserId: textValue(value.ownerUserId) || null, + ownerRole: textValue(value.ownerRole) || null, + providerProfile: textValue(value.providerProfile ?? value.codeAgentProviderProfile ?? value.backendProfile) || null, + codeAgentProviderProfile: textValue(value.codeAgentProviderProfile ?? value.providerProfile ?? value.backendProfile) || null, + backendProfile: textValue(value.backendProfile ?? value.providerProfile ?? value.codeAgentProviderProfile) || null, + providerId: textValue(value.providerId) || null, + valuesPrinted: false + }; + for (const [key, item] of Object.entries(result)) { + if (item === null || item === "") delete result[key]; + } + return result; +} + function inferProjectionStatus({ lastSourceSeq, sourceLatestSeq, resultSyncState, previous }) { if (["pending", "timed_out", "failed"].includes(resultSyncState)) return "terminal"; if (sourceLatestSeq > lastSourceSeq) return "projecting";