diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index e0fa2fea..8ee8ac0e 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -674,7 +674,7 @@ function sleepAgentRunDispatchRetry(ms) { return new Promise((resolve) => setTimeout(resolve, Math.max(0, Number(ms) || 0))); } -export async function syncAgentRunChatResult({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore, appendResultEvent = true, refreshEvents = true, forceResultSync = false }) { +export async function syncAgentRunChatResult({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore, appendResultEvent = true, refreshEvents = true, forceResultSync = false, retryParams = null }) { const initial = currentResult ?? await loadPersistedAgentRunResult(traceId, options); if (isAgentRunUserCancelSealed(initial)) { const runnerTrace = initial.runnerTrace ?? traceStore.snapshot(traceId); @@ -760,6 +760,33 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op updatedAt: nowIso(options.now), valuesPrinted: false }); + const freshSessionRetry = await retryAgentRunTurnAfterFreshSessionFailure({ + params: retryParams, + options, + traceId, + traceStore, + mapped, + failedResult: result, + failedMapping: nextMapping, + env, + managerUrl, + fetchImpl, + timeoutMs, + backendProfile + }); + if (freshSessionRetry) { + options.codeAgentChatResults?.set?.(traceId, freshSessionRetry.payload); + await writeWorkbenchProjectionStateForAgentRun(options.runtimeStore, { + currentState: projectionState, + result: freshSessionRetry.payload, + agentRun: freshSessionRetry.mapping, + eventsResponse: null, + resultSyncState: null, + projectionStatus: "projecting", + now: options.now + }); + return { result: freshSessionRetry.payload, runnerTrace: freshSessionRetry.payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: false, freshSessionRetried: true }; + } const base = { ...mapped, agentRun: nextMapping, updatedAt: nowIso(options.now) }; const payload = agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, appendResultEvent }); options.codeAgentChatResults?.set?.(traceId, payload); @@ -1296,6 +1323,132 @@ function newSessionIdAfterEviction(baseSessionId, traceId) { const profile = String(traceId).replace(/[^A-Za-z0-9]/gu, "").slice(0, 12) || "fresh"; return baseSessionId + "-reset-" + profile; } + +async function retryAgentRunTurnAfterFreshSessionFailure({ + params = null, + options = {}, + traceId, + traceStore = defaultCodeAgentTraceStore, + mapped = {}, + failedResult = {}, + failedMapping = {}, + env = process.env, + managerUrl, + fetchImpl, + timeoutMs, + backendProfile +} = {}) { + const failure = agentRunFreshSessionFailure(failedResult); + if (!failure.requiresFreshSession) return null; + if (mapped?.agentRun?.freshSessionRetryOf || failedMapping?.freshSessionRetryOf) return null; + const promptText = messageAuthorityTextValue(params?.message ?? params?.prompt ?? params?.text ?? ""); + if (!promptText) { + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "failed", + label: "agentrun:session-reset:skipped", + errorCode: "agentrun_retry_prompt_missing", + message: "AgentRun session storage was evicted, but hwlab-cloud-api cannot retry the current turn because the original prompt text is unavailable in this sync path.", + runId: failedMapping?.runId ?? mapped?.agentRun?.runId ?? null, + commandId: failedMapping?.commandId ?? mapped?.agentRun?.commandId ?? null, + terminal: true, + valuesPrinted: false + }, failedMapping || mapped?.agentRun || backendProfile)); + return null; + } + const baseSessionId = scopedAgentRunSessionIdForParams(params, traceId); + const sessionId = newSessionIdAfterEviction(baseSessionId, traceId); + const retryParams = paramsWithAgentRunSessionThread(params, null); + const startedAt = nowIso(options.now); + const previous = { + runId: failedMapping?.runId ?? mapped?.agentRun?.runId ?? null, + commandId: failedMapping?.commandId ?? mapped?.agentRun?.commandId ?? null, + sessionId: failedMapping?.sessionId ?? mapped?.agentRun?.sessionId ?? baseSessionId, + threadId: failedMapping?.threadId ?? mapped?.agentRun?.threadId ?? mapped?.threadId ?? null, + failureKind: failure.code, + failureMessage: failure.message, + valuesPrinted: false + }; + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:session-reset:retry-current-turn", + message: "AgentRun session storage/thread resume failed; hwlab-cloud-api is retrying the same user turn with a fresh AgentRun session and threadId=null.", + runId: previous.runId, + commandId: previous.commandId, + sessionId, + previousSessionId: previous.sessionId, + previousThreadId: previous.threadId, + errorCode: failure.code, + backendProfile, + waitingFor: "agentrun-run-create", + valuesPrinted: false + }, failedMapping || mapped?.agentRun || backendProfile)); + await ensureAgentRunSessionPersistent({ fetchImpl, managerUrl, sessionId, env, traceId, backendProfile, traceStore }); + const ownerApiKey = await resolveOwnerApiKey({ params: retryParams, options, now: () => nowIso(options.now) }); + const toolCapabilities = await resolveToolCapabilities({ params: retryParams, options }); + const runInput = buildAgentRunCreateRunInput({ params: retryParams, env, traceId, backendProfile, sessionId, toolCapabilities }); + const run = await agentRunDispatchJson({ fetchImpl, managerUrl, path: "/api/v1/runs", method: "POST", body: runInput, timeoutMs, env, traceStore, traceId, backendProfile, stage: "run-create-after-session-reset" }); + const runId = requiredString(run?.id, "run.id"); + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:run:created-after-session-reset", + message: "AgentRun fresh-session retry run " + runId + " created for the current user turn.", + runId, + backendProfile, + waitingFor: "agentrun-command-create", + valuesPrinted: false + }, backendProfile)); + const commandInput = buildAgentRunCommandInput({ params: retryParams, traceId, backendProfile, sessionId }); + const command = await agentRunDispatchJson({ fetchImpl, managerUrl, path: "/api/v1/runs/" + encodeURIComponent(runId) + "/commands", method: "POST", body: commandInput, timeoutMs, env, traceStore, traceId, backendProfile, runId, stage: "command-create-after-session-reset" }); + const commandId = requiredString(command?.id, "command.id"); + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:command:created-after-session-reset", + message: "AgentRun fresh-session retry command " + commandId + " created for the current user turn.", + runId, + commandId, + backendProfile, + waitingFor: "agentrun-runner-job-create", + valuesPrinted: false + }, backendProfile)); + const runnerJobInput = buildAgentRunRunnerJobInput({ env, traceId, commandId, ownerApiKey, toolCapabilities, backendProfile }); + const mapping = { + ...agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob: null, traceId, startedAt, params: retryParams }), + freshSessionRetryAttempt: 1, + freshSessionRetryOf: previous, + status: "runner-job-pending", + valuesPrinted: false + }; + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:runner-job:queued-after-session-reset", + message: "AgentRun fresh-session retry run/command admission is durable; hwlab-cloud-api will create the runner Job asynchronously.", + runId: mapping.runId, + commandId: mapping.commandId, + backendProfile, + waitingFor: "agentrun-runner-job-create", + valuesPrinted: false + }, mapping)); + scheduleAgentRunRunnerJobCreateAfterAdmission({ fetchImpl, managerUrl, runnerJobInput, timeoutMs, env, traceStore, traceId, backendProfile, run, command, mapping, options, params: retryParams }); + const base = initialAgentRunChatResult({ params: retryParams, options, traceId }); + const payload = decorateAgentRunRunningResult({ base: { ...base, freshSessionRetryOf: previous }, mapping, traceStore, traceId }); + return { payload, mapping }; +} + +function agentRunFreshSessionFailure(result = {}) { + const code = result?.failureKind ?? result?.errorCode ?? result?.code ?? (result?.terminalStatus === "failed" ? "agentrun_failed" : null); + const message = result?.failureMessage ?? result?.blocker?.message ?? result?.error?.message ?? result?.message ?? ""; + return { + code: normalizedFailureKind(code) ?? "agentrun_failed", + message, + requiresFreshSession: agentRunFailureRequiresFreshSession(code, message), + valuesPrinted: false + }; +} function buildAgentRunCreateRunInput({ params, env, traceId, backendProfile, sessionId, toolCapabilities = null }) { const commitId = requireAgentRunSourceCommit(env); const runtimeAuthority = requireAgentRunRuntimeAuthority(env); diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 43751887..cc00dcb7 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/internal/cloud/server-agent-chat.test.ts @@ -1562,22 +1562,35 @@ test("cloud api AgentRun adapter exposes invalid tool-call attribution in result } }); -test("cloud api AgentRun adapter marks evicted session storage as non-reusable", async () => { +test("cloud api AgentRun adapter retries evicted session storage with a fresh session", async () => { + const calls = []; 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, body }); const send = (data) => { response.writeHead(200, { "content-type": "application/json" }); response.end(`${JSON.stringify({ ok: true, data, traceId: "trc_fake_agentrun_evicted" })}\n`); }; + if (request.method === "POST" && url.pathname === "/api/v1/sessions") { + return send({ sessionId: body.sessionId, backendProfile: body.backendProfile }); + } if (request.method === "POST" && url.pathname === "/api/v1/runs") { - return send({ id: "run_evicted", status: "pending", backendProfile: "deepseek", sessionRef: body.sessionRef, resourceBundleRef: body.resourceBundleRef }); + const retry = String(body.sessionRef?.sessionId ?? "").includes("-reset-"); + return send({ id: retry ? "run_evicted_retry" : "run_evicted", status: "pending", backendProfile: "deepseek", sessionRef: body.sessionRef, resourceBundleRef: body.resourceBundleRef }); } if (request.method === "POST" && url.pathname === "/api/v1/runs/run_evicted/commands") { return send({ id: "cmd_evicted", runId: "run_evicted", state: "queued" }); } + if (request.method === "POST" && url.pathname === "/api/v1/runs/run_evicted_retry/commands") { + assert.equal(body.type, "turn"); + assert.equal(body.payload.threadId, null); + assert.match(body.payload.prompt, /resume evicted session/u); + assert.match(body.payload.sessionId, /-reset-/u); + return send({ id: "cmd_evicted_retry", runId: "run_evicted_retry", state: "queued" }); + } if (request.method === "POST" && url.pathname === "/api/v1/runs/run_evicted/runner-jobs") { return send({ action: "create-kubernetes-job", @@ -1590,15 +1603,37 @@ test("cloud api AgentRun adapter marks evicted session storage as non-reusable", runner: { attemptId: "attempt_evicted", runnerId: "runner_evicted" } }); } + if (request.method === "POST" && url.pathname === "/api/v1/runs/run_evicted_retry/runner-jobs") { + return send({ + action: "create-kubernetes-job", + runId: "run_evicted_retry", + commandId: "cmd_evicted_retry", + attemptId: "attempt_evicted_retry", + runnerId: "runner_evicted_retry", + namespace: "agentrun-v01", + jobName: "agentrun-v01-runner-evicted-retry", + runner: { attemptId: "attempt_evicted_retry", runnerId: "runner_evicted_retry" } + }); + } if (request.method === "GET" && url.pathname === "/api/v1/runs/run_evicted/events") { return send({ items: [ { id: "evt_evicted", runId: "run_evicted", seq: 1, type: "error", payload: { commandId: "cmd_evicted", 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-02T00:00:00.000Z" }, { id: "evt_evicted_terminal", runId: "run_evicted", seq: 2, type: "terminal_status", payload: { commandId: "cmd_evicted", terminalStatus: "failed", failureKind: "session-store-evicted" }, createdAt: "2026-06-02T00:00:01.000Z" } ] }); } + if (request.method === "GET" && url.pathname === "/api/v1/runs/run_evicted_retry/events") { + return send({ items: [ + { id: "evt_evicted_retry_1", runId: "run_evicted_retry", seq: 1, type: "backend_status", payload: { commandId: "cmd_evicted_retry", phase: "runner-job-created" }, createdAt: "2026-06-02T00:00:02.000Z" }, + { id: "evt_evicted_retry_2", runId: "run_evicted_retry", seq: 2, type: "assistant_message", payload: { commandId: "cmd_evicted_retry", text: "EVICTED_RETRY_OK", final: true, replyAuthority: true }, createdAt: "2026-06-02T00:00:03.000Z" }, + { id: "evt_evicted_retry_terminal", runId: "run_evicted_retry", seq: 3, type: "terminal_status", payload: { commandId: "cmd_evicted_retry", terminalStatus: "completed" }, createdAt: "2026-06-02T00:00:04.000Z" } + ] }); + } if (request.method === "GET" && url.pathname === "/api/v1/runs/run_evicted/commands") { return send({ items: [{ id: "cmd_evicted", runId: "run_evicted", state: "failed", terminalStatus: "failed", idempotencyKey: "trc_server-test-agentrun-evicted", payload: { traceId: "trc_server-test-agentrun-evicted" } }] }); } + if (request.method === "GET" && url.pathname === "/api/v1/runs/run_evicted_retry/commands") { + return send({ items: [{ id: "cmd_evicted_retry", runId: "run_evicted_retry", state: "completed", terminalStatus: "completed", idempotencyKey: "trc_server-test-agentrun-evicted", payload: { traceId: "trc_server-test-agentrun-evicted" } }] }); + } if (request.method === "GET" && url.pathname === "/api/v1/runs/run_evicted/commands/cmd_evicted/result") { return send({ runId: "run_evicted", @@ -1619,6 +1654,25 @@ test("cloud api AgentRun adapter marks evicted session storage as non-reusable", sessionRef: { sessionId: "ses_agentrun_evicted", conversationId: "cnv_evicted", threadId: "thread_evicted" } }); } + if (request.method === "GET" && url.pathname === "/api/v1/runs/run_evicted_retry/commands/cmd_evicted_retry/result") { + return send({ + runId: "run_evicted_retry", + commandId: "cmd_evicted_retry", + attemptId: "attempt_evicted_retry", + runnerId: "runner_evicted_retry", + jobName: "agentrun-v01-runner-evicted-retry", + namespace: "agentrun-v01", + status: "completed", + runStatus: "completed", + commandState: "completed", + terminalStatus: "completed", + completed: true, + reply: "EVICTED_RETRY_OK", + lastSeq: 3, + eventCount: 3, + sessionRef: { sessionId: "ses_agentrun_evicted-reset-trcservertes", conversationId: "cnv_evicted", threadId: "thread_evicted_retry" } + }); + } response.writeHead(404, { "content-type": "application/json" }); response.end(`${JSON.stringify({ ok: false, failureKind: "schema-invalid", message: `unexpected ${request.method} ${url.pathname}`, traceId: "trc_fake_agentrun_evicted" })}\n`); }); @@ -1669,15 +1723,18 @@ test("cloud api AgentRun adapter marks evicted session storage as non-reusable", assert.equal(submit.status, 202); const payload = await pollAgentResult(port, traceId); validateCodeAgentChatSchema(payload); - assert.equal(payload.status, "failed"); - assert.equal(payload.error.code, "session-store-evicted"); - assert.equal(payload.session.status, "thread-resume-failed"); - assert.equal(payload.session.lifecycle.requiresNewSession, true); - assert.equal(payload.sessionReuse.threadId, "thread_evicted"); - assert.equal(payload.sessionReuse.status, "thread-resume-failed"); - assert.equal(payload.agentRun.reuseEligible, false); - assert.equal(payload.reuseEligible, false); - assert.equal(ownerSessions.get("ses_server-test-evicted")?.status, "thread-resume-failed"); + assert.equal(payload.status, "completed"); + assert.equal(payload.reply.content, "EVICTED_RETRY_OK"); + assert.equal(payload.agentRun.runId, "run_evicted_retry"); + assert.equal(payload.agentRun.commandId, "cmd_evicted_retry"); + assert.equal(payload.agentRun.freshSessionRetryOf.runId, "run_evicted"); + assert.equal(payload.agentRun.freshSessionRetryOf.commandId, "cmd_evicted"); + assert.match(payload.agentRun.sessionId, /-reset-/u); + assert.equal(payload.session.threadId, "thread_evicted_retry"); + assert.equal(payload.sessionReuse.status, "thread-resumed"); + assert.equal(payload.error, undefined); + assert.equal(ownerSessions.get("ses_server-test-evicted")?.status, "completed"); + assert.equal(calls.some((call) => call.method === "POST" && call.path === "/api/v1/runs/run_evicted_retry/commands"), true); } 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()))); diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 97b03fc6..536fc67f 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -1432,7 +1432,8 @@ function scheduleAgentRunProjectionSync({ traceId, params = {}, options = {}, tr currentResult: result, options: refreshOptions, traceStore, - forceResultSync: agentRunProjectionObservedTerminal(result, traceStore?.snapshot?.(traceId)) + forceResultSync: agentRunProjectionObservedTerminal(result, traceStore?.snapshot?.(traceId)), + retryParams: params }), onTerminal: (payload, finalizerOptions = {}) => scheduleCodeAgentTerminalTurnStatusEffects({ payload: codeAgentPayloadWithObservedTerminalStatus(payload, traceStore?.snapshot?.(traceId)) ?? payload, params, options, preserveLastTraceId: finalizerOptions.preserveLastTraceId === true }) });