CB-205/CB-201: bridge_ask reverse rendezvous + lightweight question/turn kind

A worker can now pause its delegated turn to ask the primary a question and
resume the same turn with the answer — the reverse of bridge_send.

- Rendezvous: QUESTION kind carrying a turnId, plus a reverse-ask registry
  (openAsk/resolveQuestion/askSession/answerAsk/closeAsk).
- MessageService.ask(): surface a worker's question to the primary's open send,
  block for the answer. answer(): resolve the worker's ask by turnId, then block
  for its eventual bridge_reply (session derived from turnId, not an argument).
- BridgeMcp: bridge_ask tool (worker-only, identity from the connection);
  bridge_send routes a turnId to the answer path. Shared formatReply().
- BridgedApp REST parity: POST /sessions/{id}/ask, turnId on /message.
- Lightweight CB-201: the kind vocabulary is the QUESTION outcome + turn_id
  correlation, not a rigid from/to/corr envelope (the connection-identity
  mechanism already covers addressing more robustly).

Tests: MessageServiceTest ask/answer round-trip + NO_WAITER/timeout/stale;
new RendezvousTest for the reverse registry; BridgeMcp ask/answer parity.
mvn: 147 green.

bridged/src/test/java/dev/ltms/bridged/mcp/BridgeMcpTest.java

@@ -120,6 +120,63 @@ class BridgeMcpTest {
         assertTrue(textOf(res).contains("no send is awaiting"));
     }
 
+    @Test
+    void askThenAnswerRoundTrips() throws Exception {
+        // The primary delegates and blocks; wait until its waiter is open before the worker asks.
+        CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.send(messages, "term_a", "do X", 5000L));
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            //noinspection BusyWait
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"), "the send must be waiting for the ask to surface to");
+
+        // The worker asks mid-turn; the call blocks for the primary's answer.
+        CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.ask(messages, "term_a", "which config?", 5000L));
+
+        // The primary's send unblocks with the question and a turnId to answer on.
+        McpSchema.CallToolResult q = send.get(6, TimeUnit.SECONDS);
+        assertNotEquals(Boolean.TRUE, q.isError());
+        String qt = textOf(q);
+        assertTrue(qt.contains("[question]"), qt);
+        String afterMarker = qt.substring(qt.indexOf("turnId=\"") + "turnId=\"".length());
+        String turnId = afterMarker.substring(0, afterMarker.indexOf('"'));
+
+        // The primary answers via bridge_send(turnId); this blocks again for the worker's reply.
+        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.answer(messages, turnId, "config.yaml", 5000L));
+
+        // The worker's ask returns the answer — it resumes the same turn.
+        assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
+
+        // The resumed worker replies, resolving the answering send (retry past the reopen race).
+        McpSchema.CallToolResult reply = BridgeMcp.reply(rendezvous, "term_a", "done");
+        deadline = System.currentTimeMillis() + 3000;
+        while (Boolean.TRUE.equals(reply.isError()) && System.currentTimeMillis() < deadline) {
+            //noinspection BusyWait
+            Thread.sleep(10);
+            reply = BridgeMcp.reply(rendezvous, "term_a", "done");
+        }
+        assertEquals("delivered", textOf(reply));
+        assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
+    }
+
+    @Test
+    void askFromANonWorkerConnectionIsAnError() {
+        McpSchema.CallToolResult res = BridgeMcp.ask(messages, null, "which config?", 500L);
+        assertTrue(res.isError());
+        assertTrue(textOf(res).contains("workers only"), textOf(res));
+    }
+
+    @Test
+    void answerToAStaleTurnIsAnError() {
+        McpSchema.CallToolResult res = BridgeMcp.answer(messages, "term_a#999", "too late", 500L);
+        assertTrue(res.isError());
+        assertTrue(textOf(res).contains("no longer open"), textOf(res));
+    }
+
     @Test
     void spawnReturnsTheNewWorkersSessionAndPane() {
         FakeHerdr h = new FakeHerdr();

bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java

@@ -13,6 +13,7 @@ import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
@@ -107,4 +108,69 @@ class MessageServiceTest {
                 "a delivered send whose worker vanishes fails instead of hanging to the timeout");
         assertFalse(reply.completed());
     }
