Merge pull request #630 from pikasTech/fix/v02-refresh-stale-thread

fix: recover stale Code Agent browser threads
2026-05-31 17:14:34 +08:00
parent 79f2694f0c 0bfc5c39d7
commit 5d5c17153e
3 changed files with 163 additions and 47 deletions
@@ -510,6 +510,7 @@ registry 与 git mirror/relay 分属不同基础设施边界。registry 保持
 - no-op env-reuse fast lane 必须输出完整证据链：`g14-ci-plan` 全复用、`skipped-runtime-unchanged`、`gitops-commit`/`gitops-push` skipped、`runtime-ready` skipped。
  不能只用 PipelineRun `Completed` 或总耗时证明没有退化。
 - 真实 rollout 的 `runtime-ready` 必须输出 `started`、周期性 `progress` 和最终 `argo-refresh`/`argo-sync-health`/`workload-ready` 类事件；缺少这些事件时应按可见性回归处理。
+- Bun 语法检查和测试入口必须分开：`bun --check <file.ts>` 只用于非测试 TypeScript 源文件语法/转译检查；`*.test.ts`、`*.test.mts` 或其他使用 `test(...)` 的文件必须用 `bun test <file>` 或 repo-owned package script 执行。不得再用 `bun --check` 直接跑 `.test.ts`，该命令会在测试运行器外求值并报 `Cannot use test outside of the test runner`，这不是业务回归证据。

 GitOps branch 已更新、source branch render 通过、PipelineRun 名称存在或 `G14` DEV/PROD health 正常，都不能单独代表 `v0.2` CI/CD 通过。

@@ -1055,9 +1055,10 @@ async function prepareFakeCodexHome(root = null) {
  return codexHome;
 }

