11 changed files with 270 additions and 60 deletions
@@ -209,11 +209,16 @@ public final class Bridged {
        MessageService messages = new MessageService(agents, injector, rendezvous, replyInbox,
                pushLoop, metrics);

+        // CB-520: the reply inbox only consumes for agents this gateway owns. own on acquire,
+        // release on teardown. Do this before CB-516 so the inbox is owned before any reply can land.
+        sessions.onAcquire(replyInbox::own);
        // CB-516: releasing a worker must fail whatever send was waiting on it. Without this a
        // torn-down delegation kept reporting PENDING until the 30-minute async timeout, and never
        // reached /metrics — the delegation was unresolvable and nothing said so.
-        sessions.onRelease(terminal ->
-                messages.abandon(terminal, "the worker session was released before it replied"));
+        sessions.onRelease(terminal -> {
+            messages.abandon(terminal, "the worker session was released before it replied");
+            replyInbox.release(terminal);
+        });

        // MCP server face (CB-105): bridge_send/bridge_reply/bridge_status, mounted at /mcp.
        // Caller identity is resolved from the connection (peer PID → herdr pane), not arguments.
@@ -14,25 +14,30 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;

 /**
 * AMQP-backed {@link ReplyInbox} (CB-307 Stage 2): genuine cross-restart durability behind the same
 * port {@link InMemoryReplyInbox} implements as soft state.
 *
- * <p><strong>Mapping — consume-and-hold with deferred manual ack.</strong> Each target owns a durable
- * queue {@code agent.<target>.inbox}. A manual-ack consumer pulls persistent messages off that queue
- * into an in-memory <em>held</em> map (keyed by {@code msgId}) but does <em>not</em> ack them.
- * {@link #peek} returns that snapshot; {@link #ack} acks the broker delivery-tag and drops the entry.
- * Because messages stay unacked until the primary actually drains them, a crash (or a {@code java -jar}
- * bounce) before caller-ack leaves them on the broker — it redelivers on reconnect. That is the
- * durability the in-memory adapter cannot give, with the port contract preserved.
+ * <p><strong>Mapping — consume-and-hold with deferred manual ack.</strong> Each target has a durable
+ * queue {@code agent.<target>.inbox}. The gateway that owns the target starts a manual-ack consumer
+ * ({@link #own}) that pulls persistent messages off that queue into an in-memory <em>held</em> map
+ * (keyed by {@code msgId}) but does <em>not</em> ack them. {@link #peek} returns that snapshot;
+ * {@link #ack} acks the broker delivery-tag and drops the entry. Because messages stay unacked until
+ * the owning gateway actually drains them, a crash (or a {@code java -jar} bounce) before caller-ack
+ * leaves them on the broker — it redelivers on reconnect. That is the durability the in-memory
+ * adapter cannot give, with the port contract preserved.
+ *
+ * <p><strong>Ownership is explicit.</strong> {@link #own} declares the queue and starts the consumer;
+ * {@link #release} cancels it. {@link #publish} sends to the queue but does <em>not</em> imply ownership
+ * and does not attach a consumer. This split is required by CB-308 federation, where one gateway may
+ * publish to an agent owned by another gateway; in that case the publisher must not compete for
+ * deliveries.
 *
 * <p><strong>Dedup.</strong> The consumer keys the held map by {@code msgId}; a redelivered duplicate
 * (at-least-once, or a producer double-publish) is acked-and-dropped on arrival, so it never
- * double-queues. {@link #publish} additionally short-circuits an already-held {@code msgId} — a
- * fast path; the consumer-side check is the real guarantee.
+ * double-queues.
 *
 * <p><strong>Visibility.</strong> Unlike the in-memory adapter, publish → broker → consumer is
 * asynchronous, so a {@link #peek} immediately after {@link #publish} may not yet see the message
@@ -51,12 +56,12 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {

    private final Connection connection;
    private final Channel channel;
-    /** All channel operations (publish/declare/ack) serialize on this — a Channel is not thread-safe. */
+    /** All channel operations (publish/declare/ack/cancel) serialize on this — a Channel is not thread-safe. */
    private final Object channelLock = new Object();
    /** target → (msgId → held delivery). Per-target map is guarded by synchronizing on itself. */
    private final ConcurrentHashMap<String, LinkedHashMap<String, Held>> held = new ConcurrentHashMap<>();
-    /** Targets whose queue is declared and consumer is running. */
-    private final Set<String> consuming = ConcurrentHashMap.newKeySet();
+    /** Targets whose queue is declared and consumer is running, mapped to their broker consumer tag. */
+    private final ConcurrentHashMap<String, String> consumerTags = new ConcurrentHashMap<>();

    /** A message pulled off the broker but not yet acked: its delivery-tag plus the port payload. */
    private record Held(long deliveryTag, InboxMessage message) {}
@@ -103,16 +108,41 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
    }

    @Override