+
+    // --- bridge_ask reverse rendezvous (CB-205) ------------------------------------------------
+
+    @Test
+    void askSurfacesAsAQuestionAndTheAnswerResumesTheSameTurn() throws Exception {
+        CompletableFuture<MessageService.Reply> send = sendAsync();
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);    // deliver
+        injector.onStatus(T, AgentStatus.WORKING); // worker picks it up, then pauses to ask
+
+        // The worker asks mid-turn on its own thread; the call blocks for the primary's answer.
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+
+        // The primary's blocking send unblocks with the question and a turnId to answer on.
+        MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.QUESTION, q.outcome());
+        assertEquals("which config file?", q.text());
+        assertNotNull(q.turnId(), "a question carries a turnId to answer on");
+
+        // The primary answers via bridge_send(turnId); this blocks again for the worker's reply.
+        CompletableFuture<MessageService.Reply> answer =
+                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000));
+
+        // The worker's ask returns the answer — it resumes the same turn.
+        MessageService.AskResult a = ask.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.AskOutcome.ANSWERED, a.outcome());
+        assertEquals("config.yaml", a.answer());
+
+        // The resumed worker finishes with a structured reply, resolving the answering send.
+        awaitWaiting(); // the answering send has (re)opened its forward waiter
+        assertTrue(rendezvous.resolve(T, "done"), "the worker's final reply resolves the answering send");
+        MessageService.Reply done = answer.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.REPLIED, done.outcome());
+        assertEquals("done", done.text());
+    }
+
+    @Test
+    void askWithNoOpenDelegationReturnsNoWaiter() {
+        MessageService.AskResult r = messages.ask(T, "anyone listening?", 500);
+        assertEquals(MessageService.AskOutcome.NO_WAITER, r.outcome(),
+                "a question with no blocked send has no primary to answer it");
+    }
+
+    @Test
+    void askTimesOutWhenThePrimaryNeverAnswers() throws Exception {
+        CompletableFuture<MessageService.Reply> send = sendAsync();
+        awaitWaiting();
+        injector.onStatus(T, AgentStatus.IDLE);
+        injector.onStatus(T, AgentStatus.WORKING);
+
+        MessageService.AskResult r = messages.ask(T, "still there?", 200); // primary never answers
+        assertEquals(MessageService.AskOutcome.TIMED_OUT, r.outcome());
+
+        // The send itself already unblocked with the question the instant the ask surfaced.
+        MessageService.Reply q = send.get(2, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.QUESTION, q.outcome());
+    }
+
+    @Test
+    void answeringAnUnknownTurnIsStale() {
+        MessageService.Reply r = messages.answer(T + "#999", "too late", 500);
+        assertEquals(MessageService.Outcome.STALE_TURN, r.outcome(),
+                "an answer to a turn that never existed (or already lapsed) is stale, not a hang");
+    }
 }

bridged/src/test/java/dev/ltms/bridged/msg/RendezvousTest.java

@@ -0,0 +1,69 @@
+package dev.ltms.bridged.msg;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The reverse rendezvous (CB-205): the {@code bridge_ask} registry that lets a worker pause mid-turn
+ * to ask the primary. Unit-level — the message-layer round-trip is covered in {@link MessageServiceTest}.
+ */
+class RendezvousTest {
+
+    private static final String W = "term_a";
+
+    private final Rendezvous rendezvous = new Rendezvous();
+
+    @Test
+    void openAskMintsAUniqueTurnScopedToItsSession() {
+        Rendezvous.AskTicket t1 = rendezvous.openAsk(W);
+        Rendezvous.AskTicket t2 = rendezvous.openAsk(W);
+        assertNotEquals(t1.turnId(), t2.turnId(), "each ask gets its own turnId");
+        assertTrue(t1.turnId().startsWith(W + "#"), "the turnId is scoped to the worker session");
+        assertEquals(W, rendezvous.askSession(t1.turnId()));
+        assertEquals(W, rendezvous.askSession(t2.turnId()));
+    }
+
+    @Test
+    void answerAskCompletesTheWaitersFuture() {
+        Rendezvous.AskTicket t = rendezvous.openAsk(W);
+        assertTrue(rendezvous.answerAsk(t.turnId(), "config.yaml"), "answering an open ask succeeds");
+        assertEquals("config.yaml", t.answer().getNow(null), "the answer reaches the blocked worker");
+    }
+
+    @Test
+    void answerAskOnAnUnknownTurnIsFalse() {
+        assertFalse(rendezvous.answerAsk("no-such#1", "x"), "an answer to an unknown turn is a no-op");
+    }
+
+    @Test
+    void resolveQuestionResolvesAnOpenSendWithTheQuestionKindAndTurnId() {
+        CompletableFuture<Rendezvous.Resolution> send = rendezvous.open(W);
+        assertTrue(rendezvous.resolveQuestion(W, "which config?", W + "#7"),
+                "the question resolves the primary's open send");
+        Rendezvous.Resolution r = send.getNow(null);
+        assertEquals(Rendezvous.Kind.QUESTION, r.kind());
+        assertEquals("which config?", r.text());
+        assertEquals(W + "#7", r.turnId(), "the turnId rides along so the primary can answer");
+    }
+
+    @Test
+    void resolveQuestionWithNoOpenSendIsFalse() {
+        assertFalse(rendezvous.resolveQuestion(W, "anyone?", W + "#1"),
+                "no blocked send means no primary to surface the question to");
+    }
+
+    @Test
+    void closeAskRemovesTheTurn() {
+        Rendezvous.AskTicket t = rendezvous.openAsk(W);
+        rendezvous.closeAsk(t.turnId());
+        assertNull(rendezvous.askSession(t.turnId()), "a closed ask is forgotten");
+        assertFalse(rendezvous.answerAsk(t.turnId(), "late"), "a closed ask can no longer be answered");
+    }
+}