diff --git a/scripts/drivers/types/codex/codex-bridge.js b/scripts/drivers/types/codex/codex-bridge.js index afc1a76aa..9510e8a61 100755 --- a/scripts/drivers/types/codex/codex-bridge.js +++ b/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 [--type codex] [--team ] [--name ] @@ -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(); diff --git a/tests/test_codex_bridge.bats b/tests/test_codex_bridge.bats index 16d0b7dd4..240bee5d7 100644 --- a/tests/test_codex_bridge.bats +++ b/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 ] +}