fix: serialize codex submit bursts

scripts/src/code-queue.ts

@@ -1,5 +1,5 @@
-import { readFileSync } from "node:fs";
-import { type UniDeskConfig } from "./config";
+import { mkdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs";
+import { type UniDeskConfig, repoRoot, rootPath } from "./config";
 import { coreInternalFetch } from "./microservices";
 
 const defaultToolLimit = 8;
@@ -9,6 +9,10 @@ const defaultOutputLimit = 20;
 const defaultTextPreviewChars = 12_000;
 const defaultTasksLimit = 20;
 const maxTasksLimit = 100;
+const submitLockWaitMs = 60_000;
+const submitLockPollMs = 250;
+const submitLockStaleMs = 120_000;
+const submitThrottleMs = nonNegativeIntegerEnv("UNIDESK_CODEX_SUBMIT_THROTTLE_MS", 2000);
 
 interface CodexTaskOptions {
   trace: boolean;
@@ -100,6 +104,84 @@ function codeQueueProxyPath(path: string): string {
   return `${codeQueueProxyPrefix}${path}`;
 }
 
+function nonNegativeIntegerEnv(name: string, fallback: number): number {
+  const raw = process.env[name];
+  if (raw === undefined || raw.trim().length === 0) return fallback;
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed) || parsed < 0) return fallback;
+  return Math.floor(parsed);
+}
+
+function sleepSync(ms: number): void {
+  if (ms <= 0) return;
+  const buffer = new SharedArrayBuffer(4);
+  Atomics.wait(new Int32Array(buffer), 0, 0, ms);
+}
+
+function acquireSubmitLock(): { lockPath: string; acquiredAfterMs: number; release: () => void } {
+  const lockRoot = rootPath(".state", "locks");
+  const lockPath = rootPath(".state", "locks", "codex-submit.lock.d");
+  mkdirSync(lockRoot, { recursive: true });
+  const startedAt = Date.now();
+  while (Date.now() - startedAt <= submitLockWaitMs) {
+    try {
+      mkdirSync(lockPath);
+      writeFileSync(rootPath(".state", "locks", "codex-submit.lock.d", "owner.json"), `${JSON.stringify({
+        pid: process.pid,
+        createdAt: new Date().toISOString(),
+        repoRoot,
+      }, null, 2)}\n`, "utf8");
+      let released = false;
+      return {
+        lockPath,
+        acquiredAfterMs: Date.now() - startedAt,
+        release: () => {
+          if (released) return;
+          released = true;
+          rmSync(lockPath, { recursive: true, force: true });
+        },
+      };
+    } catch (error) {
+      const code = (error as { code?: string }).code;
+      if (code !== "EEXIST") throw error;
+      let stale = false;
+      try {
+        stale = Date.now() - statSync(lockPath).mtimeMs > submitLockStaleMs;
+      } catch {
+        stale = true;
+      }
+      if (stale) {
+        rmSync(lockPath, { recursive: true, force: true });
+        continue;
+      }
+      sleepSync(submitLockPollMs);
+    }
+  }
+  throw new Error(`codex submit lock timed out after ${submitLockWaitMs}ms: ${lockPath}`);
+}
+
+function runWithSubmitLock<T>(operation: () => T): { result: T; lock: Record<string, unknown> } {
+  const lock = acquireSubmitLock();
+  const operationStartedAt = Date.now();
+  try {
+    sleepSync(submitThrottleMs);
+    const result = operation();
+    return {
+      result,
+      lock: {
+        path: lock.lockPath,
+        acquiredAfterMs: lock.acquiredAfterMs,
+        heldMs: Date.now() - operationStartedAt,
+        throttleMs: submitThrottleMs,
+        staleMs: submitLockStaleMs,
+        mode: "local-atomic-directory-submit-serialization",
+      },
+    };
+  } finally {
+    lock.release();
+  }
+}
+
 function requireTaskId(value: string | undefined, command: string): string {
   if (value === undefined || value.trim().length === 0) throw new Error(`${command} requires task id`);
   return value.trim();
@@ -1235,11 +1317,13 @@ function codexSubmitTask(args: string[]): unknown {
       },
     };
   }
-  const response = unwrapCodexResponse(coreInternalFetch(codeQueueProxyPath("/api/tasks"), { method: "POST", body: payload }));
+  const locked = runWithSubmitLock(() => unwrapCodexResponse(coreInternalFetch(codeQueueProxyPath("/api/tasks"), { method: "POST", body: payload })));
+  const response = locked.result;
   return {
     upstream: response.upstream,
     tasks: asArray(response.body.tasks).map((task) => compactTaskMutationResponse(task, { fullPrompt: true })),
     queue: compactQueueMutationSummary(response.body.queue),
+    submitConcurrencyGuard: locked.lock,
   };
 }