fix(cloud): expose persistent thread resume metadata

2026-06-04 13:05:03 +08:00
parent f7e4c372ce
commit 3a55a4b99e
2 changed files with 222 additions and 5 deletions
@@ -238,7 +238,7 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI
      }
      throw error;
    }
-    mapping = agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt });
+    mapping = agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt, params });
    break;
  }
  traceStore.append(traceId, {
@@ -1302,7 +1302,8 @@ function resolveAgentRunBackendProfile(env = process.env, params = {}) {
  return AGENTRUN_BACKENDS.includes(mapped) ? mapped : "deepseek";
 }

-function agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt }) {
+function agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt, params = {} }) {
+  const threadReused = Boolean(safeOpaqueId(params.threadId));
  return {
    adapter: ADAPTER_ID,
    managerUrl,
@@ -1324,6 +1325,9 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner
    sessionId: run?.sessionRef?.sessionId ?? null,
    conversationId: run?.sessionRef?.conversationId ?? null,
    threadId: run?.sessionRef?.threadId ?? null,
+    runnerReused: false,
+    threadReused,
+    persistentResume: threadReused,
    reused: false,
    reuseEligible: true,
    createdAt: startedAt,
@@ -1333,6 +1337,7 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner
 }

 function agentRunReusedMapping({ previous = {}, run = {}, command = {}, runnerJob = null, traceId, startedAt, backendProfile, managerUrl, env }) {
+  const threadReused = Boolean(safeOpaqueId(run?.sessionRef?.threadId ?? previous.threadId));
  return {
    ...previous,
    adapter: ADAPTER_ID,
@@ -1356,6 +1361,9 @@ function agentRunReusedMapping({ previous = {}, run = {}, command = {}, runnerJo
    sessionId: run?.sessionRef?.sessionId ?? previous.sessionId ?? null,
    conversationId: run?.sessionRef?.conversationId ?? previous.conversationId ?? null,
    threadId: run?.sessionRef?.threadId ?? previous.threadId ?? null,
+    runnerReused: true,
+    threadReused,
+    persistentResume: threadReused,
    reused: true,
    reuseEligible: true,
    createdAt: previous.createdAt ?? startedAt,
@@ -1416,14 +1424,26 @@ 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));
+}
+
 function agentRunSessionReuseSummary(base, reused, options = {}) {
+  const runnerReused = Boolean(reused);
+  const threadReused = agentRunThreadReused(base, options);
+  const persistentResume = threadReused;
+  const sessionReused = runnerReused || persistentResume;
  return {
    sessionId: base.sessionId ?? base.agentRun?.sessionId ?? null,
    conversationId: base.conversationId ?? base.agentRun?.conversationId ?? null,
    threadId: base.threadId ?? base.agentRun?.threadId ?? null,
-    reused: Boolean(reused),
-    status: options.status ?? (reused ? "reused" : "new"),
-    runnerReused: Boolean(reused),
+    reused: sessionReused,
+    status: options.status ?? (runnerReused ? "reused" : persistentResume ? "thread-resumed" : "new"),
+    runnerReused,
+    threadReused,
+    persistentResume,
    valuesRedacted: true
  };
 }
@@ -584,6 +584,203 @@ test("cloud api /v1/agent/chat delegates v0.2 turns to AgentRun v0.1 over adapte
  }
 });

