From 0b9dbde61e917bf578aeaefc37cebca41ef60bc3 Mon Sep 17 00:00:00 2001 From: Seller-1990 Date: Fri, 10 Jul 2026 11:51:36 +0800 Subject: [PATCH 1/2] fix(#223): opt-in AGENTBRIDGE_ALWAYS_QUEUE mirrors Codex replies to fallback queue The Codex->Claude reply rides a fire-and-forget notifications/claude/channel notification. When the Claude session is idle the push is silently dropped upstream (Claude Code #61797), so pushViaChannel's catch never fires, the fallback queue is never populated, and get_messages stays empty -- the reply is unrecoverable (see #223). This adds an opt-in AGENTBRIDGE_ALWAYS_QUEUE=1 flag. When set, real Codex replies (non-"system_" ids) are also mirrored into the fallback queue after the channel push, so get_messages becomes a reliable pull path even for idle sessions. The channel push still fires (mid-turn liveness is unchanged), system messages are excluded to avoid flooding the queue, and a failed push still queues exactly once (no double-queue). Off by default -> delivery unchanged. - src/claude-adapter.ts: mirror-to-queue after push when flag is set - src/unit-test/message-delivery.test.ts: 4 tests - rebuilt committed bundles via bun run build:plugin Validation: bun run typecheck clean; bun run test:unit 1543 pass / 0 fail. --- plugins/agentbridge/server/bridge-server.js | 15 +++++-- plugins/agentbridge/server/daemon.js | 4 +- src/claude-adapter.ts | 24 +++++++++++ src/unit-test/message-delivery.test.ts | 47 +++++++++++++++++++++ 4 files changed, 84 insertions(+), 6 deletions(-) diff --git a/plugins/agentbridge/server/bridge-server.js b/plugins/agentbridge/server/bridge-server.js index 6b396001..0cbf6516 100755 --- a/plugins/agentbridge/server/bridge-server.js +++ b/plugins/agentbridge/server/bridge-server.js @@ -10150,7 +10150,7 @@ function finalize(ctx, schema) { result.$schema = "http://json-schema.org/draft-07/schema#"; } else if (ctx.target === "draft-04") { result.$schema = "http://json-schema.org/draft-04/schema#"; - } else if (ctx.target === "openapi-3.0") {} else {} + } else if (ctx.target === "openapi-3.0") {} if (ctx.external?.uri) { const id = ctx.external.registry.get(schema)?.id; if (!id) @@ -10372,7 +10372,7 @@ var literalProcessor = (schema, ctx, json, _params) => { if (val === undefined) { if (ctx.unrepresentable === "throw") { throw new Error("Literal `undefined` cannot be represented in JSON Schema"); - } else {} + } } else if (typeof val === "bigint") { if (ctx.unrepresentable === "throw") { throw new Error("BigInt literals cannot be represented in JSON Schema"); @@ -14286,6 +14286,8 @@ class ClaudeAdapter extends EventEmitter { async pushViaChannel(message) { const deliveryAttemptId = `codex_msg_${this.notificationIdPrefix}_${++this.notificationSeq}`; const ts = new Date(message.timestamp).toISOString(); + const mirrorToQueue = process.env.AGENTBRIDGE_ALWAYS_QUEUE === "1" && typeof message.id === "string" && !message.id.startsWith("system_"); + let queuedInCatch = false; try { await this.server.notification({ method: "notifications/claude/channel", @@ -14307,6 +14309,11 @@ class ClaudeAdapter extends EventEmitter { } catch (e) { this.log(`Push notification failed: ${e.message}`); this.queueFallbackMessage(message); + queuedInCatch = true; + } + if (mirrorToQueue && !queuedInCatch) { + this.queueFallbackMessage(message); + this.log(`Always-queue: mirrored ${message.id} to fallback queue (#223)`); } } rememberDelivery(message) { @@ -14707,10 +14714,10 @@ function defineNumber(value, fallback) { } var BUILD_INFO = Object.freeze({ version: defineString("0.1.30", "0.0.0-source"), - commit: defineString("99d0f4a", "source"), + commit: defineString("a3e927f", "source"), bundle: defineBundle("plugin"), contractVersion: defineNumber(1, CONTRACT_VERSION), - codeHash: defineString("0cb79932198b", "source") + codeHash: defineString("cb386a074708", "source") }); function sameRuntimeContract(a, b) { if (!a || !b) diff --git a/plugins/agentbridge/server/daemon.js b/plugins/agentbridge/server/daemon.js index 5388db2c..7174f8e6 100755 --- a/plugins/agentbridge/server/daemon.js +++ b/plugins/agentbridge/server/daemon.js @@ -30,10 +30,10 @@ function defineNumber(value, fallback) { } var BUILD_INFO = Object.freeze({ version: defineString("0.1.30", "0.0.0-source"), - commit: defineString("99d0f4a", "source"), + commit: defineString("a3e927f", "source"), bundle: defineBundle("plugin"), contractVersion: defineNumber(1, CONTRACT_VERSION), - codeHash: defineString("0cb79932198b", "source") + codeHash: defineString("cb386a074708", "source") }); function daemonStatusBuildInfo() { return { ...BUILD_INFO }; diff --git a/src/claude-adapter.ts b/src/claude-adapter.ts index 3d6c1d35..fe09d512 100644 --- a/src/claude-adapter.ts +++ b/src/claude-adapter.ts @@ -7,6 +7,12 @@ * (The old AGENTBRIDGE_MODE=pull delivery mode was removed: it could not wake * an idle session, which silently broke the budget RESUME chain.) * + * #223: the channel push is fire-and-forget and is silently dropped when the + * Claude session is idle, so the fallback catch never fires and the reply is + * lost. Opt-in AGENTBRIDGE_ALWAYS_QUEUE=1 additionally mirrors real Codex + * replies into the fallback queue, making get_messages a reliable pull path + * for idle sessions. Off by default → delivery behavior is unchanged. + * * Emits: * - "ready" () — MCP connected * - "reply" (msg: BridgeMessage) — Claude used the reply tool @@ -241,6 +247,18 @@ export class ClaudeAdapter extends EventEmitter { const deliveryAttemptId = `codex_msg_${this.notificationIdPrefix}_${++this.notificationSeq}`; const ts = new Date(message.timestamp).toISOString(); + // #223: notifications/claude/channel is fire-and-forget and is silently + // dropped when the Claude session is idle, so the catch below never runs + // and the reply is lost with get_messages staying empty. With + // AGENTBRIDGE_ALWAYS_QUEUE=1, mirror real Codex replies (non-system ids) + // into the fallback queue as well, so get_messages is a reliable pull path + // even for idle sessions. Off by default → delivery behavior is unchanged. + const mirrorToQueue = + process.env.AGENTBRIDGE_ALWAYS_QUEUE === "1" && + typeof message.id === "string" && + !message.id.startsWith("system_"); + let queuedInCatch = false; + try { await this.server.notification({ method: "notifications/claude/channel", @@ -265,6 +283,12 @@ export class ClaudeAdapter extends EventEmitter { } catch (e: any) { this.log(`Push notification failed: ${e.message}`); this.queueFallbackMessage(message); + queuedInCatch = true; + } + + if (mirrorToQueue && !queuedInCatch) { + this.queueFallbackMessage(message); + this.log(`Always-queue: mirrored ${message.id} to fallback queue (#223)`); } } diff --git a/src/unit-test/message-delivery.test.ts b/src/unit-test/message-delivery.test.ts index 16270db2..f2e0c4f0 100644 --- a/src/unit-test/message-delivery.test.ts +++ b/src/unit-test/message-delivery.test.ts @@ -100,6 +100,53 @@ describe("Push-only delivery: AGENTBRIDGE_MODE is ignored", () => { }); }); +// #223: with AGENTBRIDGE_ALWAYS_QUEUE=1 the channel push still fires, but real +// Codex replies are ALSO mirrored into the fallback queue so get_messages is a +// reliable pull path even when the Claude session is idle and the fire-and- +// forget channel push is silently dropped. +describe("Message delivery: AGENTBRIDGE_ALWAYS_QUEUE (#223)", () => { + const origAlwaysQueue = process.env.AGENTBRIDGE_ALWAYS_QUEUE; + afterEach(() => { + if (origAlwaysQueue !== undefined) process.env.AGENTBRIDGE_ALWAYS_QUEUE = origAlwaysQueue; + else delete process.env.AGENTBRIDGE_ALWAYS_QUEUE; + }); + + test("mirrors a real Codex reply to the fallback queue while still pushing", async () => { + const adapter = createAdapter(); + const notifications = withMockedChannel(adapter); + process.env.AGENTBRIDGE_ALWAYS_QUEUE = "1"; + await adapter.pushNotification(makeBridgeMessage("codex reply", undefined, "msg_abc123")); + expect(notifications).toHaveLength(1); // channel push still happens + expect(adapter.pendingMessages).toHaveLength(1); // and it is drainable via get_messages + expect(adapter.pendingMessages[0].content).toBe("codex reply"); + }); + + test("does not mirror system messages (system_ id prefix)", async () => { + const adapter = createAdapter(); + withMockedChannel(adapter); + process.env.AGENTBRIDGE_ALWAYS_QUEUE = "1"; + await adapter.pushNotification(makeBridgeMessage("turn started", undefined, "system_turn_started_1")); + expect(adapter.pendingMessages).toHaveLength(0); + }); + + test("does not double-queue when the channel push fails", async () => { + const adapter = createAdapter(); + withMockedChannel(adapter, "fail"); + process.env.AGENTBRIDGE_ALWAYS_QUEUE = "1"; + await adapter.pushNotification(makeBridgeMessage("codex reply", undefined, "msg_def456")); + expect(adapter.pendingMessages).toHaveLength(1); // queued exactly once (catch path) + }); + + test("default (flag unset) keeps push-only behavior — nothing queued", async () => { + const adapter = createAdapter(); + const notifications = withMockedChannel(adapter); + delete process.env.AGENTBRIDGE_ALWAYS_QUEUE; + await adapter.pushNotification(makeBridgeMessage("codex reply", undefined, "msg_ghi789")); + expect(notifications).toHaveLength(1); + expect(adapter.pendingMessages).toHaveLength(0); + }); +}); + describe("Message delivery: fallback queue", () => { test("queueFallbackMessage adds message to pendingMessages", () => { const adapter = createAdapter(); From 48ab788206afa8e9b7b47c532d64a91ede3e6064 Mon Sep 17 00:00:00 2001 From: Seller-1990 Date: Fri, 10 Jul 2026 14:50:50 +0800 Subject: [PATCH 2/2] docs: add structural-kind note and retention security caveat to #223 shim Per Codex review feedback: - Note that system_ id prefix is used because BridgeMessage lacks a structural kind field; prefer such a field if added in the future. - Note that ALWAYS_QUEUE extends the fallback queue retention window (same data class, longer lifetime). - Rebuilt bundle. --- plugins/agentbridge/server/bridge-server.js | 4 ++-- plugins/agentbridge/server/daemon.js | 4 ++-- src/claude-adapter.ts | 9 +++++++++ 3 files changed, 13 insertions(+), 4 deletions(-) diff --git a/plugins/agentbridge/server/bridge-server.js b/plugins/agentbridge/server/bridge-server.js index 0cbf6516..e5c757e3 100755 --- a/plugins/agentbridge/server/bridge-server.js +++ b/plugins/agentbridge/server/bridge-server.js @@ -14714,10 +14714,10 @@ function defineNumber(value, fallback) { } var BUILD_INFO = Object.freeze({ version: defineString("0.1.30", "0.0.0-source"), - commit: defineString("a3e927f", "source"), + commit: defineString("0b9dbde", "source"), bundle: defineBundle("plugin"), contractVersion: defineNumber(1, CONTRACT_VERSION), - codeHash: defineString("cb386a074708", "source") + codeHash: defineString("b3fd79558954", "source") }); function sameRuntimeContract(a, b) { if (!a || !b) diff --git a/plugins/agentbridge/server/daemon.js b/plugins/agentbridge/server/daemon.js index 7174f8e6..928d3157 100755 --- a/plugins/agentbridge/server/daemon.js +++ b/plugins/agentbridge/server/daemon.js @@ -30,10 +30,10 @@ function defineNumber(value, fallback) { } var BUILD_INFO = Object.freeze({ version: defineString("0.1.30", "0.0.0-source"), - commit: defineString("a3e927f", "source"), + commit: defineString("0b9dbde", "source"), bundle: defineBundle("plugin"), contractVersion: defineNumber(1, CONTRACT_VERSION), - codeHash: defineString("cb386a074708", "source") + codeHash: defineString("b3fd79558954", "source") }); function daemonStatusBuildInfo() { return { ...BUILD_INFO }; diff --git a/src/claude-adapter.ts b/src/claude-adapter.ts index fe09d512..2c01b969 100644 --- a/src/claude-adapter.ts +++ b/src/claude-adapter.ts @@ -253,6 +253,15 @@ export class ClaudeAdapter extends EventEmitter { // AGENTBRIDGE_ALWAYS_QUEUE=1, mirror real Codex replies (non-system ids) // into the fallback queue as well, so get_messages is a reliable pull path // even for idle sessions. Off by default → delivery behavior is unchanged. + // + // NOTE: system messages are identified by their id prefix ("system_*") + // because BridgeMessage has no structural kind/type field (both system and + // agent messages carry source="codex"). If a structural discriminator (e.g. + // a `kind` field) is added in the future, prefer it over the id prefix. + // + // SECURITY: with this flag on, the fallback queue retains real agent reply + // content until drained by get_messages. This does not introduce a new data + // class but extends the retention window — document accordingly. const mirrorToQueue = process.env.AGENTBRIDGE_ALWAYS_QUEUE === "1" && typeof message.id === "string" &&