-function createFakeAppServerClient({ calls, responses = ["first stdio reply", "second stdio reply"], failedTurn = null, retryingTurn = null } = {}) {
+function createFakeAppServerClient({ calls, responses = ["first stdio reply", "second stdio reply"], failedTurn = null, retryingTurn = null, staleResumeThreads = [] } = {}) {
  let notificationHandler = null;
  let turn = 0;
+  const staleThreads = new Set(staleResumeThreads);
  return {
    async initialize() {
      calls?.push({ method: "initialize" });
@@ -1073,6 +1074,10 @@ function createFakeAppServerClient({ calls, responses = ["first stdio reply", "s
    },
    async resumeThread(args) {
      calls?.push({ method: "thread/resume", args });
+      if (staleThreads.has(args.threadId)) {
+        staleThreads.delete(args.threadId);
+        throw new Error(`no rollout found for thread id ${args.threadId}`);
+      }
      notificationHandler?.({ method: "thread/started", params: { thread: { id: args.threadId } } });
      return { threadId: args.threadId };
    },
@@ -1381,6 +1386,68 @@ test("Codex stdio uses a persisted thread id after manager restart", async () =>
  }
 });

+test("Codex stdio starts a fresh thread when a restored browser thread is stale", async () => {
+  const calls = [];
+  const fakeCodex = await createFakeCodexCommand();
+  const codexHome = await prepareFakeCodexHome();
+  const manager = createCodexStdioSessionManager({
+    idFactory: () => "ses_stale_browser_thread",
+    createRpcClient: async () => createFakeAppServerClient({
+      calls,
+      responses: ["fresh thread response"],
+      staleResumeThreads: ["thread_stale_browser_restore"]
+    })
+  });
+  const env = {
+    PATH: process.env.PATH,
+    OPENAI_API_KEY: "test-openai-key-material",
+    CODEX_HOME: codexHome,
+    HWLAB_CODE_AGENT_PROVIDER: "codex-stdio",
+    HWLAB_CODE_AGENT_MODEL: "gpt-test",
+    HWLAB_CODE_AGENT_CODEX_COMMAND: fakeCodex.command,
+    HWLAB_CODE_AGENT_CODEX_STDIO_ENABLED: "1",
+    HWLAB_CODE_AGENT_CODEX_STDIO_SUPERVISOR: "repo-owned",
+    HWLAB_CODE_AGENT_CODEX_WORKSPACE: process.cwd(),
+    HWLAB_CODE_AGENT_CODEX_SANDBOX: "danger-full-access",
+    HWLAB_CODE_AGENT_OPENAI_BASE_URL: "http://127.0.0.1:49280/responses"
+  };
+
+  try {
+    const payload = await handleCodeAgentChat(
+      {
+        conversationId: "cnv_stale_browser_thread",
+        sessionId: "ses_stale_browser_thread",
+        threadId: "thread_stale_browser_restore",
+        traceId: "trc_stale_browser_thread",
+        message: "刷新浏览器后继续发送"
+      },
+      {
+        now: () => "2026-05-31T01:06:00.000Z",
+        env,
+        codexStdioManager: manager
+      }
+    );
+
+    validateCodeAgentChatSchema(payload);
+    assert.equal(payload.status, "completed");
+    assert.equal(payload.error, undefined);
+    assert.equal(payload.session.sessionId, "ses_stale_browser_thread");
+    assert.equal(payload.session.threadId, "thread_stdio_ready");
+    assert.equal(payload.providerTrace.threadId, "thread_stdio_ready");
+    assert.equal(payload.providerTrace.terminalStatus, "completed");
+    assert.ok(payload.runnerTrace.events.some((event) => event.label === "thread:resume:stale"));
+    assert.deepEqual(calls.filter((call) => call.method !== "initialize").map((call) => call.method), [
+      "thread/resume",
+      "thread/start",
+      "turn/start"
+    ]);
+    assert.equal(calls.find((call) => call.method === "turn/start")?.args.threadId, "thread_stdio_ready");
+  } finally {
+    await rm(fakeCodex.root, { recursive: true, force: true });
+    await rm(codexHome, { recursive: true, force: true });
+  }
+});
+
 test("Codex stdio manager reports concrete blockers without falling back to readonly", async () => {
  const manager = createCodexStdioSessionManager({
    idFactory: () => "ses_stdio_blocked"
@@ -1606,55 +1606,64 @@ async function runAppServerTurn({
        threadId,
        waitingFor: "thread/resume"
      });
-      const resumed = await client.resumeThread({
-        threadId,
-        model,
-        cwd: workspace,
-        sandbox,
-        approvalPolicy: "never"
-      }, effectiveMs);
-      threadId = optionalId(resumed?.threadId) ?? threadId;
-      state.setThreadId(threadId);
-      state.appendTrace({
-        type: "thread",
-        status: "completed",
-        label: "thread:resume:completed",
-        sessionId: session.sessionId,
-        sessionStatus: session.status,
-        turn: session.turn,
-        threadId,
-        waitingFor: "turn/start"
-      });
+      try {
+        const resumed = await client.resumeThread({
+          threadId,
+          model,
+          cwd: workspace,
+          sandbox,
+          approvalPolicy: "never"
+        }, effectiveMs);
+        threadId = optionalId(resumed?.threadId) ?? threadId;
+        state.setThreadId(threadId);
+        state.appendTrace({
+          type: "thread",
+          status: "completed",
+          label: "thread:resume:completed",
+          sessionId: session.sessionId,
+          sessionStatus: session.status,
+          turn: session.turn,
+          threadId,
+          waitingFor: "turn/start"
+        });
+      } catch (error) {
+        if (!isStaleAppServerThreadError(error)) throw error;
+        state.appendTrace({
+          type: "thread",
+          status: "degraded",
+          label: "thread:resume:stale",
+          sessionId: session.sessionId,
+          sessionStatus: session.status,
+          turn: session.turn,
+          threadId,
+          message: "Persisted app-server thread is no longer resumable; starting a fresh thread for this turn.",
+          waitingFor: "thread/start"
+        });
+        session.threadId = null;
+        threadId = null;
+        state.setThreadId(null);
+        const started = await startFreshAppServerThread({
+          client,
+          state,
+          session,
+          workspace,
+          sandbox,
+          model,
+          effectiveMs
+        });
+        threadId = started.threadId;
+      }
    } else {
-      state.appendTrace({
-        type: "thread",
-        status: "started",
-        label: "thread:start:started",
-        sessionId: session.sessionId,
-        sessionStatus: session.status,
-        turn: session.turn,
-        waitingFor: "thread/start"
-      });
-      const started = await client.startThread({
-        model,
-        cwd: workspace,
+      const started = await startFreshAppServerThread({
+        client,
+        state,
+        session,
+        workspace,
        sandbox,
-        approvalPolicy: "never",
-        serviceName: "hwlab-cloud-api"
-      }, effectiveMs);
-      threadId = optionalId(started?.threadId);
-      if (!threadId) throw new Error("thread/start response did not include thread.id");
-      state.setThreadId(threadId);
-      state.appendTrace({
-        type: "thread",
-        status: "completed",
-        label: "thread:start:completed",
-        sessionId: session.sessionId,
-        sessionStatus: session.status,
-        turn: session.turn,
-        threadId,
-        waitingFor: "turn/start"
+        model,
+        effectiveMs
      });
+      threadId = started.threadId;
    }

    state.appendTrace({
@@ -1698,3 +1707,42 @@ async function runAppServerTurn({
    }
  }
 }
+
+async function startFreshAppServerThread({ client, state, session, workspace, sandbox, model, effectiveMs }) {
+  state.appendTrace({
+    type: "thread",
+    status: "started",
+    label: "thread:start:started",
+    sessionId: session.sessionId,
+    sessionStatus: session.status,
+    turn: session.turn,
+    waitingFor: "thread/start"
+  });
+  const started = await client.startThread({
+    model,
+    cwd: workspace,
+    sandbox,
+    approvalPolicy: "never",
+    serviceName: "hwlab-cloud-api"
+  }, effectiveMs);
+  const threadId = optionalId(started?.threadId);
+  if (!threadId) throw new Error("thread/start response did not include thread.id");
+  state.setThreadId(threadId);
+  state.appendTrace({
+    type: "thread",
+    status: "completed",
+    label: "thread:start:completed",
+    sessionId: session.sessionId,
+    sessionStatus: session.status,
+    turn: session.turn,
+    threadId,
+    waitingFor: "turn/start"
+  });
+  return { threadId };
+}
+
+function isStaleAppServerThreadError(error) {
+  const message = String(error?.message ?? "");
+  return /\bno\s+rollout\s+found\s+for\s+thread\s+id\b/iu.test(message) ||
+    /\bthread\b.*\b(?:not\s+found|missing|expired|gone|unknown)\b/iu.test(message);
+}