diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 61ec09c..f8e54cd 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -11,7 +11,7 @@ "plugins": [ { "name": "agentbridge", - "description": "Bridge Claude Code and Codex through a shared daemon, push channel delivery, and reply/get_messages tools.", + "description": "Bridge Claude Code and Codex through a shared daemon, acknowledged channel delivery, and reply/get_messages/ack_messages tools.", "version": "0.1.30", "author": { "name": "AgentBridge Contributors", diff --git a/AGENTS.md b/AGENTS.md index 9f29a91..bc48b7c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -3,7 +3,7 @@ You are working in a **multi-agent environment** powered by AgentBridge. Another AI agent (Claude, by Anthropic) is available in a parallel session on this machine. -Communication happens via AgentBridge MCP tools — Claude has `reply` and `get_messages` tools. +Communication happens via AgentBridge MCP tools — Claude has `reply`, `get_messages`, and `ack_messages` tools. ### When to collaborate vs. work solo - **Collaborate** when the task benefits from a second perspective, parallel execution, or capabilities the other agent has. diff --git a/CLAUDE.md b/CLAUDE.md index 1299dbe..e5de3ab 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -53,7 +53,7 @@ Claude Code ── MCP stdio ──▶ bridge.ts (foreground) - **`src/bridge.ts`** — foreground MCP server registered as a Claude Code plugin channel. Exits when Claude Code closes. - **`src/daemon.ts`** — long-lived background process; owns the Codex app-server proxy and the single source of truth for bridge state. Survives Claude Code restarts; `bridge.ts` reconnects with exponential backoff. - **`src/control-protocol.ts`** — message schema for the control WebSocket between foreground and daemon. -- **`src/claude-adapter.ts`** — MCP tool surface exposed to Claude (`reply`, `get_messages`). Emits `notifications/claude/channel` on inbound messages (push mode). +- **`src/claude-adapter.ts`** — MCP tool surface exposed to Claude (`reply`, `get_messages`, `ack_messages`). Emits `notifications/claude/channel` on inbound messages (push mode) over an acknowledged in-memory mailbox. - **`src/codex-adapter.ts`** — WebSocket proxy in front of Codex app-server; intercepts `agentMessage` items and injects turns via `turn/start`. - **`src/message-filter.ts`** — collapses noisy intermediate events so only meaningful `agentMessage` payloads reach Claude. - **`src/daemon-lifecycle.ts`** — shared `ensureRunning` / `kill` / startup-lock logic; both the CLI and `bridge.ts` call into this. @@ -66,7 +66,7 @@ Claude Code ── MCP stdio ──▶ bridge.ts (foreground) ### Data flow invariants - Every `BridgeMessage` carries a `source: "claude" | "codex"` — the bridge **never forwards a message back to its origin** (loop prevention). -- Message delivery is always push (channel notifications). A failed push falls back to an in-memory queue drained by `get_messages`. (The legacy `AGENTBRIDGE_MODE=pull` mode was removed; the env var is ignored with a one-time warning.) +- Message delivery is push (channel notifications) over an authoritative in-memory mailbox: every admissible message is queued **before** its Channel push, `get_messages` re-reads pending messages non-destructively under stable delivery IDs, and only `ack_messages` removes them (bounded retries re-push unacknowledged messages). (The legacy `AGENTBRIDGE_MODE=pull` mode was removed; the env var is ignored with a one-time warning.) - Ports are allocated per pair from a registry (slot-based, +10 strides from the base 4500/4501/4502); the legacy fixed defaults remain the slot-0 values. Multiple pairs run side-by-side, one per project directory. - All state lives in the platform state dir (`AGENTBRIDGE_STATE_DIR`, default `~/Library/Application Support/AgentBridge/` on macOS, `$XDG_STATE_HOME/agentbridge/` on Linux). The daemon uses `startup.lock` + `killed` sentinel to coordinate startup and explicit-kill-don't-restart semantics. diff --git a/README.md b/README.md index 58305ee..ad92752 100644 --- a/README.md +++ b/README.md @@ -53,7 +53,7 @@ What that buys you, concretely: ## Features - **Bidirectional Claude ↔ Codex messaging** in one working session — the daemon intercepts Codex output and pushes it to Claude as channel notifications; Claude replies via the `reply` MCP tool, and the bridge injects the reply into the Codex thread as a `turn/start`. -- **Push delivery with fallback** — messages arrive as channel notifications; a failed push falls back to an in-memory queue drained by `get_messages`. Loop prevention via the per-message `source` field. +- **Acknowledged at-least-once delivery** - every Codex message enters a bounded in-memory mailbox before its Channel push. Each admission gets an immutable delivery-generation ID, separate from the daemon source ID, so a delayed ACK cannot delete a newer message. Claude acknowledges processed IDs with `ack_messages`; missed pushes remain visible through non-destructive `get_messages` polling, and unacknowledged pushes receive bounded FIFO retries. Loop prevention uses the per-message `source` field. - **Turn coordination** — a busy-guard rejects replies during an active Codex turn; a per-turn inactivity watchdog stops a lost `turn/completed` from locking injection forever; noisy intermediate events are collapsed so only meaningful `agentMessage` payloads reach Claude. - **Multiple pairs side by side** — one Claude+Codex pair per project directory, ports allocated per pair in +10 strides from 4500. Pair-aware `claude` / `codex` / `resume` / `kill` / `doctor` / `budget` via `--pair`. - **Resilient lifecycle** — a persistent background daemon survives Claude Code restarts (auto-reconnect with backoff); orphan-process cleanup; `abg doctor` read-only diagnostics; `abg pairs prune` reclaims stranded state. @@ -249,6 +249,10 @@ The config is loaded by the CLI and daemon at startup. Re-running `init` is idem | `AGENTBRIDGE_CODEX_TRANSPORT` | `auto` | How the daemon reaches the Codex app-server: `auto` (probe `codex app-server --help`, use `ws://` if supported else fall back to a `unix://` socket via a transparent relay), `ws` (force ws), or `unix` (force unix socket + relay). For builds that drop `ws://` listen support (issue #85) | | `AGENTBRIDGE_STATE_DIR` | Platform default | State directory for pid, status, logs (macOS: `~/Library/Application Support/agentbridge/`, Linux: `$XDG_STATE_HOME/agentbridge/`) | | `AGENTBRIDGE_DAEMON_ENTRY` | `./daemon.ts` | Override daemon entry point (used by plugin bundles) | +| `AGENTBRIDGE_MAX_BUFFERED_MESSAGES` | `100` | Maximum unacknowledged messages in the Claude adapter mailbox; overflow evicts the oldest entry with an observable warning | +| `AGENTBRIDGE_MAX_BUFFERED_BYTES` | `4194304` | Maximum UTF-8 content bytes in the mailbox; a single larger message is omitted with an observable warning | +| `AGENTBRIDGE_DELIVERY_RETRY_BASE_MS` | `60000` | Delay before retrying an unacknowledged Channel push; later retries use exponential backoff | +| `AGENTBRIDGE_DELIVERY_MAX_ATTEMPTS` | `3` | Total Channel attempts per ordinary message, including the initial push; exhausted messages remain available through `get_messages` | | `NO_UPDATE_NOTIFIER` | unset | Set to any value to disable the "update available" notice (ecosystem-standard opt-out) | | `AGENTBRIDGE_NO_UPDATE_NOTIFIER` | unset | Namespaced opt-out for the update notice (same effect as `NO_UPDATE_NOTIFIER`) | | `AGENTBRIDGE_UPDATE_PROMPT` | unset | Set to `0` to disable the interactive update prompt and keep pure notice-only behavior | @@ -282,6 +286,9 @@ AgentBridge can keep a long task moving across subscription-quota windows instea ## Current Limitations - Only forwards `agentMessage` items, not intermediate `commandExecution`, `fileChange`, or similar events +- The acknowledged mailbox is in the Claude adapter process and is not persisted. It survives daemon or Codex reconnects while that adapter stays alive, but a Claude adapter/plugin process restart loses unacknowledged entries. This is not crash-durable delivery. +- Channel retry improves latency and recovery probability but cannot guarantee that a fully idle Claude session wakes automatically. `get_messages` is the recovery path once Claude is active. +- Delivery is at least once, not exactly once. Claude must finish processing a stable message ID before acknowledging it; a lost acknowledgement can cause a bounded redelivery. - Single Codex thread per pair, no multi-session support within a pair yet - Single Claude foreground connection per pair; a new Claude session replaces the previous one - Multiple pairs run side-by-side on one machine (one per project directory); Windows is not an officially supported platform yet diff --git a/README.zh-CN.md b/README.zh-CN.md index 0a92bd6..b80d678 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -53,7 +53,7 @@ English version: [README.md](README.md) ## 功能 - **Claude ↔ Codex 双向消息**（同一工作会话）：拦截 Codex 输出并以 channel 通知推给 Claude；Claude 用 `reply` MCP tool 回复，作为 `turn/start` 注入 Codex thread。 -- **Push 投递 + 兜底**：消息以 channel 通知投递；推送失败回退到内存队列，由 `get_messages` 排空。靠每条消息的 `source` 字段防循环。 +- **确认制 at-least-once 投递**：每条 Codex 消息在 Channel 推送前先进入有界内存信箱；每次入箱分配一个不可变的投递代 ID（与 daemon 源 ID 分离），迟到的 ACK 不会误删复用同一源 ID 的新消息。Claude 用 `ack_messages` 确认已处理的 ID；漏掉的推送可通过非破坏性的 `get_messages` 轮询看到，未确认的推送按 FIFO 有界重试。靠每条消息的 `source` 字段防循环。 - **回合协调**：busy-guard 在 Codex 活跃 turn 期间拒绝回复；单 turn 非活动看门狗避免丢失 `turn/completed` 永久锁死注入；折叠噪声中间事件，只把有意义的 `agentMessage` 送达 Claude。 - **多对并行**：每个项目目录一对 Claude+Codex，端口按 +10 步长从 4500 分配；`claude` / `codex` / `resume` / `kill` / `doctor` / `budget` 支持 `--pair` 指定。 - **韧性生命周期**：常驻后台 daemon 跨 Claude Code 重启存活（指数退避自动重连）；孤儿进程清理；`abg doctor` 只读诊断；`abg pairs prune` 回收滞留状态。 @@ -241,6 +241,10 @@ CLI 和 daemon 启动时会加载该配置。重复运行 `init` 是幂等的， | 变量 | 默认值 | 说明 | |------|--------|------| +| `AGENTBRIDGE_MAX_BUFFERED_MESSAGES` | `100` | Claude adapter 内存信箱最多保留的未确认消息数；溢出时逐出最旧条目并给出可观察的警告 | +| `AGENTBRIDGE_MAX_BUFFERED_BYTES` | `4194304` | 信箱保留内容的 UTF-8 字节上限；单条超限消息被省略并给出可观察的警告 | +| `AGENTBRIDGE_DELIVERY_RETRY_BASE_MS` | `60000` | 未确认 Channel 推送的首次重试延迟；后续重试按指数退避 | +| `AGENTBRIDGE_DELIVERY_MAX_ATTEMPTS` | `3` | 每条普通消息的 Channel 总尝试次数（含首推）；用尽后消息仍可通过 `get_messages` 获取 | | `CODEX_WS_PORT` | `4500` | Codex app-server WebSocket 端口 | | `CODEX_PROXY_PORT` | `4501` | Bridge 代理端口，Codex TUI 连接此端口 | | `AGENTBRIDGE_CONTROL_PORT` | `4502` | bridge.ts 与 daemon.ts 之间的控制端口 | @@ -288,6 +292,12 @@ AgentBridge 能让长任务跨订阅额度窗口持续推进，而不是某一 休眠/禁用状态、Codex `.git` 限制及其它坑，见 **[排错文档](docs/TROUBLESHOOTING.md)**。 +### 投递可靠性边界 + +每条 Codex 消息都会在 Channel 推送前入箱，并获得一个不可变的投递代 ID。`get_messages` 以稳定顺序重复返回未确认消息；`ack_messages` 只移除请求中已处理的 ID。这是 Claude adapter 存活期间的有界 at-least-once 投递。投递语义是 at-least-once 而非 exactly-once：Claude 应在处理完成后再确认对应的稳定 ID；确认丢失可能造成有界的重复投递。 + +信箱不持久化：Claude adapter / 插件进程重启会丢失未确认条目。Channel 重试也无法保证完全空闲的 Claude 会话被自动唤醒。溢出与超大消息的行为可观察，但会按配置边界逐出内容。 + ## Roadmap - **更多 adapter**：今天 AgentBridge 接的是 Claude Code ↔ Codex。下一个候选：**OpenCode、OpenClaw、Hermes Agent、Gemini CLI**。到 [adapter roadmap issue](https://github.com/raysonmeng/agent-bridge/issues/212) 投票。 diff --git a/docs/test-plans/issue-223-reliable-mailbox.md b/docs/test-plans/issue-223-reliable-mailbox.md new file mode 100644 index 0000000..bd9a5e8 --- /dev/null +++ b/docs/test-plans/issue-223-reliable-mailbox.md @@ -0,0 +1,84 @@ +# Issue 223 reliable mailbox test plan + +This plan validates Codex-to-Claude delivery without weakening normal Claude or Codex permission controls. It covers the in-memory P0/P1 mailbox implemented for issue 223. It does not claim crash durability or guaranteed idle wake-up. + +## Safety setup + +Use a disposable Git repository with no production files. Build source and committed plugin artifacts with the pinned Bun version, then verify synchronization: + +```bash +bun install --frozen-lockfile --ignore-scripts +bun run typecheck +bun run build:cli +bun run build:plugin +bun run verify:plugin-sync +``` + +Launch both native subscription-authenticated CLIs with safe controls: + +```bash +AGENTBRIDGE_SAFE=1 bun dist/cli.js --pair issue223 claude \ + --safe --permission-mode manual + +AGENTBRIDGE_SAFE=1 bun dist/cli.js --pair issue223 codex \ + --safe --new --sandbox read-only --ask-for-approval untrusted --no-alt-screen +``` + +Approve only the specific AgentBridge MCP calls under test. No API key is required. + +## Deterministic suite + +Run: + +```bash +bun test src/unit-test/reliable-mailbox.test.ts +``` + +The suite must cover: + +- A resolved Channel write with no consumer, followed by two identical `get_messages` snapshots. +- Successful Channel processing followed by `ack_messages`, with no later pull or retry duplicate. +- A throwing Channel write that leaves exactly one mailbox entry. +- Partial and repeated polling with deletion limited to requested immutable delivery IDs. +- Both lost-ACK directions: no committed ACK remains recoverable; a committed ACK with a lost response is idempotent when repeated. +- Adapter restart loss and same-adapter daemon, Claude Channel, and Codex reconnect survival. +- Rapid arrivals, ACK concurrent with arrival, source-ID conflicts, TTL/capacity reuse, collision aliases, multiple adapters, and FIFO retry order. +- Count and UTF-8 byte limits, oversized messages, overflow, ordinary system notices, budget-resume aliases, retry exhaustion, and invalid ACK input. + +## Live idle recovery + +Leave Claude idle at its prompt. Ask Codex to return a unique one-line sentinel. If Claude does not visibly wake, activate Claude manually and ask it to call `get_messages` twice without acknowledging. + +Pass criteria: + +1. Both polls contain the sentinel under the same delivery-generation ID. +2. `ack_messages` for that exact ID succeeds after processing. +3. A later poll no longer contains the sentinel. + +If Claude wakes and processes the initial Channel push, that round tests successful push delivery rather than the idle-drop recovery path. Repeat with a new sentinel. If every push wakes Claude, record the live idle-drop result as inconclusive and retain the deterministic silent-consumer simulation as the proof for P0. + +## Live Channel acknowledgement + +While Claude is active, send a unique Codex sentinel. The Channel body and metadata expose the immutable delivery ID and direct Claude to call `ack_messages`. Claude must acknowledge without first discovering the ID through `get_messages`. A later poll must not return that sentinel. + +## Restart boundaries + +Leave a message unacknowledged and test each boundary separately: + +- Restart only the daemon, retaining the same Claude adapter process: the adapter mailbox remains. +- Reconnect Codex, retaining the same Claude adapter: completed replies already in the mailbox remain. +- Reconnect the Claude Channel through the same adapter process: the mailbox remains. +- Exit and relaunch the Claude plugin/adapter process: the mailbox is lost. This demonstrates that P2 is not implemented. + +Do not infer crash durability from a daemon-only restart because the authoritative mailbox lives in the Claude adapter process. + +## Cleanup + +Exit Codex and Claude, then stop and remove only the disposable pair: + +```bash +AGENTBRIDGE_SAFE=1 bun dist/cli.js --pair issue223 kill +bun dist/cli.js pairs rm issue223 +``` + +Remove any local development marketplace/plugin registration only if this test created it and no previous AgentBridge installation needs to be restored. diff --git a/plugins/agentbridge/README.md b/plugins/agentbridge/README.md index 2091bbd..932db37 100644 --- a/plugins/agentbridge/README.md +++ b/plugins/agentbridge/README.md @@ -1,6 +1,6 @@ # AgentBridge Plugin -Claude Code plugin for AgentBridge. This plugin packages the AgentBridge MCP frontend with push channel delivery (a failed push falls back to an in-memory queue drained by `get_messages`), the `/agentbridge:init` command, and a non-blocking SessionStart health check. +Claude Code plugin for AgentBridge. This plugin packages the AgentBridge MCP frontend with acknowledged Channel delivery, the `/agentbridge:init` command, and a non-blocking SessionStart health check. Every admissible message is queued before push, receives an immutable delivery-generation ID, and remains available through `get_messages` until Claude confirms that ID with `ack_messages` (a message exceeding the mailbox size bound is not retained — it is delivered best-effort only and omitted with an observable warning). ## Structure @@ -38,6 +38,7 @@ This creates self-contained bundles at: ## Notes - The plugin frontend launches the sibling daemon bundle via `AGENTBRIDGE_DAEMON_ENTRY=./daemon.js`. -- Claude delivery is always push notifications. If a push fails, the message is queued and can be drained via `get_messages` (per-message fallback — the legacy `AGENTBRIDGE_MODE=pull` mode was removed and the env var is ignored with a one-time warning). +- Claude delivery uses Channel push as a latency optimization over a bounded in-memory mailbox. `get_messages` repeats unacknowledged stable IDs without deleting them; `ack_messages` removes only IDs Claude has finished processing. Ordinary unacknowledged pushes retry twice by default with exponential backoff. The legacy `AGENTBRIDGE_MODE=pull` value remains ignored with a one-time warning. +- Mailbox state is not persisted. A Claude adapter/plugin process restart loses it, and Channel cannot guarantee that a fully idle Claude session wakes automatically. Delivery is at least once while the adapter is alive, not exactly once or crash durable. - The SessionStart hook is informational only. It never starts or stops the daemon. - The command at `/agentbridge:init` edits project-local `.agentbridge/` files only; plugin installation and marketplace registration remain terminal-side tasks (`agentbridge init` / `agentbridge dev`). diff --git a/plugins/agentbridge/server/bridge-server.js b/plugins/agentbridge/server/bridge-server.js index 6b39600..00439ef 100755 --- a/plugins/agentbridge/server/bridge-server.js +++ b/plugins/agentbridge/server/bridge-server.js @@ -13661,7 +13661,7 @@ class StdioServerTransport { // src/claude-adapter.ts import { EventEmitter } from "events"; -import { randomUUID } from "crypto"; +import { createHash, randomUUID } from "crypto"; import { performance } from "perf_hooks"; // src/rotating-log.ts @@ -14161,13 +14161,18 @@ var DEFAULT_MAX_BUFFERED_MESSAGES = 100; var DEFAULT_MAX_BUFFERED_BYTES = 4 * 1024 * 1024; var DEFAULT_DEDUPE_CAPACITY = 2048; var DEFAULT_DEDUPE_TTL_MS = 20 * 60 * 1000; +var DEFAULT_DELIVERY_RETRY_BASE_MS = 60 * 1000; +var DEFAULT_DELIVERY_MAX_ATTEMPTS = 3; var DEFAULT_BUDGET_FRESH_TTL_MS = 25 * 1000; var CLAUDE_INSTRUCTIONS = [ "Codex is an AI coding agent (OpenAI) running in a separate session on the same machine.", "", "## Message delivery", 'Messages from Codex arrive as tags (push).', - "If a push fails, the message is queued \u2014 call get_messages to drain the fallback queue.", + "Every message is queued before push. Channel delivery is a latency optimization, not proof of receipt.", + "A repeated delivery ID is the same logical message. Never repeat completed work for an ID you already processed.", + "After fully processing a pushed message, call ack_messages with its meta.message_id. Do not acknowledge before processing.", + "If a push is missed, call get_messages. It returns the same stable message IDs until ack_messages confirms them.", "", "## Collaboration roles", "Default roles in this setup:", @@ -14185,7 +14190,7 @@ var CLAUDE_INSTRUCTIONS = [ "", "## How to interact", "- Use the reply tool to send messages back to Codex \u2014 pass chat_id back.", - "- Use the get_messages tool to check for pending messages from Codex.", + "- Use the get_messages tool to check for pending messages from Codex, then call ack_messages only for IDs you fully processed.", "- After sending a reply, call get_messages to check for responses.", "- When the user asks about Codex status or progress, call get_messages.", "", @@ -14204,6 +14209,7 @@ var CLAUDE_INSTRUCTIONS = [ class ClaudeAdapter extends EventEmitter { server; notificationSeq = 0; + deliverySeq = 0; sessionId; notificationIdPrefix; instanceId; @@ -14216,6 +14222,7 @@ class ClaudeAdapter extends EventEmitter { pendingMessageBytes = 0; maxBufferedMessages; maxBufferedBytes; + ackIdsCap; droppedMessageCount = 0; oversizedMessageCount = 0; oversizedMessageBytes = 0; @@ -14224,6 +14231,10 @@ class ClaudeAdapter extends EventEmitter { dedupeTtlMs; monotonicNow; deliveredMessageIds = new Map; + deliveryRetryBaseMs; + deliveryMaxAttempts; + deliveryScheduler; + deliveryRetries = new Map; budgetSnapshot = null; budgetFreshTtlMs; wallNow; @@ -14242,9 +14253,13 @@ class ClaudeAdapter extends EventEmitter { } this.maxBufferedMessages = positiveIntegerOr(options.maxBufferedMessages, parsePositiveIntegerEnv("AGENTBRIDGE_MAX_BUFFERED_MESSAGES", DEFAULT_MAX_BUFFERED_MESSAGES)); this.maxBufferedBytes = positiveIntegerOr(options.maxBufferedBytes, parsePositiveIntegerEnv("AGENTBRIDGE_MAX_BUFFERED_BYTES", DEFAULT_MAX_BUFFERED_BYTES)); + this.ackIdsCap = Math.max(100, this.maxBufferedMessages); this.dedupeCapacity = positiveIntegerOr(options.dedupeCapacity, DEFAULT_DEDUPE_CAPACITY); this.dedupeTtlMs = positiveIntegerOr(options.dedupeTtlMs, DEFAULT_DEDUPE_TTL_MS); this.monotonicNow = options.now ?? (() => performance.now()); + this.deliveryRetryBaseMs = positiveIntegerOr(options.deliveryRetryBaseMs, parsePositiveIntegerEnv("AGENTBRIDGE_DELIVERY_RETRY_BASE_MS", DEFAULT_DELIVERY_RETRY_BASE_MS)); + this.deliveryMaxAttempts = positiveIntegerOr(options.deliveryMaxAttempts, parsePositiveIntegerEnv("AGENTBRIDGE_DELIVERY_MAX_ATTEMPTS", DEFAULT_DELIVERY_MAX_ATTEMPTS)); + this.deliveryScheduler = options.deliveryScheduler ?? globalThis; this.budgetFreshTtlMs = positiveIntegerOr(options.budgetFreshTtlMs, parsePositiveIntegerEnv("AGENTBRIDGE_BUDGET_FRESH_TTL_SEC", DEFAULT_BUDGET_FRESH_TTL_MS / 1000) * 1000); this.wallNow = options.wallNow ?? (() => Date.now()); this.server = new Server({ name: "agentbridge", version: "0.1.0" }, { @@ -14279,22 +14294,34 @@ class ClaudeAdapter extends EventEmitter { } async pushNotification(message) { this.log(`pushNotification (instance=${this.instanceId}, msgId=${message.id}, len=${message.content.length})`); - if (!this.rememberDelivery(message)) + const delivery = this.rememberDelivery(message); + if (!delivery) return; - await this.pushViaChannel(message); + const queued = this.queueFallbackMessage(delivery); + if (!queued) { + this.deliveredMessageIds.delete(delivery.sourceMessageId); + } + if (queued && !delivery.resumeId) { + this.armDeliveryRetry(delivery, 1); + } + await this.pushViaChannel(delivery, queued); } - async pushViaChannel(message) { + async pushViaChannel(message, admitted = true) { const deliveryAttemptId = `codex_msg_${this.notificationIdPrefix}_${++this.notificationSeq}`; const ts = new Date(message.timestamp).toISOString(); try { await this.server.notification({ method: "notifications/claude/channel", params: { - content: message.content, + content: this.channelContent(message, admitted), meta: { chat_id: this.sessionId, message_id: message.id, + source_message_id: message.originalSourceMessageId, + ...message.sourceMessageId !== message.originalSourceMessageId ? { dedupe_source_message_id: message.sourceMessageId } : {}, delivery_attempt_id: deliveryAttemptId, + ack_required: admitted, + ...admitted ? { ack_tool: message.resumeId ? "ack_resume" : "ack_messages" } : {}, user: "Codex", user_id: "codex", ts, @@ -14305,43 +14332,108 @@ class ClaudeAdapter extends EventEmitter { }); this.log(`Pushed notification: ${message.id} (attempt=${deliveryAttemptId})`); } catch (e) { - this.log(`Push notification failed: ${e.message}`); - this.queueFallbackMessage(message); + this.log(`Push notification failed: ${e.message} (message remains in mailbox)`); + } + } + channelContent(message, admitted = true) { + if (message.resumeId) + return message.content; + if (!admitted) { + return `[AgentBridge oversized delivery id: ${message.id}. This message exceeded the mailbox size ` + `bound and is NOT retained: it cannot be recovered via get_messages and must not be ` + `acknowledged. If you already processed a message with this exact content, do not repeat the work.] + +` + message.content; } + const ackIds = JSON.stringify([message.id]); + return `[AgentBridge delivery id: ${message.id}. If this ID is already being processed or was processed, ` + `do not repeat the work. After fully processing this message, ` + `call ack_messages with ack_ids ${ackIds}. Do not acknowledge before processing.] + +` + message.content; } - rememberDelivery(message) { + rememberDelivery(message, originalSourceMessageId) { + const sourceMessageId = normalizeDeliveryId(message.id); + const originalSourceId = originalSourceMessageId ?? sourceMessageId; + if (sourceMessageId !== message.id) { + this.log(`WARNING: normalized unsafe Codex message id to ${sourceMessageId}`); + message = { ...message, id: sourceMessageId }; + } const now = this.monotonicNow(); + const fingerprint = deliveryFingerprint(message); + const activeLogicalMessage = this.pendingMessages.find((pending) => pending.originalSourceMessageId === originalSourceId && deliveryFingerprint(pending) === fingerprint); + if (activeLogicalMessage) { + this.deliveredMessageIds.delete(activeLogicalMessage.sourceMessageId); + this.deliveredMessageIds.set(activeLogicalMessage.sourceMessageId, { seenAt: now, fingerprint }); + this.enforceDedupeCapacity(); + this.log(`Duplicate active Codex message suppressed (msgId=${sourceMessageId}, source=${message.source}, ` + `instance=${this.instanceId})`); + return null; + } + const active = this.pendingMessages.find((pending) => pending.sourceMessageId === sourceMessageId); + if (active) { + const activeFingerprint = deliveryFingerprint(active); + this.deliveredMessageIds.delete(sourceMessageId); + this.deliveredMessageIds.set(sourceMessageId, { seenAt: now, fingerprint: activeFingerprint }); + this.enforceDedupeCapacity(); + return this.preserveIdCollision(message, fingerprint, originalSourceId); + } this.pruneDeliveredMessageIds(now); - if (this.deliveredMessageIds.has(message.id)) { - this.deliveredMessageIds.delete(message.id); - this.deliveredMessageIds.set(message.id, now); - this.log(`Duplicate Codex message suppressed (msgId=${message.id}, source=${message.source}, ` + `instance=${this.instanceId})`); - return false; + const previous = this.deliveredMessageIds.get(sourceMessageId); + if (previous) { + this.deliveredMessageIds.delete(sourceMessageId); + this.deliveredMessageIds.set(sourceMessageId, { seenAt: now, fingerprint: previous.fingerprint }); + this.enforceDedupeCapacity(); + if (previous.fingerprint === fingerprint) { + this.log(`Duplicate Codex message suppressed (msgId=${message.id}, source=${message.source}, ` + `instance=${this.instanceId})`); + return null; + } + return this.preserveIdCollision(message, fingerprint, originalSourceId); } - this.deliveredMessageIds.set(message.id, now); + this.deliveredMessageIds.set(sourceMessageId, { seenAt: now, fingerprint }); + this.enforceDedupeCapacity(); + return { + ...message, + id: this.allocateDeliveryId(sourceMessageId), + sourceMessageId, + originalSourceMessageId: originalSourceId + }; + } + enforceDedupeCapacity() { while (this.deliveredMessageIds.size > this.dedupeCapacity) { const oldest = this.deliveredMessageIds.keys().next().value; if (oldest === undefined) break; this.deliveredMessageIds.delete(oldest); } - return true; + } + allocateDeliveryId(sourceMessageId) { + const suffix = `_delivery_${this.notificationIdPrefix}_${++this.deliverySeq}`; + return `${sourceMessageId.slice(0, 512 - suffix.length)}${suffix}`; + } + preserveIdCollision(message, fingerprint, originalSourceMessageId) { + const suffix = `_collision_${fingerprint.slice(0, 12)}`; + const collisionId = `${message.id.slice(0, 512 - suffix.length)}${suffix}`; + this.log(`WARNING: conflicting Codex message id ${message.id}; preserving the later payload as ${collisionId}`); + return this.rememberDelivery({ ...message, id: collisionId }, originalSourceMessageId); } pruneDeliveredMessageIds(now) { - for (const [id, seenAt] of this.deliveredMessageIds) { - if (now - seenAt <= this.dedupeTtlMs) + for (const [id, record3] of this.deliveredMessageIds) { + if (now - record3.seenAt <= this.dedupeTtlMs) break; this.deliveredMessageIds.delete(id); } } queueFallbackMessage(message) { + if (!("sourceMessageId" in message)) { + message = { + ...message, + sourceMessageId: message.id, + originalSourceMessageId: message.id + }; + } const messageBytes = utf8ByteLength(message.content); if (messageBytes > this.maxBufferedBytes) { this.oversizedMessageCount++; this.oversizedMessageBytes += messageBytes; this.oversizedMessageSourceCounts[message.source] = (this.oversizedMessageSourceCounts[message.source] ?? 0) + 1; this.log(`Fallback queue omitted oversized ${message.source} message ` + `(${formatBytes(messageBytes)} > ${formatBytes(this.maxBufferedBytes)}; ` + `total oversized: ${this.oversizedMessageCount})`); - return; + return false; } let dropped = 0; while (this.pendingMessages.length >= this.maxBufferedMessages || this.pendingMessageBytes + messageBytes > this.maxBufferedBytes) { @@ -14349,6 +14441,8 @@ class ClaudeAdapter extends EventEmitter { const droppedBytes = this.pendingMessageByteSizes.shift() ?? 0; if (!droppedMessage) break; + this.cancelDeliveryRetry(droppedMessage.id); + this.deliveredMessageIds.delete(droppedMessage.sourceMessageId); this.pendingMessageBytes = Math.max(0, this.pendingMessageBytes - droppedBytes); this.droppedMessageCount++; dropped++; @@ -14360,18 +14454,114 @@ class ClaudeAdapter extends EventEmitter { this.pendingMessageByteSizes.push(messageBytes); this.pendingMessageBytes += messageBytes; this.log(`Queued fallback message (${this.pendingMessages.length} pending, ` + `${formatBytes(this.pendingMessageBytes)} buffered, instance=${this.instanceId})`); + return true; + } + hasPendingMessage(messageId) { + return this.pendingMessages.some((message) => message.id === messageId); + } + armDeliveryRetry(message, attempts) { + this.cancelDeliveryRetry(message.id); + if (!this.hasPendingMessage(message.id)) + return; + if (attempts >= this.deliveryMaxAttempts) + return; + const exponent = Math.max(0, attempts - 1); + const delayMs = Math.min(this.deliveryRetryBaseMs * 2 ** exponent, 2147483647); + const entry = { message, attempts }; + this.deliveryRetries.set(message.id, entry); + entry.timer = this.deliveryScheduler.setTimeout(() => { + delete entry.timer; + this.retryPendingDelivery(entry); + }, delayMs); + entry.timer?.unref?.(); + } + async retryPendingDelivery(entry) { + const current = this.deliveryRetries.get(entry.message.id); + if (current !== entry || !this.hasPendingMessage(entry.message.id)) { + if (current === entry) + this.deliveryRetries.delete(entry.message.id); + return; + } + const nextAttempt = entry.attempts + 1; + this.log(`Retrying unacknowledged Channel delivery: ${entry.message.id} (attempt=${nextAttempt})`); + this.armDeliveryRetry(entry.message, nextAttempt); + await this.pushViaChannel(entry.message); + if (nextAttempt >= this.deliveryMaxAttempts && this.hasPendingMessage(entry.message.id)) { + this.log(`Channel delivery unacknowledged after ${nextAttempt} attempt(s): ${entry.message.id}; ` + "message remains available via get_messages"); + } } - drainMessages() { + cancelDeliveryRetry(messageId) { + const entry = this.deliveryRetries.get(messageId); + if (!entry) + return; + if (entry.timer !== undefined) { + this.deliveryScheduler.clearTimeout(entry.timer); + } + this.deliveryRetries.delete(messageId); + } + acknowledgeMessages(messageIds, acknowledgeResumeControl = true) { + const requested = [...new Set(messageIds)]; + const requestedSet = new Set(requested); + const resumeIds = new Set(this.pendingMessages.filter((message) => requestedSet.has(message.id) && message.resumeId).map((message) => message.resumeId)); + for (const message of this.pendingMessages) { + if (message.resumeId && resumeIds.has(message.resumeId)) { + requestedSet.add(message.id); + } + } + const acknowledged = []; + const remainingMessages = []; + const remainingSizes = []; + for (let i = 0;i < this.pendingMessages.length; i++) { + const message = this.pendingMessages[i]; + const bytes = this.pendingMessageByteSizes[i] ?? utf8ByteLength(message.content); + if (requestedSet.has(message.id)) { + acknowledged.push(message.id); + this.pendingMessageBytes = Math.max(0, this.pendingMessageBytes - bytes); + this.cancelDeliveryRetry(message.id); + } else { + remainingMessages.push(message); + remainingSizes.push(bytes); + } + } + this.pendingMessages = remainingMessages; + this.pendingMessageByteSizes = remainingSizes; + if (acknowledgeResumeControl) { + for (const resumeId of resumeIds) { + if (this.resumeAckHandler) { + this.resumeAckHandler(resumeId, "resumed"); + } else { + this.log(`Resume mailbox message acknowledged without a daemon ACK handler (resume_id=${resumeId})`); + } + } + } + const acknowledgedSet = new Set(acknowledged); + const unknown3 = requested.filter((id) => !acknowledgedSet.has(id)); + if (acknowledged.length > 0 || unknown3.length > 0) { + this.log(`ack_messages (instance=${this.instanceId}, acknowledged=${acknowledged.length}, ` + `unknown=${unknown3.length}, pending=${this.pendingMessages.length})`); + } + return { acknowledged, unknown: unknown3 }; + } + acknowledgeResume(resumeId) { + const ids = this.pendingMessages.filter((message) => message.resumeId === resumeId).map((message) => message.id); + return this.acknowledgeMessages(ids, false).acknowledged.length; + } + drainMessages(ackIds = []) { + const ackResult = ackIds.length > 0 ? this.acknowledgeMessages(ackIds) : { acknowledged: [], unknown: [] }; this.log(`get_messages called (instance=${this.instanceId}, pending=${this.pendingMessages.length}, ` + `bytes=${this.pendingMessageBytes}, dropped=${this.droppedMessageCount}, oversized=${this.oversizedMessageCount})`); if (this.pendingMessages.length === 0 && this.droppedMessageCount === 0 && this.oversizedMessageCount === 0) { + if (ackResult.acknowledged.length > 0 || ackResult.unknown.length > 0) { + return { + content: [{ + type: "text", + text: formatAckResult(ackResult) + " No unacknowledged messages from Codex." + }] + }; + } return { content: [{ type: "text", text: "No new messages from Codex." }] }; } - const messages = this.pendingMessages; - this.pendingMessages = []; - this.pendingMessageByteSizes = []; - this.pendingMessageBytes = 0; + const messages = [...this.pendingMessages]; const dropped = this.droppedMessageCount; this.droppedMessageCount = 0; const oversizedSourceCounts = this.oversizedMessageSourceCounts; @@ -14393,7 +14583,7 @@ class ClaudeAdapter extends EventEmitter { const formatted = messages.map((msg, i) => { const ts = new Date(msg.timestamp).toISOString(); return `--- -[${i + 1}] ${ts} +[${i + 1}] ${ts} [id: ${msg.id}] Codex: ${msg.content}`; }).join(` @@ -14401,14 +14591,20 @@ Codex: ${msg.content}`; const noticeText = notices.map((notice) => `WARNING: ${notice}`).join(` `); const parts2 = []; + if (ackResult.acknowledged.length > 0 || ackResult.unknown.length > 0) { + parts2.push(formatAckResult(ackResult)); + } if (count > 0) { - parts2.push(`[${count} new message${count > 1 ? "s" : ""} from Codex] + parts2.push(`[${count} unacknowledged message${count > 1 ? "s" : ""} from Codex] chat_id: ${this.sessionId}`); } if (noticeText) parts2.push(noticeText); if (formatted) parts2.push(formatted); + if (messages.length > 0) { + parts2.push(`After fully processing these messages, call ack_messages with ack_ids: ` + JSON.stringify(messages.map((message) => message.id))); + } this.log(`get_messages returning ${count} message(s) ` + `(instance=${this.instanceId}, dropped=${dropped}, oversized=${oversized}, oversizedBytes=${oversizedBytes})`); return { content: [ @@ -14461,13 +14657,37 @@ chat_id: ${this.sessionId}`); }, { name: "get_messages", - description: "Check for new messages from Codex. Call this after sending a reply or when you expect a response from Codex.", + description: "Return all unacknowledged Codex messages in stable order. Messages remain until ack_messages confirms their stable IDs. Optionally acknowledge IDs from a previous result with ack_ids before reading the remaining mailbox.", inputSchema: { type: "object", - properties: {}, + properties: { + ack_ids: { + type: "array", + items: { type: "string", minLength: 1, maxLength: 512 }, + maxItems: this.ackIdsCap, + description: "Optional stable message IDs from a previous get_messages result to acknowledge before returning the remaining mailbox." + } + }, required: [] } }, + { + name: "ack_messages", + description: "Acknowledge Codex messages only after fully processing them. Works for messages received through Channel push or get_messages. Removes only the requested stable IDs and cancels their retries.", + inputSchema: { + type: "object", + properties: { + ack_ids: { + type: "array", + items: { type: "string", minLength: 1, maxLength: 512 }, + minItems: 1, + maxItems: this.ackIdsCap, + description: "Stable message IDs to acknowledge, from Channel meta.message_id or get_messages [id: ...] labels." + } + }, + required: ["ack_ids"] + } + }, { name: "get_budget", description: "Check both agents' subscription quota usage (Claude + Codex): 5h/weekly window percentages, drift between the two sides, joint-pause state and model/effort tier recommendation.", @@ -14504,7 +14724,13 @@ chat_id: ${this.sessionId}`); return this.handleReply(args); } if (name === "get_messages") { - return this.drainMessages(); + const parsed = parseAckIds(args?.ack_ids, false, this.ackIdsCap); + if (!parsed.ok) + return ackIdsError(parsed.error); + return this.drainMessages(parsed.ids); + } + if (name === "ack_messages") { + return this.handleAckMessages(args); } if (name === "get_budget") { return this.handleGetBudget(); @@ -14518,6 +14744,15 @@ chat_id: ${this.sessionId}`); }; }); } + handleAckMessages(args) { + const parsed = parseAckIds(args?.ack_ids, true, this.ackIdsCap); + if (!parsed.ok) + return ackIdsError(parsed.error); + const result = this.acknowledgeMessages(parsed.ids); + return { + content: [{ type: "text", text: formatAckResult(result) }] + }; + } async handleAckResume(args) { const resumeIdRaw = args?.resume_id; if (typeof resumeIdRaw !== "string" || resumeIdRaw.length === 0) { @@ -14549,8 +14784,12 @@ chat_id: ${this.sessionId}`); } this.log(`ack_resume received (resume_id=${resumeIdRaw}, status=${status}, instance=${this.instanceId})`); this.resumeAckHandler(resumeIdRaw, status); + const mailboxAcknowledged = this.acknowledgeResume(resumeIdRaw); return { - content: [{ type: "text", text: `Resume acknowledged (resume_id=${resumeIdRaw}, status=${status}).` }] + content: [{ + type: "text", + text: `Resume acknowledged (resume_id=${resumeIdRaw}, status=${status}, ` + `mailbox_messages=${mailboxAcknowledged}).` + }] }; } async handleGetBudget() { @@ -14653,7 +14892,7 @@ chat_id: ${this.sessionId}`); responseText = "Reply sent to Codex as a new turn (any turn still running was interrupted first; if it had already finished, your message was simply injected)."; } if (pending > 0) { - responseText += ` Note: ${pending} unread Codex message${pending > 1 ? "s" : ""} already waiting \u2014 call get_messages to read them.`; + responseText += ` Note: ${pending} unacknowledged Codex message${pending > 1 ? "s" : ""} in the mailbox \u2014 ` + "call get_messages if any are unprocessed, and acknowledge processed IDs with ack_messages."; } return { content: [{ type: "text", text: responseText }] @@ -14663,6 +14902,51 @@ chat_id: ${this.sessionId}`); this.logger.log(msg); } } +function parseAckIds(value, required2, maxItems = 100) { + if (value === undefined) { + return required2 ? { ok: false, error: "missing required parameter 'ack_ids'" } : { ok: true, ids: [] }; + } + if (!Array.isArray(value)) { + return { ok: false, error: "ack_ids must be an array of message ID strings" }; + } + if (required2 && value.length === 0) { + return { ok: false, error: "ack_ids must contain at least one message ID" }; + } + if (value.length > maxItems) { + return { ok: false, error: `ack_ids has ${value.length} items; maximum is ${maxItems}` }; + } + for (const id of value) { + if (typeof id !== "string" || id.length === 0 || id.length > 512) { + return { ok: false, error: "each ack_ids item must be a non-empty string of at most 512 characters" }; + } + } + return { ok: true, ids: value }; +} +function ackIdsError(error2) { + return { + content: [{ type: "text", text: `Error: ${error2}.` }], + isError: true + }; +} +function formatAckResult(result) { + const parts2 = [`Acknowledged ${result.acknowledged.length} message${result.acknowledged.length === 1 ? "" : "s"}.`]; + if (result.acknowledged.length > 0) { + parts2.push(`IDs: ${JSON.stringify(result.acknowledged)}.`); + } + if (result.unknown.length > 0) { + parts2.push(`Already acknowledged or unknown IDs: ${JSON.stringify(result.unknown)}.`); + } + return parts2.join(" "); +} +function deliveryFingerprint(message) { + return createHash("sha256").update(JSON.stringify([message.source, message.content, message.resumeId ?? null])).digest("hex"); +} +function normalizeDeliveryId(id) { + if (id.length > 0 && id.length <= 512 && /^[A-Za-z0-9._:-]+$/.test(id)) + return id; + const digest = createHash("sha256").update(id).digest("hex").slice(0, 32); + return `agentbridge_${digest}`; +} function parsePositiveIntegerEnv(name, fallback) { return positiveIntegerOr(parseInt(process.env[name] ?? "", 10), fallback); } @@ -14707,10 +14991,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("816c8f8d8b5a", "source") }); function sameRuntimeContract(a, b) { if (!a || !b) @@ -16150,7 +16434,7 @@ import { unlinkSync as unlinkSync4, writeFileSync as writeFileSync3 } from "fs"; -import { createHash, randomUUID as randomUUID3 } from "crypto"; +import { createHash as createHash2, randomUUID as randomUUID3 } from "crypto"; import { basename as basename2, join as join3, resolve, sep } from "path"; var PAIR_BASE_PORT = 4500; var PAIR_SLOT_STRIDE = 10; @@ -16175,7 +16459,7 @@ function derivePairId(cwd, name) { } catch { real = cwd; } - const hash = createHash("sha256").update(real).update("\x00").update(name.toLowerCase()).digest("hex").slice(0, 8); + const hash = createHash2("sha256").update(real).update("\x00").update(name.toLowerCase()).digest("hex").slice(0, 8); const slug = name.replace(/[^A-Za-z0-9._-]+/g, "-").replace(/^-+|-+$/g, "").slice(0, 32) || "pair"; return `${slug}-${hash}`; } @@ -16573,7 +16857,7 @@ daemonClient.on("status", (status) => { "\uD83E\uDD1D Codex has connected via AgentBridge.", "You are now in a multi-agent collaboration session.", "When you receive a complex task, propose a division of labor to Codex.", - "Use `reply` to send messages and `get_messages` to check for responses." + "Use `reply` to send messages and `get_messages` to check for responses; acknowledge processed message IDs with `ack_messages`." ].join(` `))); } diff --git a/plugins/agentbridge/server/daemon.js b/plugins/agentbridge/server/daemon.js index 5388db2..63c5732 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("816c8f8d8b5a", "source") }); function daemonStatusBuildInfo() { return { ...BUILD_INFO }; diff --git a/src/bridge.ts b/src/bridge.ts index abde48c..840fc0b 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -185,7 +185,7 @@ daemonClient.on("status", (status) => { "🤝 Codex has connected via AgentBridge.", "You are now in a multi-agent collaboration session.", "When you receive a complex task, propose a division of labor to Codex.", - "Use `reply` to send messages and `get_messages` to check for responses.", + "Use `reply` to send messages and `get_messages` to check for responses; acknowledge processed message IDs with `ack_messages`.", ].join("\n"), )); } diff --git a/src/claude-adapter.ts b/src/claude-adapter.ts index 3d6c1d3..fc5952a 100644 --- a/src/claude-adapter.ts +++ b/src/claude-adapter.ts @@ -1,9 +1,9 @@ /** * Claude Code MCP Server — Push Message Transport * - * Delivery is always push (real-time notifications/claude/channel). When a - * push fails, the message falls back to an in-memory queue drained by the - * get_messages tool — a per-message fallback, not a configurable mode. + * Every logical message enters an in-memory, explicitly acknowledged mailbox + * before notifications/claude/channel is attempted. Channel push is a bounded + * retry latency optimization; get_messages is the at-least-once recovery path. * (The old AGENTBRIDGE_MODE=pull delivery mode was removed: it could not wake * an idle session, which silently broke the budget RESUME chain.) * @@ -19,7 +19,7 @@ import { CallToolRequestSchema, } from "@modelcontextprotocol/sdk/types.js"; import { EventEmitter } from "node:events"; -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { performance } from "node:perf_hooks"; import { createProcessLogger, type ProcessLogger } from "./process-log"; import { StateDirResolver } from "./state-dir"; @@ -42,6 +42,12 @@ export interface ClaudeAdapterOptions { dedupeTtlMs?: number; /** Monotonic milliseconds for internal dedupe TTL; defaults to performance.now(). */ now?: () => number; + /** Delay before the first no-ack Channel retry. Defaults to 60000 ms. */ + deliveryRetryBaseMs?: number; + /** Total Channel attempts, including the initial push. Defaults to 3. */ + deliveryMaxAttempts?: number; + /** Timer seam for deterministic delivery-retry tests. */ + deliveryScheduler?: DeliveryScheduler; /** * Freshness TTL (ms) for the get_budget tool: when the cached snapshot is older * than this, get_budget asks the daemon for a fresh read-only refresh before @@ -56,14 +62,42 @@ const DEFAULT_MAX_BUFFERED_MESSAGES = 100; const DEFAULT_MAX_BUFFERED_BYTES = 4 * 1024 * 1024; const DEFAULT_DEDUPE_CAPACITY = 2048; const DEFAULT_DEDUPE_TTL_MS = 20 * 60 * 1000; +const DEFAULT_DELIVERY_RETRY_BASE_MS = 60 * 1000; +const DEFAULT_DELIVERY_MAX_ATTEMPTS = 3; const DEFAULT_BUDGET_FRESH_TTL_MS = 25 * 1000; +export interface DeliveryScheduler { + setTimeout(callback: () => void, delayMs: number): unknown; + clearTimeout(handle: unknown): void; +} + +interface DeliveredMessageRecord { + seenAt: number; + fingerprint: string; +} + +interface MailboxMessage extends BridgeMessage { + /** Daemon/source ID before the adapter allocates a unique ACK generation. */ + sourceMessageId: string; + /** Original normalized source ID, retained when a conflict uses an alias. */ + originalSourceMessageId: string; +} + +interface PendingDeliveryRetry { + message: MailboxMessage; + attempts: number; + timer?: unknown; +} + export const CLAUDE_INSTRUCTIONS = [ "Codex is an AI coding agent (OpenAI) running in a separate session on the same machine.", "", "## Message delivery", "Messages from Codex arrive as tags (push).", - "If a push fails, the message is queued — call get_messages to drain the fallback queue.", + "Every message is queued before push. Channel delivery is a latency optimization, not proof of receipt.", + "A repeated delivery ID is the same logical message. Never repeat completed work for an ID you already processed.", + "After fully processing a pushed message, call ack_messages with its meta.message_id. Do not acknowledge before processing.", + "If a push is missed, call get_messages. It returns the same stable message IDs until ack_messages confirms them.", "", "## Collaboration roles", "Default roles in this setup:", @@ -81,7 +115,7 @@ export const CLAUDE_INSTRUCTIONS = [ "", "## How to interact", "- Use the reply tool to send messages back to Codex — pass chat_id back.", - "- Use the get_messages tool to check for pending messages from Codex.", + "- Use the get_messages tool to check for pending messages from Codex, then call ack_messages only for IDs you fully processed.", "- After sending a reply, call get_messages to check for responses.", "- When the user asks about Codex status or progress, call get_messages.", "", @@ -99,6 +133,7 @@ export const CLAUDE_INSTRUCTIONS = [ export class ClaudeAdapter extends EventEmitter { private server: Server; private notificationSeq = 0; + private deliverySeq = 0; private sessionId: string; private readonly notificationIdPrefix: string; private readonly instanceId: string; @@ -112,12 +147,16 @@ export class ClaudeAdapter extends EventEmitter { private readonly logFile: string; private readonly logger: ProcessLogger; - // Push transport with a per-message fallback queue (drained by get_messages). - private pendingMessages: BridgeMessage[] = []; + // Authoritative in-memory mailbox. Messages enter before Channel push and + // remain until explicitly acknowledged or observably evicted by a bound. + private pendingMessages: MailboxMessage[] = []; private pendingMessageByteSizes: number[] = []; private pendingMessageBytes = 0; private readonly maxBufferedMessages: number; private readonly maxBufferedBytes: number; + /** ack_ids batch cap; never below the mailbox capacity so the drain + * epilogue's "ack all pending IDs" instruction is always executable. */ + private readonly ackIdsCap: number; private droppedMessageCount = 0; private oversizedMessageCount = 0; private oversizedMessageBytes = 0; @@ -125,7 +164,11 @@ export class ClaudeAdapter extends EventEmitter { private readonly dedupeCapacity: number; private readonly dedupeTtlMs: number; private readonly monotonicNow: () => number; - private deliveredMessageIds = new Map(); + private deliveredMessageIds = new Map(); + private readonly deliveryRetryBaseMs: number; + private readonly deliveryMaxAttempts: number; + private readonly deliveryScheduler: DeliveryScheduler; + private deliveryRetries = new Map(); // Latest budget snapshot, fed by bridge from DaemonStatus.budget broadcasts. private budgetSnapshot: BudgetSnapshot | null = null; @@ -163,9 +206,19 @@ export class ClaudeAdapter extends EventEmitter { options.maxBufferedBytes, parsePositiveIntegerEnv("AGENTBRIDGE_MAX_BUFFERED_BYTES", DEFAULT_MAX_BUFFERED_BYTES), ); + this.ackIdsCap = Math.max(100, this.maxBufferedMessages); this.dedupeCapacity = positiveIntegerOr(options.dedupeCapacity, DEFAULT_DEDUPE_CAPACITY); this.dedupeTtlMs = positiveIntegerOr(options.dedupeTtlMs, DEFAULT_DEDUPE_TTL_MS); this.monotonicNow = options.now ?? (() => performance.now()); + this.deliveryRetryBaseMs = positiveIntegerOr( + options.deliveryRetryBaseMs, + parsePositiveIntegerEnv("AGENTBRIDGE_DELIVERY_RETRY_BASE_MS", DEFAULT_DELIVERY_RETRY_BASE_MS), + ); + this.deliveryMaxAttempts = positiveIntegerOr( + options.deliveryMaxAttempts, + parsePositiveIntegerEnv("AGENTBRIDGE_DELIVERY_MAX_ATTEMPTS", DEFAULT_DELIVERY_MAX_ATTEMPTS), + ); + this.deliveryScheduler = options.deliveryScheduler ?? globalThis; this.budgetFreshTtlMs = positiveIntegerOr( options.budgetFreshTtlMs, parsePositiveIntegerEnv("AGENTBRIDGE_BUDGET_FRESH_TTL_SEC", DEFAULT_BUDGET_FRESH_TTL_MS / 1000) * 1000, @@ -233,11 +286,27 @@ export class ClaudeAdapter extends EventEmitter { async pushNotification(message: BridgeMessage) { this.log(`pushNotification (instance=${this.instanceId}, msgId=${message.id}, len=${message.content.length})`); - if (!this.rememberDelivery(message)) return; - await this.pushViaChannel(message); + const delivery = this.rememberDelivery(message); + if (!delivery) return; + + // Queue before the first await. A resolved Channel write only means the + // transport accepted bytes; the mailbox remains authoritative until ACK. + const queued = this.queueFallbackMessage(delivery); + if (!queued) { + // Do not leave a dedupe tombstone for content the mailbox could not + // admit. A source replay must get another observable delivery attempt. + this.deliveredMessageIds.delete(delivery.sourceMessageId); + } + + // Budget resume already owns a dedicated ACK/retry state machine. General + // messages schedule in arrival order before any transport promise settles. + if (queued && !delivery.resumeId) { + this.armDeliveryRetry(delivery, 1); + } + await this.pushViaChannel(delivery, queued); } - private async pushViaChannel(message: BridgeMessage) { + private async pushViaChannel(message: MailboxMessage, admitted = true) { const deliveryAttemptId = `codex_msg_${this.notificationIdPrefix}_${++this.notificationSeq}`; const ts = new Date(message.timestamp).toISOString(); @@ -245,11 +314,19 @@ export class ClaudeAdapter extends EventEmitter { await this.server.notification({ method: "notifications/claude/channel", params: { - content: message.content, + content: this.channelContent(message, admitted), meta: { chat_id: this.sessionId, message_id: message.id, + source_message_id: message.originalSourceMessageId, + ...(message.sourceMessageId !== message.originalSourceMessageId + ? { dedupe_source_message_id: message.sourceMessageId } + : {}), delivery_attempt_id: deliveryAttemptId, + // An unadmitted (oversized) message is best-effort only: it is not + // in the mailbox, so an ACK contract would be a lie — see channelContent. + ack_required: admitted, + ...(admitted ? { ack_tool: message.resumeId ? "ack_resume" : "ack_messages" } : {}), user: "Codex", user_id: "codex", ts, @@ -263,43 +340,146 @@ export class ClaudeAdapter extends EventEmitter { }); this.log(`Pushed notification: ${message.id} (attempt=${deliveryAttemptId})`); } catch (e: any) { - this.log(`Push notification failed: ${e.message}`); - this.queueFallbackMessage(message); + this.log(`Push notification failed: ${e.message} (message remains in mailbox)`); + } + } + + private channelContent(message: MailboxMessage, admitted = true): string { + if (message.resumeId) return message.content; + if (!admitted) { + return ( + `[AgentBridge oversized delivery id: ${message.id}. This message exceeded the mailbox size ` + + `bound and is NOT retained: it cannot be recovered via get_messages and must not be ` + + `acknowledged. If you already processed a message with this exact content, do not repeat the work.]\n\n` + + message.content + ); } + const ackIds = JSON.stringify([message.id]); + return ( + `[AgentBridge delivery id: ${message.id}. If this ID is already being processed or was processed, ` + + `do not repeat the work. After fully processing this message, ` + + `call ack_messages with ack_ids ${ackIds}. Do not acknowledge before processing.]\n\n` + + message.content + ); } - private rememberDelivery(message: BridgeMessage): boolean { + private rememberDelivery( + message: BridgeMessage, + originalSourceMessageId?: string, + ): MailboxMessage | null { + const sourceMessageId = normalizeDeliveryId(message.id); + const originalSourceId = originalSourceMessageId ?? sourceMessageId; + if (sourceMessageId !== message.id) { + this.log(`WARNING: normalized unsafe Codex message id to ${sourceMessageId}`); + message = { ...message, id: sourceMessageId }; + } const now = this.monotonicNow(); - this.pruneDeliveredMessageIds(now); - if (this.deliveredMessageIds.has(message.id)) { - // Refresh recency so duplicate bursts do not evict a still-active key. - this.deliveredMessageIds.delete(message.id); - this.deliveredMessageIds.set(message.id, now); + const fingerprint = deliveryFingerprint(message); + + // A conflict alias must still dedupe replays addressed to the original + // source ID. Match original source + payload before considering the current + // alias key, otherwise an expired original tombstone can admit the same + // logical payload twice under two different ACK generations. + const activeLogicalMessage = this.pendingMessages.find( + (pending) => pending.originalSourceMessageId === originalSourceId && + deliveryFingerprint(pending) === fingerprint, + ); + if (activeLogicalMessage) { + this.deliveredMessageIds.delete(activeLogicalMessage.sourceMessageId); + this.deliveredMessageIds.set(activeLogicalMessage.sourceMessageId, { seenAt: now, fingerprint }); + this.enforceDedupeCapacity(); this.log( - `Duplicate Codex message suppressed (msgId=${message.id}, source=${message.source}, ` + + `Duplicate active Codex message suppressed (msgId=${sourceMessageId}, source=${message.source}, ` + `instance=${this.instanceId})`, ); - return false; + return null; + } + + // An unacknowledged mailbox entry must remain authoritative even after the + // bounded dedupe cache expires or evicts its tombstone. Otherwise the same + // source ID could be queued twice and one ACK would accidentally delete two + // different logical messages. + const active = this.pendingMessages.find((pending) => pending.sourceMessageId === sourceMessageId); + if (active) { + const activeFingerprint = deliveryFingerprint(active); + this.deliveredMessageIds.delete(sourceMessageId); + this.deliveredMessageIds.set(sourceMessageId, { seenAt: now, fingerprint: activeFingerprint }); + this.enforceDedupeCapacity(); + return this.preserveIdCollision(message, fingerprint, originalSourceId); + } + + this.pruneDeliveredMessageIds(now); + const previous = this.deliveredMessageIds.get(sourceMessageId); + if (previous) { + // Refresh recency so duplicate bursts do not evict a still-active key. + this.deliveredMessageIds.delete(sourceMessageId); + this.deliveredMessageIds.set(sourceMessageId, { seenAt: now, fingerprint: previous.fingerprint }); + this.enforceDedupeCapacity(); + if (previous.fingerprint === fingerprint) { + this.log( + `Duplicate Codex message suppressed (msgId=${message.id}, source=${message.source}, ` + + `instance=${this.instanceId})`, + ); + return null; + } + + // One ACK key cannot safely represent two different payloads. Preserve + // the later payload under a deterministic collision ID and warn loudly. + return this.preserveIdCollision(message, fingerprint, originalSourceId); } - this.deliveredMessageIds.set(message.id, now); + this.deliveredMessageIds.set(sourceMessageId, { seenAt: now, fingerprint }); + this.enforceDedupeCapacity(); + return { + ...message, + id: this.allocateDeliveryId(sourceMessageId), + sourceMessageId, + originalSourceMessageId: originalSourceId, + }; + } + + private enforceDedupeCapacity(): void { while (this.deliveredMessageIds.size > this.dedupeCapacity) { const oldest = this.deliveredMessageIds.keys().next().value; if (oldest === undefined) break; this.deliveredMessageIds.delete(oldest); } - return true; + } + + private allocateDeliveryId(sourceMessageId: string): string { + const suffix = `_delivery_${this.notificationIdPrefix}_${++this.deliverySeq}`; + return `${sourceMessageId.slice(0, 512 - suffix.length)}${suffix}`; + } + + private preserveIdCollision( + message: BridgeMessage, + fingerprint: string, + originalSourceMessageId: string, + ): MailboxMessage | null { + const suffix = `_collision_${fingerprint.slice(0, 12)}`; + const collisionId = `${message.id.slice(0, 512 - suffix.length)}${suffix}`; + this.log( + `WARNING: conflicting Codex message id ${message.id}; preserving the later payload as ${collisionId}`, + ); + return this.rememberDelivery({ ...message, id: collisionId }, originalSourceMessageId); } private pruneDeliveredMessageIds(now: number): void { - for (const [id, seenAt] of this.deliveredMessageIds) { - if (now - seenAt <= this.dedupeTtlMs) break; + for (const [id, record] of this.deliveredMessageIds) { + if (now - record.seenAt <= this.dedupeTtlMs) break; this.deliveredMessageIds.delete(id); } } - /** Per-message fallback when a push fails; drained by the get_messages tool. */ - private queueFallbackMessage(message: BridgeMessage) { + /** Insert into the authoritative mailbox before Channel push. */ + private queueFallbackMessage(message: MailboxMessage | BridgeMessage): boolean { + if (!("sourceMessageId" in message)) { + message = { + ...message, + sourceMessageId: message.id, + originalSourceMessageId: message.id, + }; + } const messageBytes = utf8ByteLength(message.content); if (messageBytes > this.maxBufferedBytes) { this.oversizedMessageCount++; @@ -311,7 +491,7 @@ export class ClaudeAdapter extends EventEmitter { `(${formatBytes(messageBytes)} > ${formatBytes(this.maxBufferedBytes)}; ` + `total oversized: ${this.oversizedMessageCount})`, ); - return; + return false; } let dropped = 0; @@ -322,6 +502,8 @@ export class ClaudeAdapter extends EventEmitter { const droppedMessage = this.pendingMessages.shift(); const droppedBytes = this.pendingMessageByteSizes.shift() ?? 0; if (!droppedMessage) break; + this.cancelDeliveryRetry(droppedMessage.id); + this.deliveredMessageIds.delete(droppedMessage.sourceMessageId); this.pendingMessageBytes = Math.max(0, this.pendingMessageBytes - droppedBytes); this.droppedMessageCount++; dropped++; @@ -341,26 +523,155 @@ export class ClaudeAdapter extends EventEmitter { `Queued fallback message (${this.pendingMessages.length} pending, ` + `${formatBytes(this.pendingMessageBytes)} buffered, instance=${this.instanceId})`, ); + return true; } // ── get_messages ─────────────────────────────────────────── - private drainMessages(): { content: Array<{ type: "text"; text: string }> } { + private hasPendingMessage(messageId: string): boolean { + return this.pendingMessages.some((message) => message.id === messageId); + } + + private armDeliveryRetry(message: MailboxMessage, attempts: number): void { + this.cancelDeliveryRetry(message.id); + if (!this.hasPendingMessage(message.id)) return; + if (attempts >= this.deliveryMaxAttempts) return; + + const exponent = Math.max(0, attempts - 1); + const delayMs = Math.min(this.deliveryRetryBaseMs * (2 ** exponent), 2_147_483_647); + const entry: PendingDeliveryRetry = { message, attempts }; + // Register before installing the timer so a scheduler seam that fires the + // callback synchronously still finds (and can advance) this entry. + this.deliveryRetries.set(message.id, entry); + entry.timer = this.deliveryScheduler.setTimeout(() => { + delete entry.timer; + void this.retryPendingDelivery(entry); + }, delayMs); + (entry.timer as { unref?: () => void } | undefined)?.unref?.(); + } + + private async retryPendingDelivery(entry: PendingDeliveryRetry): Promise { + const current = this.deliveryRetries.get(entry.message.id); + if (current !== entry || !this.hasPendingMessage(entry.message.id)) { + if (current === entry) this.deliveryRetries.delete(entry.message.id); + return; + } + + const nextAttempt = entry.attempts + 1; + this.log(`Retrying unacknowledged Channel delivery: ${entry.message.id} (attempt=${nextAttempt})`); + // Install the next timer before awaiting transport. This keeps retry + // scheduling in FIFO callback order even when Channel promises settle out + // of order. ACK still cancels the newly installed timer by delivery ID. + this.armDeliveryRetry(entry.message, nextAttempt); + await this.pushViaChannel(entry.message); + + if (nextAttempt >= this.deliveryMaxAttempts && this.hasPendingMessage(entry.message.id)) { + this.log( + `Channel delivery unacknowledged after ${nextAttempt} attempt(s): ${entry.message.id}; ` + + "message remains available via get_messages", + ); + } + } + + private cancelDeliveryRetry(messageId: string): void { + const entry = this.deliveryRetries.get(messageId); + if (!entry) return; + if (entry.timer !== undefined) { + this.deliveryScheduler.clearTimeout(entry.timer); + } + this.deliveryRetries.delete(messageId); + } + + private acknowledgeMessages( + messageIds: string[], + acknowledgeResumeControl = true, + ): { acknowledged: string[]; unknown: string[] } { + const requested = [...new Set(messageIds)]; + const requestedSet = new Set(requested); + const resumeIds = new Set( + this.pendingMessages + .filter((message) => requestedSet.has(message.id) && message.resumeId) + .map((message) => message.resumeId!), + ); + + // A budget resume is one logical directive with potentially several + // delivery-attempt IDs. Acknowledging any attempt retires every queued + // sibling so a later pull cannot repeat an already processed directive. + for (const message of this.pendingMessages) { + if (message.resumeId && resumeIds.has(message.resumeId)) { + requestedSet.add(message.id); + } + } + const acknowledged: string[] = []; + const remainingMessages: MailboxMessage[] = []; + const remainingSizes: number[] = []; + + for (let i = 0; i < this.pendingMessages.length; i++) { + const message = this.pendingMessages[i]!; + const bytes = this.pendingMessageByteSizes[i] ?? utf8ByteLength(message.content); + if (requestedSet.has(message.id)) { + acknowledged.push(message.id); + this.pendingMessageBytes = Math.max(0, this.pendingMessageBytes - bytes); + this.cancelDeliveryRetry(message.id); + } else { + remainingMessages.push(message); + remainingSizes.push(bytes); + } + } + + this.pendingMessages = remainingMessages; + this.pendingMessageByteSizes = remainingSizes; + if (acknowledgeResumeControl) { + for (const resumeId of resumeIds) { + if (this.resumeAckHandler) { + this.resumeAckHandler(resumeId, "resumed"); + } else { + this.log(`Resume mailbox message acknowledged without a daemon ACK handler (resume_id=${resumeId})`); + } + } + } + const acknowledgedSet = new Set(acknowledged); + const unknown = requested.filter((id) => !acknowledgedSet.has(id)); + if (acknowledged.length > 0 || unknown.length > 0) { + this.log( + `ack_messages (instance=${this.instanceId}, acknowledged=${acknowledged.length}, ` + + `unknown=${unknown.length}, pending=${this.pendingMessages.length})`, + ); + } + return { acknowledged, unknown }; + } + + private acknowledgeResume(resumeId: string): number { + const ids = this.pendingMessages + .filter((message) => message.resumeId === resumeId) + .map((message) => message.id); + return this.acknowledgeMessages(ids, false).acknowledged.length; + } + + private drainMessages(ackIds: string[] = []): { content: Array<{ type: "text"; text: string }> } { + const ackResult = ackIds.length > 0 + ? this.acknowledgeMessages(ackIds) + : { acknowledged: [] as string[], unknown: [] as string[] }; this.log( `get_messages called (instance=${this.instanceId}, pending=${this.pendingMessages.length}, ` + `bytes=${this.pendingMessageBytes}, dropped=${this.droppedMessageCount}, oversized=${this.oversizedMessageCount})`, ); if (this.pendingMessages.length === 0 && this.droppedMessageCount === 0 && this.oversizedMessageCount === 0) { + if (ackResult.acknowledged.length > 0 || ackResult.unknown.length > 0) { + return { + content: [{ + type: "text" as const, + text: formatAckResult(ackResult) + " No unacknowledged messages from Codex.", + }], + }; + } return { content: [{ type: "text" as const, text: "No new messages from Codex." }], }; } - // Snapshot and clear atomically to avoid issues with concurrent writes - const messages = this.pendingMessages; - this.pendingMessages = []; - this.pendingMessageByteSizes = []; - this.pendingMessageBytes = 0; + // Snapshot without clearing. Only an explicit ACK may remove a message. + const messages = [...this.pendingMessages]; const dropped = this.droppedMessageCount; this.droppedMessageCount = 0; const oversizedSourceCounts = this.oversizedMessageSourceCounts; @@ -391,17 +702,26 @@ export class ClaudeAdapter extends EventEmitter { const formatted = messages .map((msg, i) => { const ts = new Date(msg.timestamp).toISOString(); - return `---\n[${i + 1}] ${ts}\nCodex: ${msg.content}`; + return `---\n[${i + 1}] ${ts} [id: ${msg.id}]\nCodex: ${msg.content}`; }) .join("\n\n"); const noticeText = notices.map((notice) => `WARNING: ${notice}`).join("\n"); const parts: string[] = []; + if (ackResult.acknowledged.length > 0 || ackResult.unknown.length > 0) { + parts.push(formatAckResult(ackResult)); + } if (count > 0) { - parts.push(`[${count} new message${count > 1 ? "s" : ""} from Codex]\nchat_id: ${this.sessionId}`); + parts.push(`[${count} unacknowledged message${count > 1 ? "s" : ""} from Codex]\nchat_id: ${this.sessionId}`); } if (noticeText) parts.push(noticeText); if (formatted) parts.push(formatted); + if (messages.length > 0) { + parts.push( + `After fully processing these messages, call ack_messages with ack_ids: ` + + JSON.stringify(messages.map((message) => message.id)), + ); + } this.log( `get_messages returning ${count} message(s) ` + @@ -461,13 +781,38 @@ export class ClaudeAdapter extends EventEmitter { { name: "get_messages", description: - "Check for new messages from Codex. Call this after sending a reply or when you expect a response from Codex.", + "Return all unacknowledged Codex messages in stable order. Messages remain until ack_messages confirms their stable IDs. Optionally acknowledge IDs from a previous result with ack_ids before reading the remaining mailbox.", inputSchema: { type: "object" as const, - properties: {}, + properties: { + ack_ids: { + type: "array", + items: { type: "string", minLength: 1, maxLength: 512 }, + maxItems: this.ackIdsCap, + description: "Optional stable message IDs from a previous get_messages result to acknowledge before returning the remaining mailbox.", + }, + }, required: [], }, }, + { + name: "ack_messages", + description: + "Acknowledge Codex messages only after fully processing them. Works for messages received through Channel push or get_messages. Removes only the requested stable IDs and cancels their retries.", + inputSchema: { + type: "object" as const, + properties: { + ack_ids: { + type: "array", + items: { type: "string", minLength: 1, maxLength: 512 }, + minItems: 1, + maxItems: this.ackIdsCap, + description: "Stable message IDs to acknowledge, from Channel meta.message_id or get_messages [id: ...] labels.", + }, + }, + required: ["ack_ids"], + }, + }, { name: "get_budget", description: @@ -510,7 +855,13 @@ export class ClaudeAdapter extends EventEmitter { } if (name === "get_messages") { - return this.drainMessages(); + const parsed = parseAckIds((args as Record | undefined)?.ack_ids, false, this.ackIdsCap); + if (!parsed.ok) return ackIdsError(parsed.error); + return this.drainMessages(parsed.ids); + } + + if (name === "ack_messages") { + return this.handleAckMessages(args as Record); } if (name === "get_budget") { @@ -528,6 +879,16 @@ export class ClaudeAdapter extends EventEmitter { }); } + private handleAckMessages(args: Record) { + const parsed = parseAckIds(args?.ack_ids, true, this.ackIdsCap); + if (!parsed.ok) return ackIdsError(parsed.error); + + const result = this.acknowledgeMessages(parsed.ids); + return { + content: [{ type: "text" as const, text: formatAckResult(result) }], + }; + } + /** * Handle ack_resume (PR4): Claude acknowledging a system_budget_resume * directive. Mirrors handleReply's boundary checks, but routes through the @@ -573,9 +934,15 @@ export class ClaudeAdapter extends EventEmitter { this.log(`ack_resume received (resume_id=${resumeIdRaw}, status=${status}, instance=${this.instanceId})`); this.resumeAckHandler(resumeIdRaw, status); + const mailboxAcknowledged = this.acknowledgeResume(resumeIdRaw); return { - content: [{ type: "text" as const, text: `Resume acknowledged (resume_id=${resumeIdRaw}, status=${status}).` }], + content: [{ + type: "text" as const, + text: + `Resume acknowledged (resume_id=${resumeIdRaw}, status=${status}, ` + + `mailbox_messages=${mailboxAcknowledged}).`, + }], }; } @@ -716,7 +1083,8 @@ export class ClaudeAdapter extends EventEmitter { responseText = "Reply sent to Codex as a new turn (any turn still running was interrupted first; if it had already finished, your message was simply injected)."; } if (pending > 0) { - responseText += ` Note: ${pending} unread Codex message${pending > 1 ? "s" : ""} already waiting \u2014 call get_messages to read them.`; + responseText += ` Note: ${pending} unacknowledged Codex message${pending > 1 ? "s" : ""} in the mailbox \u2014 ` + + "call get_messages if any are unprocessed, and acknowledge processed IDs with ack_messages."; } return { @@ -729,6 +1097,63 @@ export class ClaudeAdapter extends EventEmitter { } } +type AckIdsParseResult = + | { ok: true; ids: string[] } + | { ok: false; error: string }; + +function parseAckIds(value: unknown, required: boolean, maxItems = 100): AckIdsParseResult { + if (value === undefined) { + return required + ? { ok: false, error: "missing required parameter 'ack_ids'" } + : { ok: true, ids: [] }; + } + if (!Array.isArray(value)) { + return { ok: false, error: "ack_ids must be an array of message ID strings" }; + } + if (required && value.length === 0) { + return { ok: false, error: "ack_ids must contain at least one message ID" }; + } + if (value.length > maxItems) { + return { ok: false, error: `ack_ids has ${value.length} items; maximum is ${maxItems}` }; + } + for (const id of value) { + if (typeof id !== "string" || id.length === 0 || id.length > 512) { + return { ok: false, error: "each ack_ids item must be a non-empty string of at most 512 characters" }; + } + } + return { ok: true, ids: value as string[] }; +} + +function ackIdsError(error: string) { + return { + content: [{ type: "text" as const, text: `Error: ${error}.` }], + isError: true, + }; +} + +function formatAckResult(result: { acknowledged: string[]; unknown: string[] }): string { + const parts = [`Acknowledged ${result.acknowledged.length} message${result.acknowledged.length === 1 ? "" : "s"}.`]; + if (result.acknowledged.length > 0) { + parts.push(`IDs: ${JSON.stringify(result.acknowledged)}.`); + } + if (result.unknown.length > 0) { + parts.push(`Already acknowledged or unknown IDs: ${JSON.stringify(result.unknown)}.`); + } + return parts.join(" "); +} + +function deliveryFingerprint(message: BridgeMessage): string { + return createHash("sha256") + .update(JSON.stringify([message.source, message.content, message.resumeId ?? null])) + .digest("hex"); +} + +function normalizeDeliveryId(id: string): string { + if (id.length > 0 && id.length <= 512 && /^[A-Za-z0-9._:-]+$/.test(id)) return id; + const digest = createHash("sha256").update(id).digest("hex").slice(0, 32); + return `agentbridge_${digest}`; +} + function parsePositiveIntegerEnv(name: string, fallback: number): number { return positiveIntegerOr(parseInt(process.env[name] ?? "", 10), fallback); } diff --git a/src/collaboration-content.ts b/src/collaboration-content.ts index 0b8913c..5b2e627 100644 --- a/src/collaboration-content.ts +++ b/src/collaboration-content.ts @@ -37,8 +37,8 @@ You are working in a **multi-agent environment** powered by AgentBridge. Another AI agent (Codex, by OpenAI) is available in a parallel session on this machine. ### Communication mechanism -- **Claude → Codex**: Use the AgentBridge MCP tools (\`reply\` / \`get_messages\`) — these are yours only. -- **Codex → Claude**: Codex has no symmetric tool. The bridge transparently intercepts Codex's normal output and forwards it to you as push notifications (if a push fails, drain the fallback queue with \`get_messages\`). +- **Claude → Codex**: Use the AgentBridge MCP tools (\`reply\` / \`get_messages\` / \`ack_messages\`) — these are yours only. +- **Codex → Claude**: Codex has no symmetric tool. The bridge transparently intercepts Codex's normal output and forwards it to you as push notifications. Every message is also held in an acknowledged mailbox: \`get_messages\` re-reads pending messages **without removing them**, and only \`ack_messages\` (with the stable message IDs) removes the ones you have fully processed. Acknowledge after processing — a repeated delivery ID is the same logical message, so never repeat completed work for an ID you already handled. - If Codex ever complains it can't find a "send-to-Claude" API, remind it that its side is transparent — it just writes a reply and you'll see it. ### When to collaborate vs. work solo @@ -74,7 +74,7 @@ Another AI agent (Claude, by Anthropic) is available in a parallel session on th AgentBridge is a **transparent proxy** on your side. You do **not** have a tool to "send a message to Claude". - **Codex → Claude**: Just write your normal response. The bridge intercepts your \`agentMessage\` output and forwards it to Claude automatically. No tool call needed. -- **Claude → Codex**: Claude uses its own MCP tools (\`reply\` / \`get_messages\`). Those messages arrive in your session as new user turns — you'll see them like any other user input. +- **Claude → Codex**: Claude uses its own MCP tools (\`reply\` / \`get_messages\` / \`ack_messages\`). Those messages arrive in your session as new user turns — you'll see them like any other user input. **Do not** search the AgentBridge source for a Codex-side "send" / "reply" / "sendToClaude" API — it does not exist, and looking for it wastes turns. If you catch yourself thinking "I need to find how to message Claude", stop and just write your reply as normal text. diff --git a/src/unit-test/claude-adapter-resume.test.ts b/src/unit-test/claude-adapter-resume.test.ts index 616aeb0..4170d9d 100644 --- a/src/unit-test/claude-adapter-resume.test.ts +++ b/src/unit-test/claude-adapter-resume.test.ts @@ -33,14 +33,13 @@ function makeCodexMessage(content: string, extra: Record = {}) } describe("ack_resume MCP tool — registration", () => { - test("ack_resume appears in ListTools as the fourth tool", async () => { + test("ack_resume appears in ListTools after the general message ACK tool", async () => { const adapter = new ClaudeAdapter() as any; const result = await listTools(adapter); const names = result.tools.map((t: any) => t.name); - // Snapshot: reply, get_messages, get_budget were the prior three. - expect(names).toEqual(["reply", "get_messages", "get_budget", "ack_resume"]); - expect(result.tools).toHaveLength(4); + expect(names).toEqual(["reply", "get_messages", "ack_messages", "get_budget", "ack_resume"]); + expect(result.tools).toHaveLength(5); }); test("ack_resume schema requires resume_id and constrains status enum", async () => { @@ -274,8 +273,9 @@ describe("channel resume push — meta.resume_id", () => { // Neither was dropped by LRU dedup — both delivered. expect(notifications).toHaveLength(2); - expect(notifications[0].params.meta.message_id).toBe(attempt0); - expect(notifications[1].params.meta.message_id).toBe(attempt1); + expect(notifications[0].params.meta.source_message_id).toBe(attempt0); + expect(notifications[1].params.meta.source_message_id).toBe(attempt1); + expect(notifications[0].params.meta.message_id).not.toBe(notifications[1].params.meta.message_id); // Both carry the SAME stable resumeId so Claude's single ack correlates. expect(notifications[0].params.meta.resume_id).toBe(rid); expect(notifications[1].params.meta.resume_id).toBe(rid); diff --git a/src/unit-test/message-delivery.test.ts b/src/unit-test/message-delivery.test.ts index 16270db..1eefb94 100644 --- a/src/unit-test/message-delivery.test.ts +++ b/src/unit-test/message-delivery.test.ts @@ -10,6 +10,12 @@ function createAdapter( dedupeCapacity?: number; dedupeTtlMs?: number; now?: () => number; + deliveryRetryBaseMs?: number; + deliveryMaxAttempts?: number; + deliveryScheduler?: { + setTimeout(callback: () => void, delayMs: number): unknown; + clearTimeout(handle: unknown): void; + }; }, ): any { const origMode = process.env.AGENTBRIDGE_MODE; @@ -60,7 +66,13 @@ function withMockedChannel(adapter: any, mode: "success" | "fail" = "success") { return notifications; } -describe("Push-only delivery: AGENTBRIDGE_MODE is ignored", () => { +function ackSourceMessage(adapter: any, sourceMessageId: string) { + const entry = adapter.pendingMessages.find((message: any) => message.sourceMessageId === sourceMessageId); + if (!entry) throw new Error(`No pending delivery for source ID ${sourceMessageId}`); + return adapter.handleAckMessages({ ack_ids: [entry.id] }); +} + +describe("Channel delivery with an authoritative mailbox: AGENTBRIDGE_MODE is ignored", () => { // Pull mode was removed (it could not wake an idle session and silently // broke the budget RESUME chain). Any legacy env value must be ignored. test("delivers via channel when AGENTBRIDGE_MODE is unset", async () => { @@ -68,7 +80,7 @@ describe("Push-only delivery: AGENTBRIDGE_MODE is ignored", () => { const notifications = withMockedChannel(adapter); await adapter.pushNotification(makeBridgeMessage("normal push")); expect(notifications).toHaveLength(1); - expect(adapter.pendingMessages).toHaveLength(0); + expect(adapter.pendingMessages).toHaveLength(1); }); test('legacy AGENTBRIDGE_MODE="pull" still delivers via channel', async () => { @@ -76,7 +88,7 @@ describe("Push-only delivery: AGENTBRIDGE_MODE is ignored", () => { const notifications = withMockedChannel(adapter); await adapter.pushNotification(makeBridgeMessage("ignored pull env")); expect(notifications).toHaveLength(1); - expect(adapter.pendingMessages).toHaveLength(0); + expect(adapter.pendingMessages).toHaveLength(1); }); test("any other AGENTBRIDGE_MODE value is equally ignored", async () => { @@ -84,7 +96,7 @@ describe("Push-only delivery: AGENTBRIDGE_MODE is ignored", () => { const notifications = withMockedChannel(adapter); await adapter.pushNotification(makeBridgeMessage("ignored auto env")); expect(notifications).toHaveLength(1); - expect(adapter.pendingMessages).toHaveLength(0); + expect(adapter.pendingMessages).toHaveLength(1); }); test("legacy warning is construction-time only — never per message", async () => { @@ -163,7 +175,7 @@ describe("Message delivery: fallback queue", () => { expect(adapter.oversizedMessageBytes).toBe(9); }); - test("push meta uses BridgeMessage.id as message_id and a separate delivery_attempt_id", async () => { + test("push meta separates source, stable ACK, and delivery-attempt IDs", async () => { const adapter = createAdapter(); const notifications: any[] = []; @@ -181,8 +193,10 @@ describe("Message delivery: fallback queue", () => { const firstMeta = notifications[0].params.meta; const secondMeta = notifications[1].params.meta; - expect(firstMeta.message_id).toBe("codex-item-1"); - expect(secondMeta.message_id).toBe("codex-item-2"); + expect(firstMeta.source_message_id).toBe("codex-item-1"); + expect(secondMeta.source_message_id).toBe("codex-item-2"); + expect(firstMeta.message_id).toMatch(/^codex-item-1_delivery_[a-f0-9]{12}_1$/); + expect(secondMeta.message_id).toMatch(/^codex-item-2_delivery_[a-f0-9]{12}_2$/); expect(firstMeta.delivery_attempt_id).toMatch(/^codex_msg_[a-f0-9]{12}_1$/); expect(secondMeta.delivery_attempt_id).toMatch(/^codex_msg_[a-f0-9]{12}_2$/); expect(firstMeta.delivery_attempt_id.replace(/_1$/, "")).toBe(secondMeta.delivery_attempt_id.replace(/_2$/, "")); @@ -196,12 +210,12 @@ describe("Message delivery: fallback queue", () => { adapter.logger = { log: (msg: string) => logs.push(msg) }; await adapter.pushNotification(makeBridgeMessage("first delivery", 1705312200000, "same-id")); - await adapter.pushNotification(makeBridgeMessage("duplicate delivery", 1705312201000, "same-id")); + await adapter.pushNotification(makeBridgeMessage("first delivery", 1705312201000, "same-id")); expect(notifications).toHaveLength(1); - expect(notifications[0].params.content).toBe("first delivery"); - expect(adapter.pendingMessages).toHaveLength(0); - expect(logs.some((line) => line.includes("Duplicate Codex message suppressed") && line.includes("same-id"))).toBe(true); + expect(notifications[0].params.content).toEndWith("first delivery"); + expect(adapter.pendingMessages).toHaveLength(1); + expect(logs.some((line) => line.includes("Duplicate active Codex message suppressed") && line.includes("same-id"))).toBe(true); }); test("pushNotification does not enqueue fallback twice for the same BridgeMessage.id", async () => { @@ -209,7 +223,7 @@ describe("Message delivery: fallback queue", () => { withMockedChannel(adapter, "fail"); await adapter.pushNotification(makeBridgeMessage("queued once", 1705312200000, "fallback-id")); - await adapter.pushNotification(makeBridgeMessage("queued duplicate", 1705312201000, "fallback-id")); + await adapter.pushNotification(makeBridgeMessage("queued once", 1705312201000, "fallback-id")); expect(adapter.pendingMessages.map((m: any) => m.content)).toEqual(["queued once"]); expect(adapter.pendingMessageBytes).toBe(Buffer.byteLength("queued once", "utf8")); @@ -220,15 +234,15 @@ describe("Message delivery: fallback queue", () => { const notifications = withMockedChannel(adapter); await adapter.pushNotification(makeBridgeMessage("first a", 1705312200000, "id-a")); + ackSourceMessage(adapter, "id-a"); await adapter.pushNotification(makeBridgeMessage("first b", 1705312201000, "id-b")); + ackSourceMessage(adapter, "id-b"); await adapter.pushNotification(makeBridgeMessage("first c", 1705312202000, "id-c")); + ackSourceMessage(adapter, "id-c"); await adapter.pushNotification(makeBridgeMessage("second a", 1705312203000, "id-a")); - expect(notifications.map((n) => n.params.content)).toEqual([ - "first a", - "first b", - "first c", - "second a", + expect(notifications.map((n) => n.params.content.split("\n\n").at(-1))).toEqual([ + "first a", "first b", "first c", "second a", ]); }); @@ -241,13 +255,14 @@ describe("Message delivery: fallback queue", () => { }); const notifications = withMockedChannel(adapter); - await adapter.pushNotification(makeBridgeMessage("first", 1705312200000, "ttl-id")); + await adapter.pushNotification(makeBridgeMessage("same payload", 1705312200000, "ttl-id")); + ackSourceMessage(adapter, "ttl-id"); now += 50; - await adapter.pushNotification(makeBridgeMessage("duplicate within ttl", 1705312201000, "ttl-id")); + await adapter.pushNotification(makeBridgeMessage("same payload", 1705312201000, "ttl-id")); now += 101; - await adapter.pushNotification(makeBridgeMessage("after ttl", 1705312202000, "ttl-id")); + await adapter.pushNotification(makeBridgeMessage("same payload", 1705312202000, "ttl-id")); - expect(notifications.map((n) => n.params.content)).toEqual(["first", "after ttl"]); + expect(notifications).toHaveLength(2); }); test("pushNotification dedupe TTL ignores wall-clock jumps", async () => { @@ -260,10 +275,11 @@ describe("Message delivery: fallback queue", () => { const notifications = withMockedChannel(adapter); await adapter.pushNotification(makeBridgeMessage("first", 1705312200000, "wall-clock-id")); + ackSourceMessage(adapter, "wall-clock-id"); wallNow += 120_000; - await adapter.pushNotification(makeBridgeMessage("duplicate after wall jump", 1705312201000, "wall-clock-id")); + await adapter.pushNotification(makeBridgeMessage("first", 1705312201000, "wall-clock-id")); - expect(notifications.map((n) => n.params.content)).toEqual(["first"]); + expect(notifications).toHaveLength(1); } finally { Date.now = originalDateNow; } @@ -288,8 +304,11 @@ describe("Message delivery: fallback queue", () => { await adapter.pushNotification(makeBridgeMessage("live after recovery")); expect(notifications).toHaveLength(1); - expect(notifications[0].params.content).toBe("live after recovery"); - expect(adapter.pendingMessages.map((m: any) => m.content)).toEqual(["queued while push failed"]); + expect(notifications[0].params.content).toEndWith("live after recovery"); + expect(adapter.pendingMessages.map((m: any) => m.content)).toEqual([ + "queued while push failed", + "live after recovery", + ]); }); }); @@ -301,7 +320,7 @@ describe("Message delivery: drainMessages (get_messages)", () => { expect(result.content[0].text).toBe("No new messages from Codex."); }); - test("returns formatted messages and clears queue", () => { + test("returns stable IDs without clearing messages before acknowledgement", () => { const adapter = createAdapter(); const ts = 1705312200000; // fixed timestamp for deterministic output @@ -311,15 +330,16 @@ describe("Message delivery: drainMessages (get_messages)", () => { const result = adapter.drainMessages(); const text = result.content[0].text; - expect(text).toContain("[2 new messages from Codex]"); + expect(text).toContain("[2 unacknowledged messages from Codex]"); expect(text).toContain("chat_id:"); expect(text).toContain("[1]"); expect(text).toContain("first message"); expect(text).toContain("[2]"); expect(text).toContain("second message"); - // Queue should be cleared - expect(adapter.pendingMessages).toHaveLength(0); + expect(text).toContain(`[id: ${adapter.pendingMessages[0].id}]`); + expect(adapter.pendingMessages).toHaveLength(2); + adapter.handleAckMessages({ ack_ids: adapter.pendingMessages.map((message: any) => message.id) }); expect(adapter.getPendingMessageCount()).toBe(0); }); @@ -347,14 +367,14 @@ describe("Message delivery: drainMessages (get_messages)", () => { const result = adapter.drainMessages(); const text = result.content[0].text; - expect(text).toContain("[1 new message from Codex]"); + expect(text).toContain("[1 unacknowledged message from Codex]"); expect(text).toContain("1 older message"); expect(text).toContain("dropped due to fallback queue overflow"); expect(text).toContain("1 oversized message from Codex omitted (>10B)"); expect(text).not.toContain("12345"); expect(text).not.toContain("xxxxxxxxxxx"); - expect(adapter.pendingMessages).toHaveLength(0); - expect(adapter.pendingMessageBytes).toBe(0); + expect(adapter.pendingMessages).toHaveLength(1); + expect(adapter.pendingMessageBytes).toBe(6); expect(adapter.droppedMessageCount).toBe(0); expect(adapter.oversizedMessageCount).toBe(0); }); @@ -372,7 +392,7 @@ describe("Message delivery: drainMessages (get_messages)", () => { expect(text).toContain("1 oversized message from Codex omitted (>8B)"); }); - test("drainMessages reports no messages after clearing since-drain drop counters", () => { + test("drainMessages resets notices but repeats unacknowledged messages", () => { const adapter = createAdapter(undefined, { maxBufferedMessages: 1 }); adapter.queueFallbackMessage(makeBridgeMessage("first")); @@ -382,7 +402,8 @@ describe("Message delivery: drainMessages (get_messages)", () => { expect(firstDrain.content[0].text).toContain("dropped due to fallback queue overflow"); const secondDrain = adapter.drainMessages(); - expect(secondDrain.content[0].text).toBe("No new messages from Codex."); + expect(secondDrain.content[0].text).not.toContain("dropped due to fallback queue overflow"); + expect(secondDrain.content[0].text).toContain("second"); }); test("singular message uses correct grammar", () => { @@ -391,7 +412,7 @@ describe("Message delivery: drainMessages (get_messages)", () => { adapter.queueFallbackMessage(makeBridgeMessage("only one")); const result = adapter.drainMessages(); - expect(result.content[0].text).toContain("[1 new message from Codex]"); + expect(result.content[0].text).toContain("[1 unacknowledged message from Codex]"); }); }); @@ -407,8 +428,9 @@ describe("Message delivery: reply pending hint", () => { const text = result.content[0].text; expect(text).toContain("Reply sent to Codex."); - expect(text).toContain("2 unread Codex message"); + expect(text).toContain("2 unacknowledged Codex message"); expect(text).toContain("get_messages"); + expect(text).toContain("ack_messages"); }); test("handleReply has no hint when queue is empty", async () => { diff --git a/src/unit-test/reliable-mailbox.test.ts b/src/unit-test/reliable-mailbox.test.ts new file mode 100644 index 0000000..bf228d9 --- /dev/null +++ b/src/unit-test/reliable-mailbox.test.ts @@ -0,0 +1,638 @@ +import { describe, expect, test } from "bun:test"; +import { ClaudeAdapter, type DeliveryScheduler } from "../claude-adapter"; + +interface ScheduledTask { + callback: () => void; + delayMs: number; + cancelled: boolean; +} + +class FakeScheduler implements DeliveryScheduler { + readonly tasks: ScheduledTask[] = []; + + setTimeout(callback: () => void, delayMs: number): ScheduledTask { + const task = { callback, delayMs, cancelled: false }; + this.tasks.push(task); + return task; + } + + clearTimeout(handle: unknown): void { + (handle as ScheduledTask).cancelled = true; + } + + activeTasks(): ScheduledTask[] { + return this.tasks.filter((task) => !task.cancelled); + } + + async fireNext(): Promise { + const task = this.tasks.find((candidate) => !candidate.cancelled); + if (!task) throw new Error("No scheduled task to fire"); + task.cancelled = true; + task.callback(); + await Bun.sleep(0); + } +} + +function message(id: string, content: string, options: { resumeId?: string; timestamp?: number } = {}) { + return { + id, + source: "codex" as const, + content, + timestamp: options.timestamp ?? 1_705_312_200_000, + ...(options.resumeId ? { resumeId: options.resumeId } : {}), + }; +} + +function adapterWithChannel( + channel: (payload: any) => Promise = async () => {}, + options: Record = {}, +): { adapter: any; notifications: any[] } { + const notifications: any[] = []; + const adapter = new ClaudeAdapter(undefined, options) as any; + adapter.server.notification = async (payload: any) => { + notifications.push(payload); + await channel(payload); + }; + return { adapter, notifications }; +} + +function mailboxText(adapter: any, ackIds: string[] = []): string { + return adapter.drainMessages(ackIds).content[0].text; +} + +function deliveryIdFor(adapter: any, sourceMessageId: string): string { + const entry = adapter.pendingMessages.find((candidate: any) => candidate.sourceMessageId === sourceMessageId); + if (!entry) throw new Error(`No pending delivery for source ID ${sourceMessageId}`); + return entry.id; +} + +function pendingSourceIds(adapter: any): string[] { + return adapter.pendingMessages.map((entry: any) => entry.sourceMessageId); +} + +function callTool(adapter: any, name: string, args: Record = {}) { + const handler = adapter.server._requestHandlers.get("tools/call"); + return handler({ method: "tools/call", params: { name, arguments: args } }, {}); +} + +describe("Reliable mailbox issue 223 behavior", () => { + test("A: a silently ignored successful Channel write stays recoverable", async () => { + const { adapter, notifications } = adapterWithChannel(); + + await adapter.pushNotification(message("silent-1", "work result")); + const id = deliveryIdFor(adapter, "silent-1"); + + expect(notifications).toHaveLength(1); + expect(mailboxText(adapter)).toContain(`[id: ${id}]`); + expect(mailboxText(adapter)).toContain(`[id: ${id}]`); + expect(adapter.getPendingMessageCount()).toBe(1); + }); + + test("B: Channel delivery exposes an ACK path that prevents later pull duplication", async () => { + const scheduler = new FakeScheduler(); + const { adapter, notifications } = adapterWithChannel(async () => {}, { + deliveryScheduler: scheduler, + deliveryRetryBaseMs: 10, + }); + + await adapter.pushNotification(message("channel-1", "apply this result")); + const id = deliveryIdFor(adapter, "channel-1"); + const pushed = notifications[0]; + expect(pushed.params.meta.message_id).toBe(id); + expect(pushed.params.meta.source_message_id).toBe("channel-1"); + expect(pushed.params.meta.ack_tool).toBe("ack_messages"); + expect(pushed.params.content).toContain("call ack_messages"); + + const ack = await callTool(adapter, "ack_messages", { ack_ids: [id] }); + expect(ack.content[0].text).toContain("Acknowledged 1 message"); + expect(mailboxText(adapter)).toBe("No new messages from Codex."); + expect(scheduler.activeTasks()).toHaveLength(0); + }); + + test("B: an ACK during an in-flight Channel write cannot re-arm retry", async () => { + const scheduler = new FakeScheduler(); + let releaseChannel: () => void = () => {}; + const { adapter } = adapterWithChannel( + () => new Promise((resolve) => { releaseChannel = resolve; }), + { deliveryScheduler: scheduler, deliveryRetryBaseMs: 10 }, + ); + + const push = adapter.pushNotification(message("in-flight-1", "process while write waits")); + expect(adapter.getPendingMessageCount()).toBe(1); + await callTool(adapter, "ack_messages", { ack_ids: [deliveryIdFor(adapter, "in-flight-1")] }); + releaseChannel(); + await push; + + expect(adapter.getPendingMessageCount()).toBe(0); + expect(scheduler.activeTasks()).toHaveLength(0); + }); + + test("C: a throwing Channel push leaves exactly one retrievable entry", async () => { + const { adapter } = adapterWithChannel(async () => { + throw new Error("transport failed"); + }); + + await adapter.pushNotification(message("throw-1", "recover me")); + + expect(pendingSourceIds(adapter)).toEqual(["throw-1"]); + expect(mailboxText(adapter)).toContain("recover me"); + }); + + test("D: repeated polls are at least once and ACK removes only requested stable IDs", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("poll-1", "first")); + await adapter.pushNotification(message("poll-2", "second")); + const firstId = deliveryIdFor(adapter, "poll-1"); + const secondId = deliveryIdFor(adapter, "poll-2"); + + const first = mailboxText(adapter); + const second = mailboxText(adapter); + expect(second).toBe(first); + + const afterPartialAck = (await callTool(adapter, "get_messages", { ack_ids: [firstId] })).content[0].text; + expect(afterPartialAck).not.toContain(`[id: ${firstId}]`); + expect(afterPartialAck).toContain(`[id: ${secondId}]`); + expect(pendingSourceIds(adapter)).toEqual(["poll-2"]); + }); + + test("E: processing without a completed ACK leaves the message recoverable", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("lost-ack-1", "processed before crash")); + + expect(mailboxText(adapter)).toContain(`[id: ${deliveryIdFor(adapter, "lost-ack-1")}]`); + expect(adapter.getPendingMessageCount()).toBe(1); + }); + + test("E: a committed ACK with a lost response is idempotent when repeated", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("lost-response-1", "processed")); + const id = deliveryIdFor(adapter, "lost-response-1"); + + await callTool(adapter, "ack_messages", { ack_ids: [id] }); + const repeated = await callTool(adapter, "ack_messages", { ack_ids: [id] }); + + expect(repeated.isError).toBeUndefined(); + expect(repeated.content[0].text).toContain("Already acknowledged or unknown IDs"); + expect(mailboxText(adapter)).toBe("No new messages from Codex."); + }); +}); + +describe("Reliable mailbox restart and reconnect boundaries", () => { + test("F: a Claude MCP adapter restart loses the in-memory mailbox", async () => { + const { adapter: beforeRestart } = adapterWithChannel(); + await beforeRestart.pushNotification(message("restart-1", "ephemeral")); + + const { adapter: afterRestart } = adapterWithChannel(); + expect(beforeRestart.getPendingMessageCount()).toBe(1); + expect(afterRestart.getPendingMessageCount()).toBe(0); + }); + + test("F: a Claude Channel reconnect using the same adapter preserves the mailbox", async () => { + const { adapter } = adapterWithChannel(async () => { + throw new Error("disconnected"); + }); + await adapter.pushNotification(message("claude-reconnect-1", "waiting")); + + adapter.server = { notification: async () => {} }; + expect(mailboxText(adapter)).toContain("waiting"); + }); + + test("F: a daemon client replacement does not clear a live adapter mailbox", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("daemon-restart-1", "waiting")); + + adapter.setReplySender(async () => ({ success: true })); + adapter.setReplySender(async () => ({ success: true })); + expect(mailboxText(adapter)).toContain("waiting"); + }); + + test("F: a Codex reconnect does not clear completed messages held by the live adapter", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("codex-reconnect-1", "completed reply")); + + adapter.setReplySender(async () => ({ success: true })); + expect(mailboxText(adapter)).toContain("completed reply"); + }); +}); + +describe("Reliable mailbox ordering, concurrency, and dedupe", () => { + test("G: rapid arrivals enter the mailbox in invocation order", async () => { + const releases: Array<() => void> = []; + const { adapter } = adapterWithChannel( + () => new Promise((resolve) => releases.push(resolve)), + { deliveryMaxAttempts: 1 }, + ); + + const first = adapter.pushNotification(message("rapid-1", "first")); + const second = adapter.pushNotification(message("rapid-2", "second")); + const third = adapter.pushNotification(message("rapid-3", "third")); + expect(pendingSourceIds(adapter)).toEqual(["rapid-1", "rapid-2", "rapid-3"]); + + releases.reverse().forEach((release) => release()); + await Promise.all([first, second, third]); + expect(pendingSourceIds(adapter)).toEqual(["rapid-1", "rapid-2", "rapid-3"]); + }); + + test("G: ACK concurrent with a new arrival cannot delete the new ID", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("old-1", "old")); + const oldId = deliveryIdFor(adapter, "old-1"); + + const arrival = adapter.pushNotification(message("new-1", "new")); + adapter.handleAckMessages({ ack_ids: [oldId] }); + await arrival; + + expect(pendingSourceIds(adapter)).toEqual(["new-1"]); + }); + + test("G: active IDs remain protected after dedupe TTL and capacity eviction", async () => { + let now = 1_000; + const { adapter } = adapterWithChannel(async () => {}, { + dedupeCapacity: 1, + dedupeTtlMs: 10, + now: () => now, + }); + + await adapter.pushNotification(message("stable-1", "original")); + await adapter.pushNotification(message("other-1", "other")); + now += 100; + await adapter.pushNotification(message("stable-1", "original")); + await adapter.pushNotification(message("stable-1", "different")); + + expect(adapter.pendingMessages.filter((entry: any) => entry.sourceMessageId === "stable-1")).toHaveLength(1); + const collision = adapter.pendingMessages.find((entry: any) => entry.content === "different"); + expect(collision.sourceMessageId).toMatch(/^stable-1_collision_[a-f0-9]{12}$/); + + adapter.handleAckMessages({ ack_ids: [deliveryIdFor(adapter, "stable-1")] }); + expect(adapter.pendingMessages.map((entry: any) => entry.id)).toContain(collision.id); + }); + + test("G: a conflict alias suppresses an original-ID replay after TTL expiry", async () => { + let now = 1_000; + const { adapter, notifications } = adapterWithChannel(async () => {}, { + dedupeTtlMs: 10, + now: () => now, + deliveryMaxAttempts: 1, + }); + await adapter.pushNotification(message("alias-ttl", "payload A")); + const originalDeliveryId = deliveryIdFor(adapter, "alias-ttl"); + await adapter.pushNotification(message("alias-ttl", "payload B")); + adapter.handleAckMessages({ ack_ids: [originalDeliveryId] }); + + now += 100; + await adapter.pushNotification(message("alias-ttl", "payload B")); + + expect(adapter.pendingMessages.map((entry: any) => entry.content)).toEqual(["payload B"]); + expect(notifications).toHaveLength(2); + }); + + test("G: a conflict alias suppresses an original-ID replay after cache eviction", async () => { + const { adapter, notifications } = adapterWithChannel(async () => {}, { + dedupeCapacity: 1, + deliveryMaxAttempts: 1, + }); + await adapter.pushNotification(message("alias-capacity", "payload A")); + const originalDeliveryId = deliveryIdFor(adapter, "alias-capacity"); + await adapter.pushNotification(message("alias-capacity", "payload B")); + adapter.handleAckMessages({ ack_ids: [originalDeliveryId] }); + await adapter.pushNotification(message("alias-filler", "filler")); + adapter.handleAckMessages({ ack_ids: [deliveryIdFor(adapter, "alias-filler")] }); + + await adapter.pushNotification(message("alias-capacity", "payload B")); + + expect(adapter.pendingMessages.map((entry: any) => entry.content)).toEqual(["payload B"]); + expect(notifications).toHaveLength(3); + }); + + test("G: unsafe or overlong source IDs become stable ACK-safe IDs", async () => { + const { adapter, notifications } = adapterWithChannel(); + await adapter.pushNotification(message(`unsafe\n${"x".repeat(600)}`, "normalized")); + + const id = adapter.pendingMessages[0].id; + expect(adapter.pendingMessages[0].sourceMessageId).toMatch(/^agentbridge_[a-f0-9]{32}$/); + expect(id).toMatch(/^agentbridge_[a-f0-9]{32}_delivery_[a-f0-9]{12}_1$/); + expect(notifications[0].params.meta.message_id).toBe(id); + expect(adapter.handleAckMessages({ ack_ids: [id] }).isError).toBeUndefined(); + expect(adapter.getPendingMessageCount()).toBe(0); + }); + + test("G: a late ACK cannot delete a newer delivery generation after dedupe TTL", async () => { + let now = 1_000; + const { adapter } = adapterWithChannel(async () => {}, { + dedupeTtlMs: 10, + now: () => now, + deliveryMaxAttempts: 1, + }); + await adapter.pushNotification(message("reuse-id", "old payload")); + const oldDeliveryId = deliveryIdFor(adapter, "reuse-id"); + adapter.handleAckMessages({ ack_ids: [oldDeliveryId] }); + + now += 100; + await adapter.pushNotification(message("reuse-id", "new payload")); + const newDeliveryId = deliveryIdFor(adapter, "reuse-id"); + expect(newDeliveryId).not.toBe(oldDeliveryId); + + adapter.handleAckMessages({ ack_ids: [oldDeliveryId] }); + expect(adapter.pendingMessages.map((entry: any) => entry.content)).toEqual(["new payload"]); + }); + + test("G: a late ACK cannot delete a newer generation after dedupe capacity eviction", async () => { + const { adapter } = adapterWithChannel(async () => {}, { + dedupeCapacity: 1, + deliveryMaxAttempts: 1, + }); + await adapter.pushNotification(message("capacity-reuse", "old payload")); + const oldDeliveryId = deliveryIdFor(adapter, "capacity-reuse"); + adapter.handleAckMessages({ ack_ids: [oldDeliveryId] }); + await adapter.pushNotification(message("capacity-filler", "filler")); + adapter.handleAckMessages({ ack_ids: [deliveryIdFor(adapter, "capacity-filler")] }); + + await adapter.pushNotification(message("capacity-reuse", "new payload")); + adapter.handleAckMessages({ ack_ids: [oldDeliveryId] }); + + expect(adapter.pendingMessages.map((entry: any) => entry.content)).toEqual(["new payload"]); + }); + + test("G: an evicted in-flight push cannot replace a newer generation's retry", async () => { + const scheduler = new FakeScheduler(); + let releaseOld: () => void = () => {}; + const { adapter } = adapterWithChannel( + (payload) => payload.params.content.endsWith("old payload") + ? new Promise((resolve) => { releaseOld = resolve; }) + : Promise.resolve(), + { maxBufferedMessages: 1, deliveryScheduler: scheduler, deliveryRetryBaseMs: 10 }, + ); + + const oldPush = adapter.pushNotification(message("inflight-reuse", "old payload")); + const oldDeliveryId = adapter.pendingMessages[0].id; + await adapter.pushNotification(message("inflight-reuse", "new payload")); + const newDeliveryId = adapter.pendingMessages[0].id; + releaseOld(); + await oldPush; + + expect(newDeliveryId).not.toBe(oldDeliveryId); + expect(adapter.pendingMessages.map((entry: any) => entry.content)).toEqual(["new payload"]); + expect(adapter.deliveryRetries.has(oldDeliveryId)).toBe(false); + expect(adapter.deliveryRetries.get(newDeliveryId)?.message.content).toBe("new payload"); + }); + + test("G: retry attempts preserve Channel and mailbox FIFO order", async () => { + const scheduler = new FakeScheduler(); + const initialResolvers = new Map void>(); + const initialSeen = new Set(); + const { adapter, notifications } = adapterWithChannel((payload) => { + const sourceId = payload.params.meta.source_message_id as string; + if (initialSeen.has(sourceId)) return Promise.resolve(); + initialSeen.add(sourceId); + return new Promise((resolve) => initialResolvers.set(sourceId, resolve)); + }, { + deliveryScheduler: scheduler, + deliveryRetryBaseMs: 10, + deliveryMaxAttempts: 2, + }); + + const first = adapter.pushNotification(message("retry-order-1", "first")); + const second = adapter.pushNotification(message("retry-order-2", "second")); + initialResolvers.get("retry-order-2")!(); + initialResolvers.get("retry-order-1")!(); + await Promise.all([first, second]); + + await scheduler.fireNext(); + await scheduler.fireNext(); + + expect(pendingSourceIds(adapter)).toEqual(["retry-order-1", "retry-order-2"]); + expect(notifications.map((item) => item.params.meta.source_message_id)).toEqual([ + "retry-order-1", "retry-order-2", "retry-order-1", "retry-order-2", + ]); + }); + + test("G: a never-settling initial Channel promise does not prevent bounded retry scheduling", async () => { + const scheduler = new FakeScheduler(); + let release: () => void = () => {}; + const blocked = new Promise((resolve) => { release = resolve; }); + const { adapter, notifications } = adapterWithChannel(() => blocked, { + deliveryScheduler: scheduler, + deliveryRetryBaseMs: 10, + deliveryMaxAttempts: 2, + }); + + const initial = adapter.pushNotification(message("blocked-1", "still recoverable")); + expect(scheduler.activeTasks()).toHaveLength(1); + await scheduler.fireNext(); + expect(notifications).toHaveLength(2); + expect(scheduler.activeTasks()).toHaveLength(0); + + release(); + await initial; + }); + + test("G: separate pairs and Claude adapter instances do not share mailbox state", async () => { + const { adapter: pairOne } = adapterWithChannel(); + const { adapter: pairTwo } = adapterWithChannel(); + await pairOne.pushNotification(message("pair-1", "only pair one")); + + expect(pairOne.getPendingMessageCount()).toBe(1); + expect(pairTwo.getPendingMessageCount()).toBe(0); + }); +}); + +describe("Reliable mailbox retries and special messages", () => { + test("H: retry is bounded with exponential delays and retains the message after exhaustion", async () => { + const scheduler = new FakeScheduler(); + const { adapter, notifications } = adapterWithChannel(async () => {}, { + deliveryScheduler: scheduler, + deliveryRetryBaseMs: 25, + deliveryMaxAttempts: 3, + }); + await adapter.pushNotification(message("bounded-1", "keep me")); + + expect(scheduler.activeTasks().map((task) => task.delayMs)).toEqual([25]); + await scheduler.fireNext(); + expect(scheduler.activeTasks().map((task) => task.delayMs)).toEqual([50]); + await scheduler.fireNext(); + + expect(notifications).toHaveLength(3); + expect(scheduler.activeTasks()).toHaveLength(0); + expect(mailboxText(adapter)).toContain(`[id: ${deliveryIdFor(adapter, "bounded-1")}]`); + }); + + test("H: system messages use the same reliable mailbox", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("system_notice_1", "important system notice")); + + expect(mailboxText(adapter)).toContain("important system notice"); + }); + + test("H: an overflow-evicted ID can be recovered if the source replays it", async () => { + const { adapter, notifications } = adapterWithChannel(async () => {}, { maxBufferedMessages: 1 }); + await adapter.pushNotification(message("overflow-1", "first")); + await adapter.pushNotification(message("overflow-2", "second")); + await adapter.pushNotification(message("overflow-1", "first")); + + expect(notifications).toHaveLength(3); + expect(pendingSourceIds(adapter)).toEqual(["overflow-1"]); + }); + + test("H: an oversized ID is not tombstoned when its content cannot be admitted", async () => { + const { adapter, notifications } = adapterWithChannel(async () => {}, { maxBufferedBytes: 4 }); + await adapter.pushNotification(message("oversized-1", "12345")); + await adapter.pushNotification(message("oversized-1", "12345")); + + expect(notifications).toHaveLength(2); + expect(adapter.getPendingMessageCount()).toBe(0); + expect(mailboxText(adapter)).toContain("2 oversized messages"); + }); + + test("H: acknowledging one budget-resume attempt retires all siblings and stops daemon retries", async () => { + const { adapter } = adapterWithChannel(); + const resumeAcks: Array<{ id: string; status: string }> = []; + adapter.setResumeAckHandler((id: string, status: string) => resumeAcks.push({ id, status })); + + await adapter.pushNotification(message("resume-attempt-1", "resume", { resumeId: "resume-logical-1" })); + await adapter.pushNotification(message("resume-attempt-2", "resume", { resumeId: "resume-logical-1" })); + expect(adapter.getPendingMessageCount()).toBe(2); + + adapter.handleAckMessages({ ack_ids: [deliveryIdFor(adapter, "resume-attempt-1")] }); + expect(adapter.getPendingMessageCount()).toBe(0); + expect(resumeAcks).toEqual([{ id: "resume-logical-1", status: "resumed" }]); + }); + + test("H: ack_resume also retires every queued delivery attempt", async () => { + const { adapter } = adapterWithChannel(); + adapter.setResumeAckHandler(() => {}); + await adapter.pushNotification(message("resume-tool-1", "resume", { resumeId: "resume-logical-2" })); + await adapter.pushNotification(message("resume-tool-2", "resume", { resumeId: "resume-logical-2" })); + + const result = await adapter.handleAckResume({ resume_id: "resume-logical-2" }); + expect(result.content[0].text).toContain("mailbox_messages=2"); + expect(adapter.getPendingMessageCount()).toBe(0); + }); + + test("H: ACK input validation is bounded and unknown IDs are idempotent", () => { + const { adapter } = adapterWithChannel(); + expect(adapter.handleAckMessages({}).isError).toBe(true); + expect(adapter.handleAckMessages({ ack_ids: [] }).isError).toBe(true); + expect(adapter.handleAckMessages({ ack_ids: Array.from({ length: 101 }, (_, i) => `id-${i}`) }).isError).toBe(true); + + const unknown = adapter.handleAckMessages({ ack_ids: ["already-gone"] }); + expect(unknown.isError).toBeUndefined(); + expect(unknown.content[0].text).toContain("Already acknowledged or unknown IDs"); + }); + + test("I: default bounds and retry parameters match the documented contract", () => { + const adapter = new ClaudeAdapter() as any; + expect(adapter.maxBufferedMessages).toBe(100); + expect(adapter.maxBufferedBytes).toBe(4 * 1024 * 1024); + expect(adapter.dedupeCapacity).toBe(2048); + expect(adapter.dedupeTtlMs).toBe(20 * 60 * 1000); + expect(adapter.deliveryRetryBaseMs).toBe(60_000); + expect(adapter.deliveryMaxAttempts).toBe(3); + expect(adapter.ackIdsCap).toBe(100); + }); + + test("I: resume pushes carry ack_resume meta and unprefixed content", async () => { + const { adapter, notifications } = adapterWithChannel(); + await adapter.pushNotification(message("resume-shape-1", "resume now", { resumeId: "resume-shape" })); + + expect(notifications).toHaveLength(1); + const pushed = notifications[0]; + expect(pushed.params.meta.ack_tool).toBe("ack_resume"); + expect(pushed.params.meta.ack_required).toBe(true); + expect(pushed.params.meta.resume_id).toBe("resume-shape"); + expect(pushed.params.content).toBe("resume now"); + }); + + test("I: every Channel retry repeats the same stable delivery ID with a fresh attempt ID", async () => { + const scheduler = new FakeScheduler(); + const { adapter, notifications } = adapterWithChannel(async () => {}, { + deliveryScheduler: scheduler, + deliveryRetryBaseMs: 10, + }); + + await adapter.pushNotification(message("retry-stable-1", "needs ack")); + await scheduler.fireNext(); + await scheduler.fireNext(); + + expect(notifications).toHaveLength(3); + const messageIds = new Set(notifications.map((n) => n.params.meta.message_id)); + expect(messageIds.size).toBe(1); + expect(messageIds.values().next().value).toBe(deliveryIdFor(adapter, "retry-stable-1")); + const attemptIds = new Set(notifications.map((n) => n.params.meta.delivery_attempt_id)); + expect(attemptIds.size).toBe(3); + }); + + test("I: acknowledging one message releases exactly its bytes from the buffer accounting", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("bytes-1", "12345")); + await adapter.pushNotification(message("bytes-2", "1234567")); + expect(adapter.pendingMessageBytes).toBe(12); + + adapter.handleAckMessages({ ack_ids: [deliveryIdFor(adapter, "bytes-1")] }); + expect(adapter.pendingMessageBytes).toBe(7); + adapter.handleAckMessages({ ack_ids: [deliveryIdFor(adapter, "bytes-2")] }); + expect(adapter.pendingMessageBytes).toBe(0); + }); + + test("I: get_messages rejects malformed ack_ids without draining", async () => { + const { adapter } = adapterWithChannel(); + await adapter.pushNotification(message("guard-1", "still here")); + + const notArray = await callTool(adapter, "get_messages", { ack_ids: "not-an-array" }); + expect(notArray.isError).toBe(true); + const emptyItem = await callTool(adapter, "get_messages", { ack_ids: [""] }); + expect(emptyItem.isError).toBe(true); + const overCap = await callTool(adapter, "get_messages", { + ack_ids: Array.from({ length: 101 }, (_, i) => `id-${i}`), + }); + expect(overCap.isError).toBe(true); + expect(overCap.content[0].text).toContain("maximum is 100"); + expect(adapter.getPendingMessageCount()).toBe(1); + }); + + test("I: a mailbox configured above 100 keeps its drain epilogue acknowledgeable in one call", async () => { + const { adapter } = adapterWithChannel(async () => {}, { maxBufferedMessages: 150 }); + for (let i = 0; i < 120; i++) { + await adapter.pushNotification(message(`bulk-${i}`, `payload ${i}`)); + } + expect(adapter.getPendingMessageCount()).toBe(120); + + const listed = await adapter.server._requestHandlers.get("tools/list")({ method: "tools/list", params: {} }, {}); + const ackTool = listed.tools.find((tool: any) => tool.name === "ack_messages"); + expect(ackTool.inputSchema.properties.ack_ids.maxItems).toBe(150); + + const allIds = adapter.pendingMessages.map((entry: any) => entry.id); + const ack = await callTool(adapter, "ack_messages", { ack_ids: allIds }); + expect(ack.isError).toBeUndefined(); + expect(ack.content[0].text).toContain("Acknowledged 120 messages"); + expect(adapter.getPendingMessageCount()).toBe(0); + }); + + test("I: the byte bound measures UTF-8 bytes, not string length", async () => { + const { adapter } = adapterWithChannel(async () => {}, { maxBufferedBytes: 8 }); + // Three euro signs: length 3, but 9 UTF-8 bytes — must be rejected as oversized. + await adapter.pushNotification(message("utf8-over", "€€€")); + expect(adapter.getPendingMessageCount()).toBe(0); + expect(adapter.oversizedMessageCount).toBe(1); + expect(adapter.oversizedMessageBytes).toBe(9); + + // Eight ASCII bytes of the same magnitude are admitted. + await adapter.pushNotification(message("utf8-fit", "12345678")); + expect(adapter.getPendingMessageCount()).toBe(1); + }); + + test("I: an unadmitted oversized push is honest — no ACK contract, no mailbox claim", async () => { + const { adapter, notifications } = adapterWithChannel(async () => {}, { maxBufferedBytes: 4 }); + await adapter.pushNotification(message("oversized-honest", "12345")); + + expect(notifications).toHaveLength(1); + const pushed = notifications[0]; + expect(pushed.params.meta.ack_required).toBe(false); + expect(pushed.params.meta.ack_tool).toBeUndefined(); + expect(pushed.params.content).toContain("[AgentBridge oversized delivery id:"); + expect(pushed.params.content).not.toContain("call ack_messages with ack_ids"); + expect(pushed.params.content).toContain("12345"); + expect(adapter.getPendingMessageCount()).toBe(0); + // No retry may be armed for a message the mailbox never admitted. + expect(adapter.deliveryRetries.size).toBe(0); + }); +});