fix(codex-bridge): make the failure cap reach and bound the re-arm rate (#906 link 3)

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

@@ -19,6 +19,15 @@ const RUN_DIR = path.join(SKILL_DIR, "run");
 // context); honour the same overrides delivery.sh's windows_wrap uses.
 const BASH_BIN = process.env.GIT_BASH || process.env.AGMSG_BASH || "bash";
 
+// A ceiling on how often watch-once may be re-armed, across every re-arm path
+// (a clean deadline, a wake and its turn, an idle transition). watch-once's own
+// deadline paces the healthy case at one arm per --timeout, so this is only ever
+// felt by a degenerate loop: a stream of DISTINCT wakes re-arms with no delay
+// otherwise -- 2094 arms in 56 s measured against the real bridge (#936) -- and
+// every arm forks watch-once's library sourcing, which is the fork pressure the
+// #906 incident saturated a per-user pid limit with. A rate, not a poll cadence.
+const MIN_ARM_INTERVAL_MS = 1000;
+
 function usage() {
   console.log(`Usage: codex-bridge.js --project <path> [--type codex] [--team <team>] [--name <agent>]
 
@@ -924,6 +933,7 @@ class CodexBridge {
     this.staleWakeCount = 0;
     this.watchFailureCount = 0;
     this.watchRearmTimer = null;
+    this.lastArmAt = 0;
     this.inlineInboxText = "";
     this.stopping = false;
     const key = identities.length === 1
@@ -1143,6 +1153,19 @@ class CodexBridge {
   async armWatch() {
     this.clearWatchRearmTimer();
     if (this.stopping || this.watchHandle) return;
+    // The rate ceiling, on the one path every re-arm goes through. If the last
+    // arm was too recent, defer this one to fill the interval rather than spawn
+    // now; the watchHandle guard above and clearWatchRearmTimer keep a single
+    // pending arm. Nothing is dropped -- a deferred arm still runs.
+    const wait = MIN_ARM_INTERVAL_MS - (Date.now() - this.lastArmAt);
+    if (wait > 0) {
+      this.watchRearmTimer = setTimeout(() => {
+        this.watchRearmTimer = null;
+        this.armWatch().catch((error) => this.failClientHandler("process/exited", error));
+      }, wait);
+      return;
+    }
+    this.lastArmAt = Date.now();
     const handle = `agmsg-watch-${Date.now()}-${Math.random().toString(36).slice(2)}`;
     this.watchHandle = handle;
     const command = [
@@ -1179,7 +1202,15 @@ class CodexBridge {
     this.watchHandle = null;
 
     if (params.exitCode === 0) {
-      this.watchFailureCount = 0;
+      // Decay, not reset. A wake is progress, but a wake arriving amid failures
+      // does not prove the host recovered -- it proves one message moved. The
+      // old reset-to-0 let a fail/fail/wake churn hold the counter below the
+      // limit forever, so a bridge that never stopped delivering also never
+      // stopped failing (#936 (b)). Forgiving ONE failure per delivery lets a
+      // genuinely-recovered bridge (mostly wakes) fall to 0 while a churn still
+      // climbs to the cap. A clean deadline (exit 2 below) is the stronger
+      // signal -- a full timeout ran end to end -- and still resets outright.
+      this.watchFailureCount = Math.max(0, this.watchFailureCount - 1);
       const maxId = parseMaxId(params.stdout);
       if (this.isStaleWake(maxId)) {
         await this.shutdown();

tests/test_codex_bridge.bats

@@ -1809,3 +1809,86 @@ EOF
   [[ "$output" =~ "started turn" ]]
   grep -q "turn/start" "$log"
 }
+
+# A WS app-server whose watch-once (process/spawn) exit codes are scripted by
+# $SCENARIO, so the bridge's re-arm accounting can be driven deterministically.
+# Shared by the two #936 tests below.
+_write_rearm_fake() {
+  cat >"$1" <<'EOF'
+const crypto = require("crypto"), fs = require("fs"), net = require("net");
+const [sock, logf] = process.argv.slice(2);
+const scenario = process.env.SCENARIO || "all124";
+try { fs.unlinkSync(sock); } catch (_) {}
+let arms = 0;
+function nextExit() {
+  if (scenario === "alt124_0") { const m = arms % 3; return m === 2 ? { code: 0, stdout: `status=pending count=1 max_id=${arms}\n` } : { code: 124, stdout: "" }; }
+  if (scenario === "flood0") return { code: 0, stdout: `status=pending count=1 max_id=${arms}\n` };
+  return { code: 124, stdout: "" };
+}
+function sendFrame(s, v) { const p = Buffer.from(JSON.stringify(v), "utf8"); let h; if (p.length < 126) h = Buffer.from([0x81, p.length]); else { h = Buffer.alloc(4); h[0]=0x81; h[1]=126; h.writeUInt16BE(p.length,2); } s.write(Buffer.concat([h, p])); }
+function handle(s, msg) {
+  if (msg.method === "initialize") return sendFrame(s, {jsonrpc:"2.0", id:msg.id, result:{}});
+  if (msg.method === "thread/resume") return sendFrame(s, {jsonrpc:"2.0", id:msg.id, result:{thread:{id:msg.params.threadId, status:{type:"idle"}}}});
+  if (msg.method === "turn/start") { sendFrame(s,{jsonrpc:"2.0",id:msg.id,result:{}}); setTimeout(()=>sendFrame(s,{jsonrpc:"2.0",method:"turn/completed",params:{threadId:msg.params.threadId,turn:{id:"t"}}}),5); return; }
+  if (msg.method === "process/spawn") { arms++; const { code, stdout } = nextExit(); fs.appendFileSync(logf, `${Date.now()} arm ${arms} exit ${code}\n`); sendFrame(s, {jsonrpc:"2.0", id:msg.id, result:{}}); setTimeout(()=>sendFrame(s,{jsonrpc:"2.0",method:"process/exited",params:{processHandle:msg.params.processHandle, exitCode:code, stdout, stderr:""}}), 5); return; }
+}
+function frames(s, st, chunk) { st.buffer = Buffer.concat([st.buffer, chunk]); while (st.buffer.length >= 2) { const op = st.buffer[0] & 0x0f; let len = st.buffer[1] & 0x7f; let off = 2; if (len === 126) { if (st.buffer.length < off+2) return; len = st.buffer.readUInt16BE(off); off+=2; } else if (len === 127) { if (st.buffer.length < off+8) return; len = st.buffer.readUInt32BE(off+4); off+=8; } const masked = (st.buffer[1] & 0x80) !== 0; const mo = off; if (masked) off += 4; if (st.buffer.length < off+len) return; let pl = st.buffer.slice(off, off+len); if (masked) { const mk = st.buffer.slice(mo, mo+4); pl = Buffer.from(pl.map((b,i)=>b^mk[i%4])); } st.buffer = st.buffer.slice(off+len); if (op === 0x1) handle(s, JSON.parse(pl.toString("utf8"))); } }
+const server = net.createServer((s) => { const st = { buffer: Buffer.alloc(0), upgraded: false, header: Buffer.alloc(0) }; s.on("data", (chunk) => { if (!st.upgraded) { st.header = Buffer.concat([st.header, chunk]); const end = st.header.indexOf("\r\n\r\n"); if (end === -1) return; const hdr = st.header.slice(0,end).toString("utf8"); const rest = st.header.slice(end+4); const key = (hdr.match(/Sec-WebSocket-Key: (.*)\r\n/i)||[])[1].trim(); const acc = crypto.createHash("sha1").update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest("base64"); s.write(["HTTP/1.1 101 Switching Protocols","Upgrade: websocket","Connection: Upgrade",`Sec-WebSocket-Accept: ${acc}`,"",""].join("\r\n")); st.upgraded = true; if (rest.length) frames(s, st, rest); return; } frames(s, st, chunk); }); s.on("close", () => server.close(()=>process.exit(0))); });
+server.listen(sock);
+EOF
+}
+
+@test "codex-bridge: the failure cap reaches even when wakes interleave (#936)" {
+  run node -e 'const net=require("net"),crypto=require("crypto");if(!net||!crypto)process.exit(1);'
+  [ "$status" -eq 0 ] || skip "node net/crypto not available"
+  run node -e 'const fs=require("fs"),net=require("net");const s=process.argv[1];try{fs.unlinkSync(s)}catch(_){}const sv=net.createServer();sv.on("error",()=>process.exit(2));sv.listen(s,()=>sv.close(()=>{try{fs.unlinkSync(s)}catch(_){}process.exit(0)}));' "$TEST_SKILL_DIR/probe3.sock"
+  [ "$status" -eq 0 ] || skip "unix socket listen not available"
+
+  local fake="$TEST_SKILL_DIR/rearm-fake.js" sock="$TEST_SKILL_DIR/rearm.sock" flog="$TEST_SKILL_DIR/rearm.log"
+  _write_rearm_fake "$fake"; : > "$flog"
+  SCENARIO=alt124_0 node "$fake" "$sock" "$flog" 3>&- &
+  local server_pid="$!"
+  for _ in {1..50}; do [ -S "$sock" ] && break; sleep 0.1; done
+
+  # fail, fail, wake, repeating: the old reset-to-0 held the counter below the
+  # limit forever. With the decay it climbs, so the bridge stops itself.
+  run node "$TYPES/codex/codex-bridge.js" \
+    --project "$PROJ" --team team --name alice --thread thread-x \
+    --app-server "unix://$sock" --timeout 1 --interval 1
+  kill "$server_pid" 2>/dev/null || true
+
+  [ "$status" -ne 0 ]
+  grep -q "stopping after" <<<"$output"
+  grep -q "consecutive watch-once failure" <<<"$output"
+}
+
+@test "codex-bridge: a flood of distinct wakes is rate-limited, not a re-arm storm (#936)" {
+  run node -e 'const net=require("net"),crypto=require("crypto");if(!net||!crypto)process.exit(1);'
+  [ "$status" -eq 0 ] || skip "node net/crypto not available"
+  run node -e 'const fs=require("fs"),net=require("net");const s=process.argv[1];try{fs.unlinkSync(s)}catch(_){}const sv=net.createServer();sv.on("error",()=>process.exit(2));sv.listen(s,()=>sv.close(()=>{try{fs.unlinkSync(s)}catch(_){}process.exit(0)}));' "$TEST_SKILL_DIR/probe4.sock"
+  [ "$status" -eq 0 ] || skip "unix socket listen not available"
+
+  local fake="$TEST_SKILL_DIR/rearm-fake2.js" sock="$TEST_SKILL_DIR/rearm2.sock" flog="$TEST_SKILL_DIR/rearm2.log"
+  _write_rearm_fake "$fake"; : > "$flog"
+  SCENARIO=flood0 node "$fake" "$sock" "$flog" 3>&- &
+  local server_pid="$!"
+  for _ in {1..50}; do [ -S "$sock" ] && break; sleep 0.1; done
+
+  # Every watch-once returns a wake with a fresh max_id, so the stale-wake guard
+  # never fires and this would re-arm with no delay. Let it run ~6 s, then stop.
+  node "$TYPES/codex/codex-bridge.js" \
+    --project "$PROJ" --team team --name alice --thread thread-x \
+    --app-server "unix://$sock" --timeout 1 --interval 1 >/dev/null 2>&1 3>&- &
+  local bpid="$!"
+  sleep 6
+  kill "$bpid" 2>/dev/null || true; wait "$bpid" 2>/dev/null || true
+  kill "$server_pid" 2>/dev/null || true
+
+  # The 1 s floor caps this near one arm per second. Without it the same 6 s
+  # produced hundreds. Assert a generous ceiling so the test is not timing-flaky
+  # but still fails a regression to the unbounded loop.
+  local arms
+  arms="$(grep -c ' arm ' "$flog")"
+  [ "$arms" -ge 1 ]
+  [ "$arms" -le 20 ]
+}