fix: fail closed on invalid codex resume bindings

2026-04-04 13:04:12 +08:00
parent 4ab0414d43
commit 40b78c5cae
3 changed files with 367 additions and 9 deletions
--- a/local-agent/codex-task-runner.mjs
+++ b/local-agent/codex-task-runner.mjs
@@ -1,9 +1,179 @@
+import { constants } from "node:fs";
+import { access, stat } from "node:fs/promises";
+import { DatabaseSync } from "node:sqlite";
+import { resolve } from "node:path";
+
+function trimToDefined(value) {
+  const trimmed = String(value ?? "").trim();
+  return trimmed ? trimmed : undefined;
+}
+
+function resolveResumeTarget(config, task) {
+  const targetThreadRef = trimToDefined(task?.targetCodexThreadRef || task?.targetThreadId);
+  const targetFolderRef = trimToDefined(
+    task?.targetCodexFolderRef || config.masterAgentWorkdir || process.cwd(),
+  );
+  const cwd = resolve(targetFolderRef || process.cwd());
+  return {
+    targetThreadRef,
+    targetFolderRef: targetFolderRef || process.cwd(),
+    cwd,
+  };
+}
+
+function shouldPreflightResumeTask(task) {
+  const taskType = String(task?.taskType || "").trim();
+  if (taskType === "dispatch_execution") {
+    return true;
+  }
+  if (taskType !== "conversation_reply") {
+    return false;
+  }
+
+  return Boolean(
+    trimToDefined(task?.targetCodexThreadRef || task?.targetThreadId) ||
+      trimToDefined(task?.targetCodexFolderRef) ||
+      trimToDefined(task?.targetProjectId) ||
+      trimToDefined(task?.targetThreadDisplayName),
+  );
+}
+
+function buildStructuredTaskBindingError(code, message, details) {
+  return {
+    code,
+    message,
+    details,
+  };
+}
+
+function inspectCodexThreadBinding(config, targetThreadRef, targetFolderRef) {
+  const stateDbPath = trimToDefined(config?.codexStateDbPath);
+  if (!stateDbPath) {
+    return {
+      status: "skipped",
+    };
+  }
+
+  try {
+    const db = new DatabaseSync(stateDbPath, { readonly: true });
+    try {
+      const row = db
+        .prepare("SELECT id, cwd, archived FROM threads WHERE id = ? LIMIT 1")
+        .get(targetThreadRef);
+      if (!row || row.archived) {
+        return {
+          status: "missing",
+        };
+      }
+
+      const threadCwd = trimToDefined(row.cwd) || "";
+      if (threadCwd && resolve(threadCwd) !== resolve(targetFolderRef)) {
+        return {
+          status: "mismatch",
+          threadCwd,
+        };
+      }
+
+      return {
+        status: "ok",
+        threadCwd,
+      };
+    } finally {
+      db.close();
+    }
+  } catch {
+    return {
+      status: "unavailable",
+    };
+  }
+}
+
+export async function prepareCodexTaskExecution(config, task, outputFile) {
+  if (!shouldPreflightResumeTask(task)) {
+    return {
+      ok: true,
+      execution: buildCodexTaskExecution(config, task, outputFile),
+    };
+  }
+
+  const targetThreadRef = trimToDefined(task?.targetCodexThreadRef || task?.targetThreadId);
+  if (!targetThreadRef) {
+    return {
+      ok: false,
+      error: buildStructuredTaskBindingError(
+        "LOCAL_AGENT_CODEX_THREAD_BINDING_MISSING",
+        "LOCAL_AGENT_CODEX_THREAD_BINDING_MISSING: 目标线程绑定缺失，已拒绝 codex exec resume。",
+        {
+          taskType: String(task?.taskType || "").trim() || undefined,
+          targetProjectId: trimToDefined(task?.targetProjectId),
+          targetCodexFolderRef: trimToDefined(task?.targetCodexFolderRef),
+          targetThreadDisplayName: trimToDefined(task?.targetThreadDisplayName),
+        },
+      ),
+    };
+  }
+
+  const resumeTarget = resolveResumeTarget(config, task);
+  const bindingInspection = inspectCodexThreadBinding(config, targetThreadRef, resumeTarget.cwd);
+  if (bindingInspection.status === "missing") {
+    return {
+      ok: false,
+      error: buildStructuredTaskBindingError(
+        "LOCAL_AGENT_CODEX_THREAD_BINDING_STALE",
+        "LOCAL_AGENT_CODEX_THREAD_BINDING_STALE: 目标线程绑定在 Codex 状态库中不存在或已归档，已拒绝 codex exec resume。",
+        {
+          targetThreadRef,
+          targetCodexFolderRef: resumeTarget.targetFolderRef,
+        },
+      ),
+    };
+  }
+
+  if (bindingInspection.status === "mismatch") {
+    return {
+      ok: false,
+      error: buildStructuredTaskBindingError(
+        "LOCAL_AGENT_CODEX_THREAD_BINDING_MISMATCH",
+        `LOCAL_AGENT_CODEX_THREAD_BINDING_MISMATCH: 目标线程绑定的 cwd 与当前目录不一致，已拒绝 codex exec resume。cwd=${resumeTarget.cwd} liveCwd=${bindingInspection.threadCwd}`,
+        {
+          cwd: resumeTarget.cwd,
+          liveCwd: bindingInspection.threadCwd,
+          targetThreadRef,
+          targetCodexFolderRef: resumeTarget.targetFolderRef,
+        },
+      ),
+    };
+  }
+
+  try {
+    const folderStat = await stat(resumeTarget.cwd);
+    if (!folderStat.isDirectory()) {
+      throw new Error("NOT_A_DIRECTORY");
+    }
+    await access(resumeTarget.cwd, constants.R_OK | constants.X_OK);
+  } catch {
+    return {
+      ok: false,
+      error: buildStructuredTaskBindingError(
+        "LOCAL_AGENT_CODEX_WORKDIR_INVALID",
+        `LOCAL_AGENT_CODEX_WORKDIR_INVALID: 线程工作目录不可访问，已拒绝 codex exec resume。cwd=${resumeTarget.cwd}`,
+        {
+          cwd: resumeTarget.cwd,
+          targetThreadRef,
+          targetCodexFolderRef: resumeTarget.targetFolderRef,
+        },
+      ),
+    };
+  }
+
+  return {
+    ok: true,
+    execution: buildCodexTaskExecution(config, task, outputFile),
+  };
+}
+
 export function buildCodexTaskExecution(config, task, outputFile) {
-  const targetThreadRef =
-    String(task?.targetCodexThreadRef || task?.targetThreadId || "").trim();
-  const targetFolderRef =
-    String(task?.targetCodexFolderRef || config.masterAgentWorkdir || process.cwd()).trim() ||
-    process.cwd();
+  const { targetThreadRef, cwd } = resolveResumeTarget(config, task);
  const prompt = String(task?.executionPrompt || "");

  if (
@@ -23,7 +193,7 @@ export function buildCodexTaskExecution(config, task, outputFile) {
    args.push(targetThreadRef, prompt);
    return {
      mode: "resume",
-      cwd: targetFolderRef,
+      cwd,
      args,
    };
  }
--- a/local-agent/server.mjs
+++ b/local-agent/server.mjs
@@ -6,7 +6,7 @@ import { access, readFile, readdir, rm } from "node:fs/promises";
 import os from "node:os";
 import { join, resolve } from "node:path";
 import { discoverCodexProjectCandidatesInWorker } from "./codex-session-discovery.mjs";
-import { buildCodexTaskExecution } from "./codex-task-runner.mjs";
+import { prepareCodexTaskExecution } from "./codex-task-runner.mjs";
 import {
  executeOmxTeamTask,
  getOmxTeamTaskRunnerConfig,
@@ -383,7 +383,11 @@ async function runMasterAgentTask(config, runtime, task) {
        replyBody: omxResult.replyBody,
      };
    } else {
-      const codexExecution = buildCodexTaskExecution(config, task, outputFile);
+      const codexPreparation = await prepareCodexTaskExecution(config, task, outputFile);
+      if (!codexPreparation.ok) {
+        throw new Error(codexPreparation.error.message);
+      }
+      const codexExecution = codexPreparation.execution;
      await new Promise((resolveTask, rejectTask) => {
        const child = spawn("codex", codexExecution.args, {
          cwd: codexExecution.cwd,
--- a/tests/local-agent-codex-task-runner.test.mjs
+++ b/tests/local-agent-codex-task-runner.test.mjs
@@ -1,7 +1,83 @@
 import test from "node:test";
 import assert from "node:assert/strict";
+import os from "node:os";
+import path from "node:path";
+import { mkdtemp, mkdir, rm } from "node:fs/promises";
+import { DatabaseSync } from "node:sqlite";

-import { buildCodexTaskExecution } from "../local-agent/codex-task-runner.mjs";
+import {
+  buildCodexTaskExecution,
+  prepareCodexTaskExecution,
+} from "../local-agent/codex-task-runner.mjs";
+
+let runtimeRoot = "";
+
+async function ensureRuntimeRoot() {
+  if (!runtimeRoot) {
+    runtimeRoot = await mkdtemp(path.join(os.tmpdir(), "boss-local-agent-task-runner-"));
+  }
+  return runtimeRoot;
+}
+
+test.after(async () => {
+  if (runtimeRoot) {
+    await rm(runtimeRoot, { recursive: true, force: true });
+  }
+});
+
+async function createCodexStateDb(threads) {
+  const root = await ensureRuntimeRoot();
+  const dbPath = path.join(root, `state-${Math.random().toString(16).slice(2)}.sqlite`);
+  const db = new DatabaseSync(dbPath);
+  db.exec(`
+    CREATE TABLE threads (
+      id TEXT PRIMARY KEY,
+      rollout_path TEXT NOT NULL,
+      created_at INTEGER NOT NULL,
+      updated_at INTEGER NOT NULL,
+      source TEXT NOT NULL,
+      model_provider TEXT NOT NULL,
+      cwd TEXT NOT NULL,
+      title TEXT NOT NULL,
+      sandbox_policy TEXT NOT NULL,
+      approval_mode TEXT NOT NULL,
+      tokens_used INTEGER NOT NULL DEFAULT 0,
+      has_user_event INTEGER NOT NULL DEFAULT 0,
+      archived INTEGER NOT NULL DEFAULT 0,
+      archived_at INTEGER,
+      git_sha TEXT,
+      git_branch TEXT,
+      git_origin_url TEXT,
+      cli_version TEXT NOT NULL DEFAULT '',
+      first_user_message TEXT NOT NULL DEFAULT '',
+      agent_nickname TEXT,
+      agent_role TEXT,
+      memory_mode TEXT NOT NULL DEFAULT 'enabled',
+      model TEXT,
+      reasoning_effort TEXT,
+      agent_path TEXT
+    );
+  `);
+  const insertThread = db.prepare(`
+    INSERT INTO threads (
+      id, rollout_path, created_at, updated_at, source, model_provider, cwd, title,
+      sandbox_policy, approval_mode, tokens_used, has_user_event, archived,
+      cli_version, first_user_message, agent_nickname, agent_role, memory_mode, model, reasoning_effort
+    ) VALUES (?, ?, ?, ?, 'desktop', 'openai', ?, ?, 'workspace-write', 'never', 0, 1, 0, '0.118.0', '', '', '', 'enabled', 'gpt-5.4', 'medium')
+  `);
+  for (const thread of threads) {
+    insertThread.run(
+      thread.id,
+      thread.rolloutPath ?? path.join(root, `${thread.id}.jsonl`),
+      1774845600,
+      1774845618,
+      thread.cwd,
+      thread.title ?? thread.id,
+    );
+  }
+  db.close();
+  return dbPath;
+}

 test("conversation reply resumes the real Codex thread when thread ref is available", () => {
  const execution = buildCodexTaskExecution(
@@ -91,3 +167,111 @@ test("master agent reply without target thread stays on ephemeral exec", () => {
    "你是主 Agent",
  ]);
 });
+
+test("conversation reply preflight fails closed when target cwd is missing", async () => {
+  const missingFolder = "/tmp/boss-local-agent-missing-workdir";
+  const result = await prepareCodexTaskExecution(
+    {
+      masterAgentWorkdir: "/Users/kris/code/boss",
+      masterAgentSandbox: "workspace-write",
+    },
+    {
+      taskType: "conversation_reply",
+      executionPrompt: "请回复用户",
+      targetCodexThreadRef: "019d-thread-real",
+      targetCodexFolderRef: missingFolder,
+    },
+    "/tmp/reply.txt",
+  );
+
+  assert.equal(result.ok, false);
+  assert.equal(result.error.code, "LOCAL_AGENT_CODEX_WORKDIR_INVALID");
+  assert.match(result.error.message, /LOCAL_AGENT_CODEX_WORKDIR_INVALID/);
+  assert.match(result.error.message, /missing-workdir/);
+});
+
+test("dispatch execution preflight fails closed when target thread ref is missing", async () => {
+  const result = await prepareCodexTaskExecution(
+    {
+      masterAgentWorkdir: "/Users/kris/code/boss",
+      masterAgentSandbox: "workspace-write",
+    },
+    {
+      taskType: "dispatch_execution",
+      executionPrompt: "请执行群聊任务",
+      targetCodexFolderRef: "/Users/kris/code/boss",
+    },
+    "/tmp/reply.txt",
+  );
+
+  assert.equal(result.ok, false);
+  assert.equal(result.error.code, "LOCAL_AGENT_CODEX_THREAD_BINDING_MISSING");
+  assert.match(result.error.message, /LOCAL_AGENT_CODEX_THREAD_BINDING_MISSING/);
+});
+
+test("conversation reply preflight fails closed when target thread ref is stale in local Codex state", async () => {
+  const root = await ensureRuntimeRoot();
+  const validCwd = path.join(root, "project-stale");
+  await mkdir(validCwd, { recursive: true });
+  const stateDbPath = await createCodexStateDb([
+    {
+      id: "019d-thread-other",
+      cwd: validCwd,
+      title: "Other thread",
+    },
+  ]);
+
+  const result = await prepareCodexTaskExecution(
+    {
+      masterAgentWorkdir: "/Users/kris/code/boss",
+      masterAgentSandbox: "workspace-write",
+      codexStateDbPath: stateDbPath,
+    },
+    {
+      taskType: "conversation_reply",
+      executionPrompt: "请回复用户",
+      targetCodexThreadRef: "019d-thread-stale",
+      targetCodexFolderRef: validCwd,
+    },
+    "/tmp/reply.txt",
+  );
+
+  assert.equal(result.ok, false);
+  assert.equal(result.error.code, "LOCAL_AGENT_CODEX_THREAD_BINDING_STALE");
+  assert.match(result.error.message, /LOCAL_AGENT_CODEX_THREAD_BINDING_STALE/);
+});
+
+test("conversation reply preflight fails closed when target cwd mismatches the live Codex thread binding", async () => {
+  const root = await ensureRuntimeRoot();
+  const liveCwd = path.join(root, "project-live");
+  const staleCwd = path.join(root, "project-stale-mismatch");
+  await mkdir(liveCwd, { recursive: true });
+  await mkdir(staleCwd, { recursive: true });
+  const stateDbPath = await createCodexStateDb([
+    {
+      id: "019d-thread-live",
+      cwd: liveCwd,
+      title: "Live thread",
+    },
+  ]);
+
+  const result = await prepareCodexTaskExecution(
+    {
+      masterAgentWorkdir: "/Users/kris/code/boss",
+      masterAgentSandbox: "workspace-write",
+      codexStateDbPath: stateDbPath,
+    },
+    {
+      taskType: "conversation_reply",
+      executionPrompt: "请回复用户",
+      targetCodexThreadRef: "019d-thread-live",
+      targetCodexFolderRef: staleCwd,
+    },
+    "/tmp/reply.txt",
+  );
+
+  assert.equal(result.ok, false);
+  assert.equal(result.error.code, "LOCAL_AGENT_CODEX_THREAD_BINDING_MISMATCH");
+  assert.match(result.error.message, /LOCAL_AGENT_CODEX_THREAD_BINDING_MISMATCH/);
+  assert.match(result.error.message, /project-live/);
+});