feat: ship enterprise control and desktop governance

local-agent/codex-task-runner.mjs

@@ -1,9 +1,9 @@
 import os from "node:os";
 import { readFileSync } from "node:fs";
 import { constants } from "node:fs";
-import { access, stat } from "node:fs/promises";
+import { access, readFile, readdir, stat } from "node:fs/promises";
 import { DatabaseSync } from "node:sqlite";
-import { resolve } from "node:path";
+import { dirname, resolve } from "node:path";
 
 function trimToDefined(value) {
   const trimmed = String(value ?? "").trim();
@@ -45,6 +45,14 @@ function defaultCodexPath(relativePath) {
   return resolve(os.homedir(), ".codex", relativePath);
 }
 
+function defaultSessionsDirForStateDb(stateDbPath) {
+  const resolvedStateDbPath = trimToDefined(stateDbPath);
+  if (resolvedStateDbPath) {
+    return resolve(dirname(resolve(resolvedStateDbPath)), "sessions");
+  }
+  return defaultCodexPath("sessions");
+}
+
 function loadThreadWorkspaceHints(globalStatePath) {
   try {
     const raw = readFileSync(resolve(globalStatePath), "utf8");
@@ -72,6 +80,29 @@ function shouldPreflightResumeTask(task) {
   );
 }
 
+function buildDesktopMirrorPlan(task, targetThreadRef) {
+  if (task?.taskType !== "conversation_reply") {
+    return { enabled: false };
+  }
+  if (task?.mirrorBossUserMessageToCodexDesktop !== true) {
+    return { enabled: false };
+  }
+
+  const sourceMessageId = trimToDefined(task?.sourceMessageId || task?.requestMessageId);
+  const sourceMessageBody = trimToDefined(task?.sourceMessageBody || task?.requestText);
+  if (!targetThreadRef || !sourceMessageId || !sourceMessageBody) {
+    return { enabled: false };
+  }
+
+  return {
+    enabled: true,
+    targetThreadRef,
+    sourceMessageId,
+    sourceMessageBody,
+    sourceMessageSentAt: trimToDefined(task?.sourceMessageSentAt),
+  };
+}
+
 function buildStructuredTaskBindingError(code, message, details) {
   return {
     code,
@@ -80,7 +111,71 @@ function buildStructuredTaskBindingError(code, message, details) {
   };
 }
 
-function inspectCodexThreadBinding(config, targetThreadRef, targetFolderRef) {
+function parseSessionMetaLine(line) {
+  try {
+    const parsed = JSON.parse(line);
+    if (parsed?.type !== "session_meta" || !parsed?.payload?.id || !parsed?.payload?.cwd) {
+      return null;
+    }
+    return {
+      id: String(parsed.payload.id),
+      cwd: String(parsed.payload.cwd),
+    };
+  } catch {
+    return null;
+  }
+}
+
+async function findSessionThreadBinding(config, targetThreadRef) {
+  const root = trimToDefined(
+    config?.codexSessionsDir || defaultSessionsDirForStateDb(config?.codexStateDbPath),
+  );
+  if (!root) {
+    return {
+      status: "missing",
+    };
+  }
+
+  const stack = [resolve(root)];
+  const suffix = `-${targetThreadRef}.jsonl`;
+  while (stack.length > 0) {
+    const current = stack.pop();
+    if (!current) continue;
+    let entries = [];
+    try {
+      entries = await readdir(current, { withFileTypes: true });
+    } catch {
+      continue;
+    }
+    for (const entry of entries) {
+      const entryPath = resolve(current, entry.name);
+      if (entry.isDirectory()) {
+        stack.push(entryPath);
+        continue;
+      }
+      if (!entry.isFile() || !entry.name.endsWith(suffix)) {
+        continue;
+      }
+      try {
+        const raw = await readFile(entryPath, "utf8");
+        const meta = parseSessionMetaLine(raw.split(/\r?\n/, 1)[0] ?? "");
+        if (meta?.id === targetThreadRef) {
+          return {
+            status: "ok",
+            threadCwd: trimToDefined(meta.cwd) || "",
+          };
+        }
+      } catch {
+        continue;
+      }
+    }
+  }
+  return {
+    status: "missing",
+  };
+}
+
+async function inspectCodexThreadBinding(config, targetThreadRef, targetFolderRef) {
   const stateDbPath = trimToDefined(config?.codexStateDbPath || defaultCodexPath("state_5.sqlite"));
   if (!stateDbPath) {
     return {
@@ -94,7 +189,10 @@ function inspectCodexThreadBinding(config, targetThreadRef, targetFolderRef) {
       const row = db
         .prepare("SELECT id, cwd, archived, sandbox_policy FROM threads WHERE id = ? LIMIT 1")
         .get(targetThreadRef);
-      if (!row || row.archived) {
+      if (!row) {
+        return await findSessionThreadBinding(config, targetThreadRef);
+      }
+      if (row.archived) {
         return {
           status: "missing",
         };
@@ -127,9 +225,7 @@ function inspectCodexThreadBinding(config, targetThreadRef, targetFolderRef) {
       db.close();
     }
   } catch {
-    return {
-      status: "unavailable",
-    };
+    return await findSessionThreadBinding(config, targetThreadRef);
   }
 }
 
@@ -159,7 +255,7 @@ export async function prepareCodexTaskExecution(config, task, outputFile) {
   }
 
   const resumeTarget = resolveResumeTarget(config, task);
-  const bindingInspection = inspectCodexThreadBinding(config, targetThreadRef, resumeTarget.cwd);
+  const bindingInspection = await inspectCodexThreadBinding(config, targetThreadRef, resumeTarget.cwd);
   if (bindingInspection.status === "missing") {
     return {
       ok: false,
@@ -255,6 +351,7 @@ export function buildCodexTaskExecution(config, task, outputFile) {
       mode: "resume",
       cwd,
       args,
+      desktopMirror: buildDesktopMirrorPlan(task, targetThreadRef),
     };
   }
 
@@ -277,5 +374,6 @@ export function buildCodexTaskExecution(config, task, outputFile) {
     mode: "ephemeral",
     cwd: config.masterAgentWorkdir || process.cwd(),
     args,
+    desktopMirror: { enabled: false },
   };
 }