fix(codex-bridge): exit when another writer owns the thread (#906 link 1)

scripts/drivers/types/codex/codex-bridge.js

@@ -442,7 +442,13 @@ class AppServerClient {
       if (!pending) return;
       this.pending.delete(message.id);
       if (message.error) {
-        pending.reject(new Error(message.error.message || JSON.stringify(message.error)));
+        const rpcError = new Error(message.error.message || JSON.stringify(message.error));
+        // Carry the JSON-RPC error code through, not just its text. ensureThread
+        // decides on the message ("already has an active writer"), so the code is
+        // not what gates that today; it is kept for diagnostics and any future
+        // caller that wants the numeric reason without parsing the text (#906).
+        if (typeof message.error.code === "number") rpcError.code = message.error.code;
+        pending.reject(rpcError);
       } else {
         pending.resolve(message.result);
       }
@@ -784,7 +790,13 @@ class WebSocketAppServerClient {
       if (!pending) return;
       this.pending.delete(message.id);
       if (message.error) {
-        pending.reject(new Error(message.error.message || JSON.stringify(message.error)));
+        const rpcError = new Error(message.error.message || JSON.stringify(message.error));
+        // Carry the JSON-RPC error code through, not just its text. ensureThread
+        // decides on the message ("already has an active writer"), so the code is
+        // not what gates that today; it is kept for diagnostics and any future
+        // caller that wants the numeric reason without parsing the text (#906).
+        if (typeof message.error.code === "number") rpcError.code = message.error.code;
+        pending.reject(rpcError);
       } else {
         pending.resolve(message.result);
       }
@@ -1121,6 +1133,23 @@ class CodexBridge {
         // below is a distinct failure (a resume that succeeded but returned
         // the wrong thread) and should still die() as before, not be
         // silently swallowed by this fallback.
+        // Two failures reach this catch and they need opposite handling. The
+        // benign one below -- a Codex 0.142+ --remote session that never created
+        // a rollout -- is what it was written for: turn/start needs only the
+        // threadId, so the bridge stays alive idle.
+        //
+        // "already has an active writer" is the other, and it is deterministic:
+        // another writer -- a co-resident Codex Desktop, or a second bridge --
+        // owns this thread, and resume cannot succeed while that holds. A bridge
+        // that proceeds anyway still arms watchers and holds ~10 threads, so
+        // duplicates accumulate until a per-user pid limit is saturated (#906).
+        // Match the message, not the JSON-RPC code alone (-32600 is the generic
+        // "invalid request", carried here now for diagnostics): the message is
+        // the condition, and a rewording that kept the code would not be this.
+        // Exit non-zero so a bridge that cannot own its thread does not linger.
+        if (/already has an active writer/iu.test(err && err.message ? err.message : "")) {
+          die(`thread/resume failed: ${err.message}`);
+        }
         console.error(`codex-bridge: thread/resume failed (${err.message}); proceeding without resume`);
         this.threadIdle = true;
         this.turnActive = false;

tests/test_codex_bridge.bats

@@ -1892,3 +1892,105 @@ EOF
   [ "$arms" -ge 1 ]
   [ "$arms" -le 20 ]
 }
+@test "codex-bridge: a thread owned by another writer is fatal, not proceed-without-resume (#906)" {
+  run node -e 'const net = require("net"); const crypto = require("crypto"); if (!net || !crypto) process.exit(1);'
+  if [ "$status" -ne 0 ]; then
+    skip "node net/crypto modules are not available in this sandbox"
+  fi
+  run node -e 'const fs = require("fs"); const net = require("net"); const sock = process.argv[1]; try { fs.unlinkSync(sock); } catch (_) {} const server = net.createServer(); server.on("error", () => process.exit(2)); server.listen(sock, () => server.close(() => { try { fs.unlinkSync(sock); } catch (_) {} process.exit(0); }));' "$TEST_SKILL_DIR/probe2.sock"
+  if [ "$status" -ne 0 ]; then
+    skip "unix socket listen is not available in this sandbox"
+  fi
+
+  local fake="$TEST_SKILL_DIR/fake-writer-owned.js"
+  local sock="$TEST_SKILL_DIR/fake-writer-owned.sock"
+  local log="$TEST_SKILL_DIR/fake-writer-owned.log"
+  # A minimal WS app-server that answers thread/resume with the deterministic
+  # "already has an active writer" JSON-RPC error. process/spawn is deliberately
+  # NOT handled: a correct bridge dies before it ever arms a watcher.
+  cat >"$fake" <<'EOF'
+const crypto = require("crypto");
+const fs = require("fs");
+const net = require("net");
+const sock = process.argv[2];
+const log = process.argv[3];
+try { fs.unlinkSync(sock); } catch (_) {}
+function sendFrame(socket, value) {
+  const payload = Buffer.from(JSON.stringify(value), "utf8");
+  let header;
+  if (payload.length < 126) { header = Buffer.from([0x81, payload.length]); }
+  else { header = Buffer.alloc(4); header[0] = 0x81; header[1] = 126; header.writeUInt16BE(payload.length, 2); }
+  socket.write(Buffer.concat([header, payload]));
+}
+function handleMessage(socket, message) {
+  fs.appendFileSync(log, `${message.method}\n`);
+  if (message.method === "initialize") {
+    sendFrame(socket, { jsonrpc: "2.0", id: message.id, result: {} });
+  } else if (message.method === "thread/resume") {
+    sendFrame(socket, {
+      jsonrpc: "2.0",
+      id: message.id,
+      error: { code: -32600, message: `thread ${message.params.threadId} already has an active writer (code -32600)` },
+    });
+  }
+}
+function parseFrames(socket, state, chunk) {
+  state.buffer = Buffer.concat([state.buffer, chunk]);
+  while (state.buffer.length >= 2) {
+    const opcode = state.buffer[0] & 0x0f;
+    let length = state.buffer[1] & 0x7f;
+    let offset = 2;
+    if (length === 126) { if (state.buffer.length < offset + 2) return; length = state.buffer.readUInt16BE(offset); offset += 2; }
+    else if (length === 127) { if (state.buffer.length < offset + 8) return; length = state.buffer.readUInt32BE(offset + 4); offset += 8; }
+    const masked = (state.buffer[1] & 0x80) !== 0;
+    const maskOffset = offset;
+    if (masked) offset += 4;
+    if (state.buffer.length < offset + length) return;
+    let payload = state.buffer.slice(offset, offset + length);
+    if (masked) { const mask = state.buffer.slice(maskOffset, maskOffset + 4); payload = Buffer.from(payload.map((b, i) => b ^ mask[i % 4])); }
+    state.buffer = state.buffer.slice(offset + length);
+    if (opcode === 0x1) handleMessage(socket, JSON.parse(payload.toString("utf8")));
+  }
+}
+const server = net.createServer((socket) => {
+  const state = { buffer: Buffer.alloc(0), upgraded: false, header: Buffer.alloc(0) };
+  socket.on("data", (chunk) => {
+    if (!state.upgraded) {
+      state.header = Buffer.concat([state.header, chunk]);
+      const end = state.header.indexOf("\r\n\r\n");
+      if (end === -1) return;
+      const header = state.header.slice(0, end).toString("utf8");
+      const rest = state.header.slice(end + 4);
+      const key = (header.match(/Sec-WebSocket-Key: (.*)\r\n/i) || [])[1].trim();
+      const accept = crypto.createHash("sha1").update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest("base64");
+      socket.write(["HTTP/1.1 101 Switching Protocols", "Upgrade: websocket", "Connection: Upgrade", `Sec-WebSocket-Accept: ${accept}`, "", ""].join("\r\n"));
+      state.upgraded = true;
+      if (rest.length > 0) parseFrames(socket, state, rest);
+      return;
+    }
+    parseFrames(socket, state, chunk);
+  });
+  socket.on("close", () => server.close(() => process.exit(0)));
+});
+server.listen(sock);
+EOF
+
+  node "$fake" "$sock" "$log" 3>&- &
+  local server_pid="$!"
+  for _ in {1..50}; do [ -S "$sock" ] && break; sleep 0.1; done
+
+  run node "$TYPES/codex/codex-bridge.js" \
+    --project "$PROJ" --team team --name alice --thread thread-owned-elsewhere \
+    --app-server "unix://$sock" --timeout 1 --interval 1 --max-wakes 1
+
+  kill "$server_pid" 2>/dev/null || true
+
+  # The bridge exits non-zero, says why, and NEVER armed a watcher: a bridge
+  # that cannot own its thread is exactly what accumulates in #906.
+  [ "$status" -ne 0 ]
+  grep -q "already has an active writer" <<<"$output"
+  [ "$(grep -c "codex-bridge: armed" <<<"$output")" -eq 0 ]
+  [ "$(grep -c "proceeding without resume" <<<"$output")" -eq 0 ]
+  grep -q "thread/resume" "$log"
+  ! grep -q "process/spawn" "$log"
+}