Merge pull request #2184 from pikasTech/fix/2171-agentrun-run-reuse

fix: 复用 AgentRun warm runner
2026-06-26 08:58:30 +08:00
parent ff15725fcb c9dd8150f2
commit 6cfb3b47cd
2 changed files with 221 additions and 80 deletions
@@ -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) {
@@ -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);
+});