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

fix: scope AgentRun thread reuse to backend profile session

Browse Source
This commit is contained in:
lyon
2026-06-20 10:10:47 +08:00
parent 15e7d74762
commit e91f9b5c0e
1 changed files with 63 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/code-agent-agentrun-adapter.ts
+63 -15
View File
@@ -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,