From e91f9b5c0e7391d11cd7c1a6344a443d6e3dbf49 Mon Sep 17 00:00:00 2001 From: lyon Date: Sat, 20 Jun 2026 10:10:47 +0800 Subject: [PATCH] fix: scope AgentRun thread reuse to backend profile session --- internal/cloud/code-agent-agentrun-adapter.ts | 78 +++++++++++++++---- 1 file changed, 63 insertions(+), 15 deletions(-) diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 1561401e..df064204 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -286,7 +286,8 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI const reusable = await resolveReusableAgentRun({ params, options, env, managerUrl, fetchImpl, timeoutMs, backendProfile, traceId, traceStore }); if (reusable?.mapping) { try { - const commandInput = buildAgentRunCommandInput({ params, traceId, backendProfile, sessionId: reusable.sessionId }); + const reusableParams = paramsWithAgentRunSessionThread(params, reusable.mapping.threadId); + const commandInput = buildAgentRunCommandInput({ params: reusableParams, traceId, backendProfile, sessionId: reusable.sessionId }); const command = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(reusable.mapping.runId)}/commands`, { method: "POST", body: commandInput, @@ -348,7 +349,7 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI waitingFor: "agentrun-result", valuesPrinted: false }, mapping)); - return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params, options, traceId }), mapping, traceStore, traceId }); + return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params: reusableParams, options, traceId }), mapping, traceStore, traceId }); } catch (error) { traceStore.append(traceId, agentRunTraceEvent({ type: "backend", @@ -367,11 +368,13 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI const baseSessionId = scopedAgentRunSessionIdForParams(params, traceId, backendProfile); let sessionId = baseSessionId; let sessionReset = false; + let finalDispatchParams = params; let run, command, runnerJob, mapping; for (let attempt = 0; attempt < 2; attempt += 1) { + let attemptParams = params; if (sessionReset) { sessionId = newSessionIdAfterEviction(baseSessionId, traceId); - const resetParams = { ...params, threadId: null }; + attemptParams = paramsWithAgentRunSessionThread(params, null); traceStore.append(traceId, agentRunTraceEvent({ type: "backend", status: "running", label: "agentrun:session-reset", @@ -379,8 +382,9 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI sessionId, previousSessionId: baseSessionId, backendProfile, waitingFor: "agentrun-run-create", valuesPrinted: false, }, backendProfile)); } + let ensuredSession = null; try { - await ensureAgentRunSessionPersistent({ fetchImpl, managerUrl, sessionId, env, traceId, backendProfile, traceStore }); + ensuredSession = await ensureAgentRunSessionPersistent({ fetchImpl, managerUrl, sessionId, env, traceId, backendProfile, traceStore }); } catch (error) { if (attempt === 0 && await shouldResetSessionAfterEviction("session-store-evicted", error?.message)) { sessionReset = true; @@ -388,7 +392,21 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI } throw error; } - const runInput = buildAgentRunCreateRunInput({ params, env, traceId, backendProfile, sessionId, toolCapabilities }); + const profileThreadId = agentRunSessionThreadIdFromEnsure(ensuredSession); + const dispatchParams = paramsWithAgentRunSessionThread(attemptParams, profileThreadId); + if (safeOpaqueId(attemptParams.threadId) && !profileThreadId) { + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:profile-session-thread-reset", + message: "AgentRun profile-scoped session " + sessionId + " has no threadId; hwlab-cloud-api will start a new thread instead of resuming a thread from another backend profile.", + sessionId, + backendProfile, + waitingFor: "agentrun-run-create", + valuesPrinted: false + }, backendProfile)); + } + const runInput = buildAgentRunCreateRunInput({ params: dispatchParams, env, traceId, backendProfile, sessionId, toolCapabilities }); const runCreateStartedAt = Date.now(); run = await agentRunJson(fetchImpl, managerUrl, "/api/v1/runs", { method: "POST", body: runInput, timeoutMs, env }); const runId = requiredString(run?.id, "run.id"); @@ -399,7 +417,7 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI message: "AgentRun run " + runId + " created through internal k3s Service DNS.", runId, backendProfile, waitingFor: "agentrun-command-create", valuesPrinted: false, }, backendProfile)); - const commandInput = buildAgentRunCommandInput({ params, traceId, backendProfile, sessionId }); + const commandInput = buildAgentRunCommandInput({ params: dispatchParams, traceId, backendProfile, sessionId }); const commandCreateStartedAt = Date.now(); command = await agentRunJson(fetchImpl, managerUrl, "/api/v1/runs/" + encodeURIComponent(runId) + "/commands", { method: "POST", body: commandInput, timeoutMs, env }); const commandId = requiredString(command?.id, "command.id"); @@ -422,7 +440,8 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI } throw error; } - mapping = agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt, params }); + mapping = agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt, params: dispatchParams }); + finalDispatchParams = dispatchParams; break; } traceStore.append(traceId, agentRunTraceEvent({ @@ -439,7 +458,7 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI waitingFor: "agentrun-result", valuesPrinted: false }, mapping)); - return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params, options, traceId }), mapping, traceStore, traceId }); + return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params: finalDispatchParams, options, traceId }), mapping, traceStore, traceId }); } function isAgentRunCommandAlreadyClaimed(error) { @@ -943,13 +962,13 @@ function withoutUserBillingReservation(value = {}) { async function ensureAgentRunSessionPersistent({ fetchImpl, managerUrl, sessionId, env, traceId, backendProfile, traceStore }) { const defaultPolicy = firstNonEmpty(env.HWLAB_CODE_AGENT_AGENTRUN_SESSION_STORAGE, "persistent"); - if (defaultPolicy !== "persistent") return; + if (defaultPolicy !== "persistent") return null; try { const tenantId = firstNonEmpty(env.HWLAB_CODE_AGENT_AGENTRUN_TENANT_ID, DEFAULT_TENANT_ID); const projectId = agentRunProjectIdForEnv(env); const expiresInDays = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_SESSION_TTL_DAYS, 30); const expiresAt = new Date(Date.now() + Math.max(1, expiresInDays) * 24 * 60 * 60 * 1000).toISOString(); - await agentRunJson(fetchImpl, managerUrl, "/api/v1/sessions", { + return await agentRunJson(fetchImpl, managerUrl, "/api/v1/sessions", { method: "POST", timeoutMs: 15_000, env, @@ -972,9 +991,29 @@ async function ensureAgentRunSessionPersistent({ fetchImpl, managerUrl, sessionI message: "AgentRun session " + sessionId + " pre-flight could not ensure PVC; falling back to legacy metadata-only session (" + message + ").", sessionId, backendProfile, valuesPrinted: false, }, backendProfile)); + return null; } } +function agentRunSessionRecordFromEnsure(value = null) { + const record = value && typeof value === "object" ? value : null; + if (!record) return null; + const session = record.session && typeof record.session === "object" ? record.session : record; + return session && typeof session === "object" ? session : null; +} + +function agentRunSessionThreadIdFromEnsure(value = null) { + return safeOpaqueId(agentRunSessionRecordFromEnsure(value)?.threadId) || null; +} + +function paramsWithAgentRunSessionThread(params = {}, threadId = null) { + const next = { ...(params ?? {}) }; + const safeThreadId = safeOpaqueId(threadId); + if (safeThreadId) next.threadId = safeThreadId; + else delete next.threadId; + return next; +} + function agentRunProjectIdForEnv(env = process.env) { return firstNonEmpty(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECT_ID, DEFAULT_PROJECT_ID); } @@ -1607,7 +1646,7 @@ function agentRunProviderTrace({ base = {}, result = {}, terminalStatus = null } jobName: base.agentRun?.jobName ?? result?.jobName ?? null, namespace: base.agentRun?.namespace ?? result?.namespace ?? null, traceId: base.traceId ?? base.agentRun?.traceId ?? null, - threadId: result?.sessionRef?.threadId ?? base.agentRun?.threadId ?? base.threadId ?? null, + threadId: result?.sessionRef?.threadId ?? agentRunEffectiveThreadId(base), turnId: result?.turnId ?? null, terminalStatus: terminalStatus ?? result?.terminalStatus ?? null, failureKind: result?.failureKind ?? null, @@ -1743,6 +1782,7 @@ function agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, a return { ...base, status: "completed", + threadId: agentRunEffectiveThreadId(base), updatedAt: now, workspace: base.workspace ?? "/home/agentrun/workspace", sandbox: base.sandbox ?? "danger-full-access", @@ -1796,10 +1836,11 @@ function agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, a const partialContext = partialAgentRunContext(runnerTrace); const freshSessionRequired = !canceled && agentRunFailureRequiresFreshSession(code, message); const failedSessionStatus = freshSessionRequired ? "thread-resume-failed" : "failed"; - const resumableAfterFailure = !canceled && !freshSessionRequired && Boolean(safeOpaqueId(base.threadId ?? base.agentRun?.threadId)); + const resumableAfterFailure = !canceled && !freshSessionRequired && Boolean(agentRunEffectiveThreadId(base)); return { ...base, status: canceled ? "canceled" : "failed", + threadId: agentRunEffectiveThreadId(base), canceled, updatedAt: now, session: agentRunSessionSummary(base, canceled ? "canceled" : failedSessionStatus), @@ -1970,6 +2011,7 @@ function decorateAgentRunRunningResult({ base, mapping, traceStore, traceId }) { return { ...base, status: "running", + threadId: agentRunEffectiveThreadId({ agentRun: mapping }), accepted: true, shortConnection: true, updatedAt: nowIso(), @@ -2660,11 +2702,17 @@ function agentRunSessionId(traceId) { return `ses_agentrun_${String(safeTraceId(traceId) || `trc_${randomUUID()}`).slice(4)}`; } +function agentRunEffectiveThreadId(base = {}) { + const agentRun = base?.agentRun && typeof base.agentRun === "object" ? base.agentRun : null; + if (agentRun && Object.hasOwn(agentRun, "threadId")) return safeOpaqueId(agentRun.threadId) || null; + return safeOpaqueId(base?.threadId) || null; +} + function agentRunSessionSummary(base, status) { return decorateCodeAgentSession({ sessionId: base.sessionId ?? base.agentRun?.sessionId ?? null, conversationId: base.conversationId ?? base.agentRun?.conversationId ?? null, - threadId: base.threadId ?? base.agentRun?.threadId ?? null, + threadId: agentRunEffectiveThreadId(base), status, sessionMode: AGENTRUN_SESSION_MODE, runnerKind: AGENTRUN_RUNNER_KIND, @@ -2684,7 +2732,7 @@ function agentRunSessionSummary(base, status) { function agentRunThreadReused(base = {}, options = {}) { if (options.status === "failed-requires-new-session") return false; if (base.agentRun?.threadReused === true || base.agentRun?.persistentResume === true) return true; - return Boolean(safeOpaqueId(base.threadId)); + return Boolean(agentRunEffectiveThreadId(base)); } function agentRunSessionReuseSummary(base, reused, options = {}) { @@ -2695,7 +2743,7 @@ function agentRunSessionReuseSummary(base, reused, options = {}) { return { sessionId: base.sessionId ?? base.agentRun?.sessionId ?? null, conversationId: base.conversationId ?? base.agentRun?.conversationId ?? null, - threadId: base.threadId ?? base.agentRun?.threadId ?? null, + threadId: agentRunEffectiveThreadId(base), reused: sessionReused, status: options.status ?? (runnerReused ? "reused" : persistentResume ? "thread-resumed" : "new"), runnerReused,