From c9dd8150f29cc47cf90bd0eaf85c3f93f3b37f38 Mon Sep 17 00:00:00 2001 From: lyon Date: Fri, 26 Jun 2026 08:57:42 +0800 Subject: [PATCH] fix: reuse warm AgentRun runner for HWLAB turns --- internal/cloud/code-agent-agentrun-adapter.ts | 189 ++++++++++-------- .../cloud/code-agent-agentrun-reuse.test.ts | 112 +++++++++++ 2 files changed, 221 insertions(+), 80 deletions(-) create mode 100644 internal/cloud/code-agent-agentrun-reuse.test.ts diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 04ee5b67..9e4a6860 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; draft-2026-06-25-p0-session-warm-runner-contract; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. // Responsibility: AgentRun v0.1 adapter, incremental Workbench projection cursor sync, and upstream timing projection for Cloud API observability. import { randomUUID } from "node:crypto"; @@ -320,8 +320,8 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI type: "request", status: "accepted", label: "agentrun:request:accepted", - message: "HWLAB Code Agent request accepted by the AgentRun v0.1 adapter; hwlab-cloud-api will create a fresh run/command tied to the session-scoped AgentRun PVC, then ensure the runner Job outside the browser admission path.", - waitingFor: "agentrun-run-create", + message: "HWLAB Code Agent request accepted by the AgentRun v0.1 adapter; hwlab-cloud-api will reuse a warm session run when its runner lease is valid, otherwise create a fresh run and runner Job outside the browser admission path.", + waitingFor: "agentrun-run-reuse-check", adapter: ADAPTER_ID, managerHost: new URL(managerUrl).hostname, ...codeAgentOtelTraceFields(traceId, env), @@ -331,10 +331,12 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI const reusable = await resolveReusableAgentRun({ params, options, env, managerUrl, fetchImpl, timeoutMs, backendProfile, traceId, traceStore }); if (reusable?.mapping) { + const reusableParams = paramsWithAgentRunSessionThread(params, reusable.mapping.threadId); + const commandInput = buildAgentRunCommandInput({ params: reusableParams, traceId, backendProfile, sessionId: reusable.sessionId }); + let command = null; + let commandId = null; try { - const reusableParams = paramsWithAgentRunSessionThread(params, reusable.mapping.threadId); - const commandInput = buildAgentRunCommandInput({ params: reusableParams, traceId, backendProfile, sessionId: reusable.sessionId }); - const command = await agentRunDispatchJson({ fetchImpl, managerUrl, path: `/api/v1/runs/${encodeURIComponent(reusable.mapping.runId)}/commands`, + command = await agentRunDispatchJson({ fetchImpl, managerUrl, path: `/api/v1/runs/${encodeURIComponent(reusable.mapping.runId)}/commands`, method: "POST", body: commandInput, timeoutMs, @@ -346,69 +348,7 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI stage: "command-create-reused", terminalOnFailure: false }); - const commandId = requiredString(command?.id, "command.id"); - traceStore.append(traceId, agentRunTraceEvent({ - type: "backend", - status: "running", - label: "agentrun:command:created", - message: `AgentRun command ${commandId} created on reused run ${reusable.mapping.runId}; hwlab-cloud-api will ensure a runner Job is available for this turn.`, - runId: reusable.mapping.runId, - commandId, - backendProfile, - waitingFor: "agentrun-runner-job-ensure", - valuesPrinted: false - }, backendProfile)); - let runnerJob = null; - try { - const runnerJobInput = buildAgentRunRunnerJobInput({ env, traceId, commandId, ownerApiKey, toolCapabilities, backendProfile }); - runnerJob = await agentRunDispatchJson({ fetchImpl, managerUrl, path: `/api/v1/runs/${encodeURIComponent(reusable.mapping.runId)}/runner-jobs`, - method: "POST", - body: runnerJobInput, - timeoutMs, - env, - traceStore, - traceId, - backendProfile, - runId: reusable.mapping.runId, - commandId, - stage: "runner-job-ensure-reused", - terminalOnFailure: false - }); - } catch (error) { - if (!isAgentRunCommandAlreadyClaimed(error)) throw error; - traceStore.append(traceId, agentRunTraceEvent({ - type: "backend", - status: "running", - label: "agentrun:runner-job:already-active", - message: `AgentRun command ${commandId} is already claimed by an active runner; no replacement runner Job is needed for this turn.`, - errorCode: error?.code ?? "agentrun_command_already_claimed", - runId: reusable.mapping.runId, - commandId, - runnerId: reusable.mapping.runnerId ?? null, - jobName: reusable.mapping.jobName ?? null, - namespace: reusable.mapping.namespace ?? null, - waitingFor: "agentrun-result", - valuesPrinted: false - }, backendProfile)); - } - const mapping = agentRunReusedMapping({ previous: reusable.mapping, run: reusable.run, command, runnerJob, traceId, startedAt, backendProfile, managerUrl, env }); - traceStore.append(traceId, agentRunTraceEvent({ - type: "backend", - status: "running", - label: runnerJob ? "agentrun:runner-job:ensured" : "agentrun:runner-job:reused", - message: runnerJob - ? `AgentRun runner Job ${mapping.jobName ?? "unknown"} ensured for reused run ${mapping.runId}; this keeps the persistent session resumable after pod replacement.` - : `AgentRun runner Job ${mapping.jobName ?? "unknown"} is already active for this HWLAB session turn.`, - runId: mapping.runId, - commandId: mapping.commandId, - attemptId: mapping.attemptId, - runnerId: mapping.runnerId, - jobName: mapping.jobName, - namespace: mapping.namespace, - waitingFor: "agentrun-result", - valuesPrinted: false - }, mapping)); - return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params: reusableParams, options, traceId }), mapping, traceStore, traceId }); + commandId = requiredString(command?.id, "command.id"); } catch (error) { traceStore.append(traceId, agentRunTraceEvent({ type: "backend", @@ -421,6 +361,40 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI waitingFor: "agentrun-run-create", valuesPrinted: false }, backendProfile)); + command = null; + commandId = null; + } + if (command && commandId) { + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:command:created", + message: `AgentRun command ${commandId} created on reused run ${reusable.mapping.runId}; the active runner will pick it up without creating another runner Job.`, + runId: reusable.mapping.runId, + commandId, + backendProfile, + waitingFor: "agentrun-result", + valuesPrinted: false + }, backendProfile)); + const runnerJob = null; + const mapping = agentRunReusedMapping({ previous: reusable.mapping, run: reusable.run, command, runnerJob, traceId, startedAt, backendProfile, managerUrl, env }); + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: runnerJob ? "agentrun:runner-job:ensured" : "agentrun:runner-job:reused", + message: runnerJob + ? `AgentRun runner Job ${mapping.jobName ?? "unknown"} ensured for reused run ${mapping.runId}; this keeps the persistent session resumable after pod replacement.` + : `AgentRun runner Job ${mapping.jobName ?? "unknown"} remains active for this HWLAB session turn; no replacement Job was created.`, + runId: mapping.runId, + commandId: mapping.commandId, + attemptId: mapping.attemptId, + runnerId: mapping.runnerId, + jobName: mapping.jobName, + namespace: mapping.namespace, + waitingFor: "agentrun-result", + valuesPrinted: false + }, mapping)); + return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params: reusableParams, options, traceId }), mapping, traceStore, traceId }); } } @@ -698,12 +672,6 @@ function sleepAgentRunDispatchRetry(ms) { return new Promise((resolve) => setTimeout(resolve, Math.max(0, Number(ms) || 0))); } -function isAgentRunCommandAlreadyClaimed(error) { - const statusCode = Number(error?.statusCode ?? 0); - const message = String(error?.message ?? error?.agentRunError?.message ?? ""); - return statusCode === 409 && /command\s+[^\s]+\s+is not pending:/u.test(message); -} - export async function syncAgentRunChatResult({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore, appendResultEvent = true, refreshEvents = true, forceResultSync = false }) { const initial = currentResult ?? await loadPersistedAgentRunResult(traceId, options); if (isAgentRunUserCancelSealed(initial)) { @@ -1683,22 +1651,83 @@ async function resolveReusableAgentRun({ params = {}, options = {}, env = proces }, backendProfile)); return null; } + let run = null; + try { + run = await agentRunDispatchJson({ + fetchImpl, + managerUrl, + path: `/api/v1/runs/${encodeURIComponent(mapping.runId)}`, + method: "GET", + timeoutMs, + env, + traceStore, + traceId, + backendProfile, + runId: mapping.runId, + stage: "run-reuse-check", + terminalOnFailure: false + }); + } catch (error) { + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:run:reuse-blocked", + message: `Stored AgentRun run ${mapping.runId} could not be read from AgentRun manager; hwlab-cloud-api will create a fresh run.`, + errorCode: error?.code ?? error?.agentRunError?.failureKind ?? "agentrun_reuse_check_failed", + runId: mapping.runId, + commandId: mapping.commandId ?? null, + runnerId: mapping.runnerId ?? null, + jobName: mapping.jobName ?? null, + namespace: mapping.namespace ?? null, + sessionId: hwlabSessionId, + agentRunSessionId: mapping.sessionId ?? null, + waitingFor: "agentrun-run-create", + valuesPrinted: false + }, backendProfile)); + return null; + } + const blocker = agentRunReuseBlocker(run, mapping); + if (blocker) { + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:run:reuse-blocked", + message: `Stored AgentRun run ${mapping.runId} cannot be reused (${blocker}); hwlab-cloud-api will create a fresh run while preserving the session-scoped AgentRun PVC.`, + runId: mapping.runId, + commandId: mapping.commandId ?? null, + runnerId: mapping.runnerId ?? null, + jobName: mapping.jobName ?? null, + namespace: mapping.namespace ?? null, + sessionId: hwlabSessionId, + agentRunSessionId: mapping.sessionId ?? null, + reuseBlocker: blocker, + runStatus: run?.terminalStatus ?? run?.status ?? mapping.runStatus ?? null, + leaseExpiresAt: run?.leaseExpiresAt ?? null, + waitingFor: "agentrun-run-create", + valuesPrinted: false + }, backendProfile)); + return null; + } + const agentRunScopedSessionId = safeSessionId(run?.sessionRef?.sessionId) || safeSessionId(mapping.sessionId) || scopedAgentRunSessionIdForParams(params, traceId); traceStore.append(traceId, agentRunTraceEvent({ type: "backend", status: "running", - label: "agentrun:run:reuse-disabled", - message: `Stored AgentRun run ${mapping.runId} is not reused for a new HWLAB turn; hwlab-cloud-api will create a fresh run while preserving the session-scoped AgentRun PVC.`, + label: "agentrun:run:reuse-enabled", + message: `Stored AgentRun run ${mapping.runId} is still claimed by an active runner; hwlab-cloud-api will append the next command without creating another runner Job.`, runId: mapping.runId, commandId: mapping.commandId ?? null, runnerId: mapping.runnerId ?? null, jobName: mapping.jobName ?? null, namespace: mapping.namespace ?? null, sessionId: hwlabSessionId, - agentRunSessionId: mapping.sessionId ?? null, - waitingFor: "agentrun-run-create", + agentRunSessionId: agentRunScopedSessionId, + runStatus: run?.status ?? mapping.runStatus ?? null, + claimedBy: run?.claimedBy ?? null, + leaseExpiresAt: run?.leaseExpiresAt ?? null, + waitingFor: "agentrun-command-create", valuesPrinted: false }, backendProfile)); - return null; + return { mapping, run, sessionId: agentRunScopedSessionId }; } function isReusableAgentRunMapping(mapping, backendProfile) { diff --git a/internal/cloud/code-agent-agentrun-reuse.test.ts b/internal/cloud/code-agent-agentrun-reuse.test.ts new file mode 100644 index 00000000..ea219ce4 --- /dev/null +++ b/internal/cloud/code-agent-agentrun-reuse.test.ts @@ -0,0 +1,112 @@ +// SPEC: PJ2026-010205 HWLAB接入 draft-2026-06-25-p0-session-warm-runner-contract. +// Responsibility: AgentRun warm-runner reuse regressions for HWLAB Code Agent turns. + +import assert from "node:assert/strict"; +import { test } from "bun:test"; + +import { submitAgentRunChatTurn } from "./code-agent-agentrun-adapter.ts"; +import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts"; + +test("AgentRun adapter reuses a claimed session run without creating another runner Job", async () => { + const traceId = "trc_agentrun_reuse_warm_runner"; + const calls = []; + const leaseExpiresAt = new Date(Date.now() + 120_000).toISOString(); + const traceStore = createCodeAgentTraceStore(); + + const result = await submitAgentRunChatTurn({ + traceId, + traceStore, + params: { + sessionId: "ses_reuse_warm_runner", + conversationId: "cnv_reuse_warm_runner", + message: "reuse warm runner", + providerProfile: "deepseek" + }, + options: { + env: { + AGENTRUN_MGR_URL: "http://127.0.0.1:65535", + HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1", + HWLAB_CODE_AGENT_AGENTRUN_HTTP_TIMEOUT_MS: "1000", + HWLAB_CODE_AGENT_AGENTRUN_DISPATCH_RETRY_MAX: "0", + HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601", + HWLAB_CODE_AGENT_AGENTRUN_SOURCE_COMMIT: "0123456789abcdef0123456789abcdef01234567", + HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE: "deepseek" + }, + accessController: { + async getAgentSession(sessionId) { + assert.equal(sessionId, "ses_reuse_warm_runner"); + return { + ownerUserId: null, + session: { + agentRun: { + adapter: "agentrun-v01", + managerUrl: "http://127.0.0.1:65535", + backendProfile: "deepseek", + providerId: "D601", + runId: "run_reuse_warm_runner", + commandId: "cmd_previous_turn", + runnerId: "runner_warm", + jobName: "agentrun-v01-runner-warm", + namespace: "agentrun-v01", + status: "runner-job-created", + runStatus: "claimed", + sessionId: "ses_agentrun_reuse_warm_runner", + conversationId: "cnv_reuse_warm_runner", + threadId: "thr_reuse_warm_runner", + reuseEligible: true, + runnerJobCount: 1 + } + } + }; + } + }, + async fetchImpl(url, init = {}) { + const parsed = new URL(url); + const method = init.method ?? "GET"; + const body = init.body ? JSON.parse(String(init.body)) : null; + calls.push({ method, path: parsed.pathname, body }); + + const send = (data) => new Response(JSON.stringify({ ok: true, data }), { status: 200, headers: { "content-type": "application/json" } }); + if (method === "GET" && parsed.pathname === "/api/v1/runs/run_reuse_warm_runner") { + return send({ + id: "run_reuse_warm_runner", + status: "claimed", + claimedBy: "runner_warm", + leaseExpiresAt, + sessionRef: { + sessionId: "ses_agentrun_reuse_warm_runner", + conversationId: "cnv_reuse_warm_runner", + threadId: "thr_reuse_warm_runner" + } + }); + } + if (method === "POST" && parsed.pathname === "/api/v1/runs/run_reuse_warm_runner/commands") { + return send({ id: "cmd_reused_turn", runId: "run_reuse_warm_runner", state: "pending", type: "turn", seq: 2 }); + } + throw new Error(`unexpected AgentRun call ${method} ${parsed.pathname}`); + } + } + }); + + assert.equal(result.status, "running"); + assert.equal(result.agentRun.reused, true); + assert.equal(result.agentRun.runnerReused, true); + assert.equal(result.agentRun.runId, "run_reuse_warm_runner"); + assert.equal(result.agentRun.commandId, "cmd_reused_turn"); + assert.equal(result.agentRun.runnerJobCount, 1); + assert.deepEqual(calls.map((call) => `${call.method} ${call.path}`), [ + "GET /api/v1/runs/run_reuse_warm_runner", + "POST /api/v1/runs/run_reuse_warm_runner/commands" + ]); + assert.equal(calls[1].body.type, "turn"); + assert.equal(calls[1].body.payload.sessionId, "ses_agentrun_reuse_warm_runner"); + assert.equal(calls[1].body.payload.hwlabSessionId, "ses_reuse_warm_runner"); + assert.equal(calls[1].body.payload.threadId, "thr_reuse_warm_runner"); + assert.equal(calls[1].body.idempotencyKey, traceId); + assert.equal(calls.some((call) => call.path.endsWith("/runner-jobs")), false); + + const labels = traceStore.snapshot(traceId).events.map((event) => event.label); + assert.ok(labels.includes("agentrun:run:reuse-enabled")); + assert.ok(labels.includes("agentrun:runner-job:reused")); + assert.equal(labels.includes("agentrun:run:reuse-disabled"), false); +});