From 40b78c5cae94168e1ab5d266b52d43036d50cb85 Mon Sep 17 00:00:00 2001 From: kris Date: Sat, 4 Apr 2026 13:04:12 +0800 Subject: [PATCH] fix: fail closed on invalid codex resume bindings --- local-agent/codex-task-runner.mjs | 182 +++++++++++++++++- local-agent/server.mjs | 8 +- tests/local-agent-codex-task-runner.test.mjs | 186 ++++++++++++++++++- 3 files changed, 367 insertions(+), 9 deletions(-) diff --git a/local-agent/codex-task-runner.mjs b/local-agent/codex-task-runner.mjs index 8eb3b98..5406eb7 100644 --- 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, }; } diff --git a/local-agent/server.mjs b/local-agent/server.mjs index 506f3c6..6d155b6 100755 --- 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, diff --git a/tests/local-agent-codex-task-runner.test.mjs b/tests/local-agent-codex-task-runner.test.mjs index ae0e556..8e21cb2 100644 --- 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/); +});