+test("cloud api AgentRun adapter reports persistent thread resume when a completed run needs a new runner", async () => {
+  const calls = [];
+  const hwlabSessionId = "ses_server-test-thread-resume";
+  const agentRunSessionId = "ses_agentrun_deepseek_server_test_thread_resume";
+  const conversationId = "cnv_server-test-thread-resume";
+  const threadId = "019e90e0-9535-7130-a894-47ef4e127206";
+  const ownerSessions = new Map([[hwlabSessionId, testAgentSessionRecord({
+    sessionId: hwlabSessionId,
+    conversationId,
+    threadId,
+    traceId: "trc_previous_thread_resume",
+    status: "idle",
+    session: {
+      agentRun: {
+        adapter: "agentrun-v01",
+        backendProfile: "deepseek",
+        managerUrl: "http://127.0.0.1:1",
+        runId: "run_issue812_previous",
+        commandId: "cmd_issue812_previous",
+        jobName: "agentrun-v01-runner-issue812-previous",
+        namespace: "agentrun-v01",
+        sessionId: agentRunSessionId,
+        conversationId,
+        threadId,
+        terminalStatus: "completed",
+        runStatus: "completed",
+        reuseEligible: true,
+        reused: false
+      }
+    }
+  })]]);
+
+  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_issue812_resume" })}\n`);
+    };
+    if (request.method === "GET" && url.pathname === "/api/v1/runs/run_issue812_previous") {
+      return send({
+        id: "run_issue812_previous",
+        status: "completed",
+        terminalStatus: "completed",
+        backendProfile: "deepseek",
+        sessionRef: { sessionId: agentRunSessionId, conversationId, threadId, metadata: {} }
+      });
+    }
+    if (request.method === "POST" && url.pathname === "/api/v1/sessions") {
+      assert.equal(body.sessionId, agentRunSessionId);
+      assert.equal(body.backendProfile, "deepseek");
+      return send({ sessionId: body.sessionId, backendProfile: body.backendProfile });
+    }
+    if (request.method === "POST" && url.pathname === "/api/v1/runs") {
+      assert.equal(body.sessionRef.sessionId, agentRunSessionId);
+      assert.equal(body.sessionRef.conversationId, conversationId);
+      assert.equal(body.sessionRef.threadId, threadId);
+      assert.equal(body.sessionRef.metadata.hwlabSessionId, hwlabSessionId);
+      assert.equal(body.sessionRef.metadata.threadContinuityPolicy, "hwlab-agentrun-v01-reuse-runner-thread");
+      return send({ id: "run_issue812_resume", status: "pending", backendProfile: "deepseek", sessionRef: body.sessionRef, resourceBundleRef: body.resourceBundleRef });
+    }
+    if (request.method === "POST" && url.pathname === "/api/v1/runs/run_issue812_resume/commands") {
+      assert.equal(body.type, "turn");
+      assert.equal(body.payload.sessionId, agentRunSessionId);
+      assert.equal(body.payload.hwlabSessionId, hwlabSessionId);
+      assert.equal(body.payload.threadId, threadId);
+      assert.match(body.payload.prompt, /ISSUE812_RESUME/u);
+      return send({ id: "cmd_issue812_resume", runId: "run_issue812_resume", state: "pending", type: "turn", seq: 1 });
+    }
+    if (request.method === "POST" && url.pathname === "/api/v1/runs/run_issue812_resume/runner-jobs") {
+      assert.equal(body.commandId, "cmd_issue812_resume");
+      return send({
+        action: "create-kubernetes-job",
+        runId: "run_issue812_resume",
+        commandId: "cmd_issue812_resume",
+        attemptId: "attempt_issue812_resume",
+        runnerId: "runner_issue812_resume",
+        namespace: "agentrun-v01",
+        jobName: "agentrun-v01-runner-issue812-resume",
+        jobIdentity: { namespace: "agentrun-v01", name: "agentrun-v01-runner-issue812-resume" },
+        runner: { attemptId: "attempt_issue812_resume", runnerId: "runner_issue812_resume" }
+      });
+    }
+    if (request.method === "GET" && url.pathname === "/api/v1/runs/run_issue812_resume/events") {
+      return send({ items: [
+        { id: "evt_issue812_1", runId: "run_issue812_resume", seq: 1, type: "backend_status", payload: { phase: "runner-job-created", commandId: "cmd_issue812_resume", attemptId: "attempt_issue812_resume", jobName: "agentrun-v01-runner-issue812-resume", namespace: "agentrun-v01" }, createdAt: "2026-06-04T00:00:00.000Z" },
+        { id: "evt_issue812_2", runId: "run_issue812_resume", seq: 2, type: "backend_status", payload: { phase: "initial-prompt-assembly", commandId: "cmd_issue812_resume", initialPromptInjected: false, reason: "thread-resume" }, createdAt: "2026-06-04T00:00:00.500Z" },
+        { id: "evt_issue812_3", runId: "run_issue812_resume", seq: 3, type: "assistant_message", payload: { commandId: "cmd_issue812_resume", text: "ISSUE812_RESUME_OK" }, createdAt: "2026-06-04T00:00:01.000Z" },
+        { id: "evt_issue812_4", runId: "run_issue812_resume", seq: 4, type: "terminal_status", payload: { commandId: "cmd_issue812_resume", terminalStatus: "completed" }, createdAt: "2026-06-04T00:00:02.000Z" }
+      ] });
+    }
+    if (request.method === "GET" && url.pathname === "/api/v1/runs/run_issue812_resume/commands/cmd_issue812_resume/result") {
+      return send({
+        runId: "run_issue812_resume",
+        commandId: "cmd_issue812_resume",
+        attemptId: "attempt_issue812_resume",
+        runnerId: "runner_issue812_resume",
+        jobName: "agentrun-v01-runner-issue812-resume",
+        namespace: "agentrun-v01",
+        status: "completed",
+        runStatus: "completed",
+        commandState: "completed",
+        terminalStatus: "completed",
+        completed: true,
+        reply: "ISSUE812_RESUME_OK",
+        lastSeq: 4,
+        eventCount: 4,
+        sessionRef: { sessionId: agentRunSessionId, conversationId, threadId }
+      });
+    }
+    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_issue812_resume" })}\n`);
+  });
+  await new Promise((resolve) => agentRunServer.listen(0, "127.0.0.1", resolve));
+
+  const agentRunPort = agentRunServer.address().port;
+  for (const record of ownerSessions.values()) {
+    record.session.agentRun.managerUrl = `http://127.0.0.1:${agentRunPort}`;
+  }
+  const traceStore = createCodeAgentTraceStore();
+  const server = createCloudApiServer({
+    traceStore,
+    env: {
+      HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01",
+      AGENTRUN_MGR_URL: `http://127.0.0.1:${agentRunPort}`,
+      HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1",
+      HWLAB_CODE_AGENT_AGENTRUN_SOURCE_COMMIT: "0123456789abcdef0123456789abcdef01234567",
+      HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE: "deepseek",
+      HWLAB_ENVIRONMENT: "v02",
+      HWLAB_GITOPS_PROFILE: "v02"
+    },
+    accessController: {
+      required: false,
+      async authenticate() {
+        return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION };
+      },
+      async createCodeAgentDevicePodApiKey() {
+        return { apiKey: "test-device-pod-api-key" };
+      },
+      async recordAgentSessionOwner(input) {
+        const record = testAgentSessionRecord(input);
+        ownerSessions.set(record.id, record);
+        return record;
+      },
+      async getAgentSession(sessionId) {
+        return ownerSessions.get(sessionId) ?? null;
+      }
+    }
+  });
+  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
+
+  try {
+    const { port } = server.address();
+    const traceId = "trc_server-test-issue812-resume";
+    const submit = await fetch(`http://127.0.0.1:${port}/v1/agent/chat`, {
+      method: "POST",
+      headers: { "content-type": "application/json", "x-trace-id": traceId, cookie: "hwlab_session=test-stub-session" },
+      body: JSON.stringify({
+        conversationId,
+        sessionId: hwlabSessionId,
+        ownerUserId: "usr_agent_owner",
+        ownerRole: "user",
+        threadId,
+        message: "ISSUE812_RESUME after completed AgentRun run"
+      })
+    });
+    assert.equal(submit.status, 202);
+
+    const payload = await pollAgentResult(port, traceId);
+    validateCodeAgentChatSchema(payload);
+    assert.equal(payload.status, "completed");
+    assert.equal(payload.sessionId, hwlabSessionId);
+    assert.equal(payload.threadId, threadId);
+    assert.equal(payload.agentRun.runId, "run_issue812_resume");
+    assert.equal(payload.agentRun.reused, false);
+    assert.equal(payload.agentRun.runnerReused, false);
+    assert.equal(payload.agentRun.threadReused, true);
+    assert.equal(payload.agentRun.persistentResume, true);
+    assert.equal(payload.sessionReuse.threadId, threadId);
+    assert.equal(payload.sessionReuse.reused, true);
+    assert.equal(payload.sessionReuse.status, "thread-resumed");
+    assert.equal(payload.sessionReuse.runnerReused, false);
+    assert.equal(payload.sessionReuse.threadReused, true);
+    assert.equal(payload.sessionReuse.persistentResume, true);
+    assert.equal(payload.reply.content, "ISSUE812_RESUME_OK");
+    assert.equal(calls.some((call) => call.method === "GET" && call.path === "/api/v1/runs/run_issue812_previous"), true);
+    assert.equal(calls.filter((call) => call.method === "POST" && call.path === "/api/v1/runs").length, 1);
+    assert.equal(calls.filter((call) => call.method === "POST" && call.path === "/api/v1/runs/run_issue812_resume/runner-jobs").length, 1);
+  } 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())));
+  }
+});
+
 test("cloud api AgentRun adapter exposes invalid tool-call attribution in result payload", async () => {
  const agentRunServer = createHttpServer(async (request, response) => {
    const url = new URL(request.url || "/", "http://127.0.0.1");