mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: retry agentrun evicted session turns

Browse Source
This commit is contained in:
root
2026-06-29 16:50:25 +00:00
parent 98de27dd7b
commit 596342da38
3 changed files with 224 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/code-agent-agentrun-adapter.ts
+154 -1
View File
@@ -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);
internal/cloud/server-agent-chat.test.ts
+68 -11
View File
@@ -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())));
internal/cloud/server-code-agent-http.ts
+2 -1
View File
@@ -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 })
});