-    public void publish(String target, String msgId, String content) {
-        ensureConsuming(target);
-        var perTarget = held.get(target);
-        if (perTarget != null) {
-            synchronized (perTarget) {
-                if (perTarget.containsKey(msgId)) {
-                    return;                        // already held — producer-side fast dedup
-                }
+    public void own(String target) {
+        synchronized (channelLock) {
+            if (consumerTags.containsKey(target)) {
+                return;                            // already owning this target
+            }
+            String queue = queueName(target);
+            try {
+                channel.queueDeclare(queue, true, false, false, null); // durable, non-exclusive, keep on idle
+                String tag = channel.basicConsume(queue, false, deliverCallback(target), _ -> { });
+                consumerTags.put(target, tag);
+                log.debug("AMQP inbox owns queue {} for target {}", queue, target);
+            } catch (IOException e) {
+                throw new IllegalStateException("cannot own queue " + queue, e);
            }
        }
+    }
+
+    @Override
+    public void release(String target) {
+        synchronized (channelLock) {
+            String tag = consumerTags.remove(target);
+            held.remove(target);                   // stale delivery tags must not survive release
+            if (tag == null) {
+                return;
+            }
+            try {
+                channel.basicCancel(tag);
+            } catch (IOException e) {
+                throw new IllegalStateException("cannot cancel consumer for " + target, e);
+            }
+        }
+    }
+
+    @Override
+    public void publish(String target, String msgId, String content) {
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .messageId(msgId)
                .deliveryMode(2)                   // persistent — survives a broker restart
@@ -129,7 +159,6 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {

    @Override
    public List<InboxMessage> peek(String target) {
-        ensureConsuming(target);
        var perTarget = held.get(target);
        if (perTarget == null) {
            return List.of();
@@ -166,26 +195,6 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
        }
    }

-    /** Declare the durable per-target queue and start its manual-ack consumer, once per target. */
-    private void ensureConsuming(String target) {
-        if (consuming.contains(target)) {
-            return;
-        }
-        synchronized (channelLock) {
-            if (!consuming.add(target)) {
-                return;                            // another thread just set it up
-            }
-            String queue = queueName(target);
-            try {
-                channel.queueDeclare(queue, true, false, false, null); // durable, non-exclusive, keep on idle
-                channel.basicConsume(queue, false, deliverCallback(target), _ -> { });
-            } catch (IOException e) {
-                consuming.remove(target);
-                throw new IllegalStateException("cannot consume queue " + queue, e);
-            }
-        }
-    }
-
    private DeliverCallback deliverCallback(String target) {
        return (_, delivery) -> {
            String msgId = delivery.getProperties().getMessageId();
@@ -2,6 +2,7 @@ package dev.ltms.bridged.msg;

 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;

 /**
@@ -9,12 +10,29 @@ import java.util.concurrent.ConcurrentHashMap;
 * Per-target FIFO ordering (insertion order via {@link LinkedHashMap}). Dedup by {@code msgId}
 * within a target. Thread-safe for concurrent publish vs. drain.
 *
+ * <p><strong>Ownership is explicit.</strong> {@link #own} marks a target as locally owned so that
+ * {@link #peek} and {@link #ack} operate on it; {@link #publish} works whether or not the target is
+ * owned. {@link #release} clears the local snapshot. This mirrors the AMQP adapter's contract so the
+ * non-broker path stays interchangeable.
+ *
 * <p><strong>This is soft-state, NOT persistence.</strong> Lost on a {@code java -jar} bounce — that
 * is correct and consistent with "bridged stays soft-state." The Stage-2 AMQP adapter replaces this.
 */
 public final class InMemoryReplyInbox implements ReplyInbox {

    private final ConcurrentHashMap<String, LinkedHashMap<String, InboxMessage>> store = new ConcurrentHashMap<>();
+    private final Set<String> owned = ConcurrentHashMap.newKeySet();
+
+    @Override
+    public void own(String target) {
+        owned.add(target);
+    }
+
+    @Override
+    public void release(String target) {
+        owned.remove(target);
+        store.remove(target);
+    }

    @Override
    public void publish(String target, String msgId, String content) {
@@ -27,6 +45,9 @@ public final class InMemoryReplyInbox implements ReplyInbox {

    @Override
    public List<InboxMessage> peek(String target) {
+        if (!owned.contains(target)) {
+            return List.of();
+        }
        var perTarget = store.get(target);
        if (perTarget == null) {
            return List.of();
@@ -39,6 +60,9 @@ public final class InMemoryReplyInbox implements ReplyInbox {

    @Override
    public void ack(String target, String msgId) {
+        if (!owned.contains(target)) {
+            return;
+        }
        var perTarget = store.get(target);
        if (perTarget != null) {
            //noinspection SynchronizationOnLocalVariableOrMethodParameter
@@ -10,16 +10,36 @@ import java.util.List;
 * <p><strong>This interface is the port.</strong> {@link InMemoryReplyInbox} is the Stage-1 adapter;
 * an AMQP-backed adapter (Stage 2) must implement the same contract (idempotent publish, FIFO peek,
 * at-least-once ack).
+ *
+ * <p><strong>Ownership is explicit.</strong> A gateway {@link #own owns} the inbox for each agent it
+ * spawned; only the owner consumes and drains it. {@link #publish} sends a reply to the target's
+ * inbox but does <em>not</em> imply ownership or start a consumer. This separation is required by
+ * CB-308 federation, where one gateway may publish to an agent owned by another gateway.
 */
 public interface ReplyInbox {

    /** A queued reply: an idempotency id, the worker session it came from, and the reply text. */
    record InboxMessage(String msgId, String target, String content) {}

+    /**
+     * Start owning (consuming) the inbox for {@code target}. Idempotent: multiple calls for the same
+     * target are no-ops. The owner is the only gateway that may {@link #peek} and {@link #ack} replies
+     * for this target.
+     */
+    void own(String target);
+
+    /**
+     * Stop owning (consuming) the inbox for {@code target}. Idempotent. Any replies held locally but
+     * not yet acked are dropped from the local snapshot; the underlying durable queue keeps
+     * unacked messages for redelivery when the target is re-owned.
+     */
+    void release(String target);
+
    /**
     * Queue {@code content} from worker {@code target} under {@code msgId}. Idempotent: publishing an
     * already-present {@code msgId} for {@code target} is a no-op (dedup), so an at-least-once Stage-2
-     * redelivery cannot double-queue.
+     * redelivery cannot double-queue. Publishing does <em>not</em> imply ownership and must not start a
+     * consumer.
     */
    void publish(String target, String msgId, String content);

@@ -48,8 +48,10 @@ public final class SessionManager implements TurnListener {
    private final LongSupplier nowNanos;
    private final int contextCap;

+    /** CB-520: notified with a terminalId on every acquire; no-op until wired. */
+    private final List<Consumer<String>> acquireListeners = new java.util.concurrent.CopyOnWriteArrayList<>();
    /** CB-516: notified with a terminalId on every release; no-op until wired. */
-    private volatile Consumer<String> releaseListener = _ -> { };
+    private final List<Consumer<String>> releaseListeners = new java.util.concurrent.CopyOnWriteArrayList<>();

    /** Backward-compatible constructor: shared-tree sessions, production git seam. */
    public SessionManager(PeerLauncher launcher) {
@@ -129,6 +131,7 @@ public final class SessionManager implements TurnListener {
            registry.put(handle.id(), session);
            log.debug("acquired session id={} terminal={} profile={} owner={}",
                    handle.id(), handle.terminalId(), session.profile(), session.ownerTerminal());
+            notifyAcquired(session.terminalId());
            return session;
        }
        return acquireWithWorktree(profile, requestedCwd, callerCwd, ownerTerminal, wt);
@@ -151,18 +154,44 @@ public final class SessionManager implements TurnListener {
        }
    }

+    /**
+     * Register a callback invoked with a session's {@code terminalId} whenever it is acquired
+     * (CB-520). This is the hook that lets the reply inbox {@code own} a target's queue.
+     */
+    public void onAcquire(Consumer<String> listener) {
+        if (listener != null) {
+            acquireListeners.add(listener);
+        }
+    }
+
    /**
     * Register a callback invoked with a session's {@code terminalId} whenever it is released
     * (CB-516). Every teardown path funnels through {@link #release}, so one hook covers the REST
     * and MCP stop tools, the idle-TTL reaper, {@code recycle}, and shutdown drain alike.
     *
-     * <p>Set rather than injected because {@code MessageService} — the intended listener — is
+     * <p>Added rather than injected because {@code MessageService} — one intended listener — is
     * constructed after this manager (it needs the injector and rendezvous, which need the session
     * presence view this manager exposes). Wiring it at construction would require breaking that
     * cycle for one callback.
     */
    public void onRelease(Consumer<String> listener) {
-        this.releaseListener = (listener == null) ? _ -> { } : listener;
+        if (listener != null) {
+            releaseListeners.add(listener);
+        }
+    }
+
+    /** A listener failure must never prevent the acquisition it is reacting to. */
+    private void notifyAcquired(String terminalId) {
+        if (terminalId == null) {
+            return;
+        }
+        for (Consumer<String> listener : acquireListeners) {
+            try {
+                listener.accept(terminalId);
+            } catch (RuntimeException e) {
+                log.warn("acquire listener failed for terminal {}: {}", terminalId, e.toString());
+            }
+        }
    }

    /** A listener failure must never prevent the teardown it is reacting to. */
@@ -170,10 +199,12 @@ public final class SessionManager implements TurnListener {
        if (terminalId == null) {
            return;
        }
-        try {
-            releaseListener.accept(terminalId);
-        } catch (RuntimeException e) {
-            log.warn("release listener failed for terminal {}: {}", terminalId, e.toString());
+        for (Consumer<String> listener : releaseListeners) {
+            try {
+                listener.accept(terminalId);
+            } catch (RuntimeException e) {
+                log.warn("release listener failed for terminal {}: {}", terminalId, e.toString());
+            }
        }
    }

@@ -221,6 +252,7 @@ public final class SessionManager implements TurnListener {
        registry.put(handle.id(), session);
        log.debug("acquired worktree session id={} terminal={} profile={} branch={} path={}",
                handle.id(), handle.terminalId(), session.profile(), session.branch(), session.worktree());
+        notifyAcquired(session.terminalId());
        return session;
    }

@@ -15,6 +15,8 @@ import dev.ltms.bridged.session.WorkerSession;
 import dev.ltms.bridged.session.WorktreeRequest;
 import dev.ltms.bridged.worker.ClaudeCodeLauncher;
 import io.modelcontextprotocol.spec.McpSchema;
+import dev.ltms.bridged.msg.InMemoryReplyInbox;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;

 import java.util.Map;
@@ -31,10 +33,19 @@ import static org.junit.jupiter.api.Assertions.*;
 */
 class BridgeMcpTest {

+    private static final String T = "term_a";
+
    private final FakeHerdr herdr = new FakeHerdr();
    private final AgentControl agents = new AgentControl(herdr);
    private final Rendezvous rendezvous = new Rendezvous();
-    private final MessageService messages = new MessageService(agents, new Injector(agents), rendezvous);
+    private final InMemoryReplyInbox inbox = new InMemoryReplyInbox();
+    private final MessageService messages = new MessageService(agents, new Injector(agents), rendezvous, inbox);
+
+    @BeforeEach
+    void setUp() {
+        // CB-520: the inbox only peeks/acks targets it owns.
+        inbox.own(T);
+    }

    private static String textOf(McpSchema.CallToolResult r) {
        return ((McpSchema.TextContent) r.content().getFirst()).text();
@@ -1,5 +1,9 @@
 package dev.ltms.bridged.msg;

+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.RabbitMQContainer;
@@ -20,8 +24,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 * run it with Docker present via {@code mvn test -Pcontract}.
 *
 * <p>It proves the port contract on genuine infrastructure: eventual visibility of a published reply,
- * ack removal, msgId dedup, and — the reason Stage 2 exists — cross-restart durability: an unacked
- * reply survives closing the inbox and is redelivered to a fresh connection.
+ * ack removal, msgId dedup, explicit ownership, and — the reason Stage 2 exists — cross-restart
+ * durability: an unacked reply survives closing the inbox and is redelivered to a fresh connection.
 */
@Tag("contract")
@Testcontainers
@@ -41,6 +45,7 @@ class AmqpReplyInboxContractTest {
    void publishThenPeekThenAck() throws Exception {
        String target = "worker-pub-" + System.nanoTime();
        try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
+            inbox.own(target);
            inbox.publish(target, "m1", "hello primary");

            List<ReplyInbox.InboxMessage> got = awaitPeek(inbox, target);
@@ -58,6 +63,7 @@ class AmqpReplyInboxContractTest {
    void duplicateMsgIdIsNotDoubleQueued() throws Exception {
        String target = "worker-dedup-" + System.nanoTime();
        try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
+            inbox.own(target);
            inbox.publish(target, "dup", "first");
            awaitPeek(inbox, target);
            inbox.publish(target, "dup", "second");   // same msgId — must be a no-op
@@ -74,8 +80,9 @@ class AmqpReplyInboxContractTest {
    void unackedReplySurvivesRestartAndIsRedelivered() throws Exception {
        String target = "worker-durable-" + System.nanoTime();

-        // First "process life": publish, see it held, but crash before acking.
+        // First "process life": own, publish, see it held, but crash before acking.
        try (AmqpReplyInbox first = AmqpReplyInbox.open(uri())) {
+            first.own(target);
            first.publish(target, "persist-1", "survive me");
            assertEquals(1, awaitPeek(first, target).size());
            // no ack — simulate a java -jar bounce with the reply still pending
@@ -83,6 +90,7 @@ class AmqpReplyInboxContractTest {

        // Second "process life": a fresh connection to the same broker must be redelivered the reply.
        try (AmqpReplyInbox second = AmqpReplyInbox.open(uri())) {
+            second.own(target);
            List<ReplyInbox.InboxMessage> got = awaitPeek(second, target);
            assertEquals(1, got.size(), "an unacked persistent reply is redelivered after restart");
            assertEquals("persist-1", got.getFirst().msgId());
@@ -93,11 +101,43 @@ class AmqpReplyInboxContractTest {

        // Third life: once acked, it is gone for good — durability is not endless replay.
        try (AmqpReplyInbox third = AmqpReplyInbox.open(uri())) {
+            third.own(target);
            Thread.sleep(500);
            assertTrue(third.peek(target).isEmpty(), "an acked reply does not come back on the next restart");
        }
    }

+    @Test
+    void publishDoesNotAttachAConsumer() throws Exception {
+        String target = "worker-pub-no-consumer-" + System.nanoTime();
+        try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri());
+             Connection inspect = newConnection()) {
+            inbox.own(target);
+            inbox.publish(target, "m1", "published");
+            awaitPeek(inbox, target); // ensure the owner's consumer received it
+
+            try (Channel ch = inspect.createChannel()) {
+                AMQP.Queue.DeclareOk ok = ch.queueDeclare(queueName(target), true, false, false, null);
+                assertEquals(1, ok.getConsumerCount(),
+                        "publish must not attach a consumer; only the owner's consumer should exist");
+            }
+        }
+    }
+
+    @Test
+    void releaseCancelsConsumerAndClearsHeld() throws Exception {
+        String target = "worker-release-" + System.nanoTime();
+        try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
+            inbox.own(target);
+            inbox.publish(target, "m1", "release me");
+            assertEquals(1, awaitPeek(inbox, target).size(), "owned target holds the reply");
+
+            inbox.release(target);
+            assertTrue(inbox.peek(target).isEmpty(),
+                    "release clears the local held snapshot");
+        }
+    }
+
    /** Poll peek (broker delivery is async) until a reply for {@code target} appears or ~10s elapse. */
    @SuppressWarnings("BusyWait") // deliberate poll for async broker delivery, bounded by the deadline
    private static List<ReplyInbox.InboxMessage> awaitPeek(AmqpReplyInbox inbox, String target)
@@ -110,4 +150,15 @@ class AmqpReplyInboxContractTest {
        }
        return msgs;
    }
+
+    /** A separate broker connection for inspecting queue state without disturbing the inbox. */
+    private static Connection newConnection() throws Exception {
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setUri(uri());
+        return factory.newConnection("contract-inspector");
+    }
+
+    private static String queueName(String target) {
+        return "agent." + target + ".inbox";
+    }
 }
@@ -1,5 +1,6 @@
 package dev.ltms.bridged.msg;

+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;

 import java.util.concurrent.CountDownLatch;
@@ -10,13 +11,18 @@ import java.util.concurrent.atomic.AtomicReference;
 import static org.junit.jupiter.api.Assertions.*;

 /**
- * Unit tests for {@link InMemoryReplyInbox}: publish, peek, ack, dedup, FIFO ordering, and thread
- * safety under concurrent publish vs. drain.
+ * Unit tests for {@link InMemoryReplyInbox}: publish, peek, ack, dedup, FIFO ordering, thread
+ * safety under concurrent publish vs. drain, and explicit ownership.
 */
 class InMemoryReplyInboxTest {

    private final ReplyInbox inbox = new InMemoryReplyInbox();

+    @BeforeEach
+    void setUp() {
+        inbox.own("term_a");
+    }
+
    @Test
    void publishThenPeekReturnsTheMessage() {
        inbox.publish("term_a", "m1", "hello");
@@ -72,6 +78,7 @@ class InMemoryReplyInboxTest {

    @Test
    void perTargetIsolation() {
+        inbox.own("term_b");
        inbox.publish("term_a", "m1", "for-a");
        inbox.publish("term_b", "m2", "for-b");
        assertEquals(1, inbox.peek("term_a").size());
@@ -142,4 +149,40 @@ class InMemoryReplyInboxTest {
            exec.shutdown();
        }
    }
+
+    @Test
+    void publishWithoutOwnDoesNotClaimOwnership() {
+        String unowned = "term_unowned";
+        inbox.publish(unowned, "m1", "hello");
+        // Without an owner, peek returns nothing — publish did not imply consume.
+        assertTrue(inbox.peek(unowned).isEmpty(),
+                "publishing to an unowned target must not make it peekable");
+        // Owning afterwards makes the already-published message visible.
+        inbox.own(unowned);
+        var msgs = inbox.peek(unowned);
+        assertEquals(1, msgs.size());
+        assertEquals("m1", msgs.getFirst().msgId());
+    }
+
+    @Test
+    void peekAndAckAreNoOpsForUnownedTarget() {
+        assertTrue(inbox.peek("term_not_owned").isEmpty());
+        inbox.ack("term_not_owned", "m1"); // no-op, should not throw
+    }
+
+    @Test
+    void releaseStopsConsumingAndClearsHeld() {
+        inbox.publish("term_a", "m1", "hello");
+        assertEquals(1, inbox.peek("term_a").size());
+        inbox.release("term_a");
+        assertTrue(inbox.peek("term_a").isEmpty(),
+                "after release, the local snapshot is cleared");
+    }
+
+    @Test
+    void ownIsIdempotent() {
+        inbox.own("term_a");
+        inbox.publish("term_a", "m1", "hello");
+        assertEquals(1, inbox.peek("term_a").size());
+    }
 }
@@ -6,6 +6,7 @@ import dev.ltms.bridged.herdr.FakeHerdr;
 import dev.ltms.bridged.herdr.HerdrException;
 import dev.ltms.bridged.inject.CompletionResolver;
 import dev.ltms.bridged.inject.Injector;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;

 import java.util.concurrent.CompletableFuture;
@@ -30,7 +31,14 @@ class MessageServiceTest {
    private final Rendezvous rendezvous = new Rendezvous();
    private final CompletionResolver completion = new CompletionResolver(agents, rendezvous);
    private final Injector injector = new Injector(agents, completion);
-    private final MessageService messages = new MessageService(agents, injector, rendezvous);
+    private final InMemoryReplyInbox inbox = new InMemoryReplyInbox();
+    private final MessageService messages = new MessageService(agents, injector, rendezvous, inbox);
+
+    @BeforeEach
+    void setUp() {
+        // CB-520: the inbox only peeks/acks targets it owns.
+        inbox.own(T);
+    }

    /** Run {@code send} on a background thread; the current thread drives the worker's turn. */
    private CompletableFuture<MessageService.Reply> sendAsync() {
@@ -45,6 +45,7 @@ class ReplyPushLoopTest {
    void setUp() {
        registry = new PrimaryRegistry(PRIMARY);
        inbox = new InMemoryReplyInbox();
+        inbox.own(WORKER); // CB-520: the inbox only peeks/acks targets it owns
        scheduler = Executors.newSingleThreadScheduledExecutor();
    }

@@ -10,6 +10,7 @@ import dev.ltms.bridged.herdr.WorkspaceControl;
 import dev.ltms.bridged.inject.Injector;
 import dev.ltms.bridged.inject.StatusPoller;
 import dev.ltms.bridged.inject.WorkerPresence;
+import dev.ltms.bridged.msg.InMemoryReplyInbox;
 import dev.ltms.bridged.msg.MessageService;
 import dev.ltms.bridged.msg.Rendezvous;
 import dev.ltms.bridged.session.FakeWorktrees;
@@ -74,7 +75,12 @@ class BridgedAppTest {
        poller = new StatusPoller(agents, injector, 5); // delivers when the fake reports idle
        poller.start();
        Rendezvous rendezvous = new Rendezvous();
-        MessageService messages = new MessageService(agents, injector, rendezvous);
+        InMemoryReplyInbox inbox = new InMemoryReplyInbox();
+        sessions.onAcquire(inbox::own);
+        // The static REST tests address "term_a" without acquiring it through SessionManager, so own
+        // it directly so the inbox contract holds for those endpoints.
+        inbox.own("term_a");
+        MessageService messages = new MessageService(agents, injector, rendezvous, inbox);
        app = new BridgedApp(herdr, workers, sessions, messages, this.presence, null)
                .build().start("127.0.0.1", 0);
        return app.port();