CB-205: coalesce duplicate bridge_ask calls onto one turn

A worker's tool execution is single-threaded, so two bridge_ask calls from the
same session can only be a transport retry — yet openAsk minted a fresh turnId
each time and both raced to resolve the primary's single forward waiter, the
loser returning NO_WAITER and leaving a dangling turn (observed as turnId #2 in
the live e2e). Now openAsk is idempotent per session: a second open ask coalesces
onto the existing turnId + answer future (fresh=false), and only the fresh owner
surfaces the question and tears the turn down. closeAsk clears the per-session
index (conditional by value) so a later ask reopens fresh.

Implemented by an off-subscription worker delegated over the bridge; reviewed and
validated on the primary (IDE-clean, mvn clean install 149 green, +2 tests:
concurrent double-ask coalescing + post-close reopen).

bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java

@@ -212,11 +212,15 @@ public final class MessageService {
      */
     public AskResult ask(String workerSession, String question, long timeoutMillis) {
         Rendezvous.AskTicket ticket = rendezvous.openAsk(workerSession);
-        // Register the reverse waiter first (above), then surface the question — so the answer, which
-        // can arrive the instant the primary reacts, always finds an open waiter to resolve.
-        if (!rendezvous.resolveQuestion(workerSession, question, ticket.turnId())) {
-            rendezvous.closeAsk(ticket.turnId());
-            return new AskResult(AskOutcome.NO_WAITER, null); // no primary is blocked on this worker
+        // Only the freshly-opening caller surfaces the question; a coalesced duplicate simply blocks on
+        // the shared answer future that the fresh owner is already responsible for.
+        if (ticket.fresh()) {
+            // Register the reverse waiter first, then surface the question — so the answer, which can
+            // arrive the instant the primary reacts, always finds an open waiter to resolve.
+            if (!rendezvous.resolveQuestion(workerSession, question, ticket.turnId())) {
+                rendezvous.closeAsk(ticket.turnId());
+                return new AskResult(AskOutcome.NO_WAITER, null); // no primary is blocked on this worker
+            }
         }
         try {
             String answer = ticket.answer().get(timeoutMillis, TimeUnit.MILLISECONDS);
@@ -231,7 +235,10 @@ public final class MessageService {
             Thread.currentThread().interrupt();
             throw new IllegalStateException("interrupted awaiting the primary's answer for " + workerSession, e);
         } finally {
-            rendezvous.closeAsk(ticket.turnId());
+            // Only the fresh owner tears down the shared turn; a duplicate must leave it open.
+            if (ticket.fresh()) {
+                rendezvous.closeAsk(ticket.turnId());
+            }
         }
     }
 

bridged/src/main/java/dev/ltms/bridged/msg/Rendezvous.java

@@ -56,8 +56,11 @@ public final class Rendezvous {
     private record AskWaiter(String session, CompletableFuture<String> answer) {
     }
 
-    /** Handle to a just-opened reverse-rendezvous turn: the minted {@code turnId} and its answer future. */
-    public record AskTicket(String turnId, CompletableFuture<String> answer) {
+    /**
+     * Handle to a reverse-rendezvous turn: the {@code turnId}, its answer future, and whether this
+     * call freshly opened it (versus coalescing onto an already-open ask).
+     */
+    public record AskTicket(String turnId, CompletableFuture<String> answer, boolean fresh) {
     }
 
     private final ConcurrentHashMap<String, CompletableFuture<Resolution>> waiters = new ConcurrentHashMap<>();
@@ -65,6 +68,8 @@ public final class Rendezvous {
     /** Reverse rendezvous (CB-205): worker questions awaiting the primary's answer, keyed by {@code turnId}. */
     private final ConcurrentHashMap<String, AskWaiter> asks = new ConcurrentHashMap<>();
     private final AtomicLong askSeq = new AtomicLong();
+    /** Per-session index of the currently-open ask, so duplicate bridge_ask calls coalesce onto one turn. */
+    private final ConcurrentHashMap<String, String> openAsksBySession = new ConcurrentHashMap<>();
 
     /**
      * Register a waiter for {@code session} — the await side of the public {@code resolve*} methods.
@@ -108,16 +113,34 @@ public final class Rendezvous {
     // --- reverse rendezvous (CB-205 bridge_ask) ------------------------------------------------
 
     /**
-     * Open a reverse-rendezvous waiter for a worker's mid-turn question: mint a fresh {@code turnId}
-     * (scoped to the worker session), register the answer future under it, and hand both back. The
-     * caller then {@link #resolveQuestion surfaces the question} to the primary and blocks on the
-     * returned future until the primary {@link #answerAsk answers} — a no-op if it never does.
+     * Open a reverse-rendezvous waiter for a worker's mid-turn question. If {@code session} already has
+     * an open ask, coalesce onto it (same {@code turnId}, same answer future). Otherwise atomically mint
+     * a fresh {@code turnId}, register it in both the per-turn and per-session indexes, and hand it back
+     * marked fresh. The caller then {@link #resolveQuestion surfaces the question} to the primary and
+     * blocks on the returned future until the primary {@link #answerAsk answers}.
      */
     public AskTicket openAsk(String session) {
-        CompletableFuture<String> answer = new CompletableFuture<>();
-        String turnId = session + "#" + askSeq.incrementAndGet();
-        asks.put(turnId, new AskWaiter(session, answer));
-        return new AskTicket(turnId, answer);
+        while (true) {
+            AskWaiter[] minted = { null };
+            String turnId = openAsksBySession.computeIfAbsent(session, _ -> {
+                String newTurnId = session + "#" + askSeq.incrementAndGet();
+                CompletableFuture<String> answer = new CompletableFuture<>();
+                AskWaiter waiter = new AskWaiter(session, answer);
+                asks.put(newTurnId, waiter);
+                minted[0] = waiter;
+                return newTurnId;
+            });
+            if (minted[0] != null) {
+                return new AskTicket(turnId, minted[0].answer(), true);
+            }
+            AskWaiter existing = asks.get(turnId);
+            if (existing != null) {
+                return new AskTicket(turnId, existing.answer(), false);
+            }
+            // A close raced and removed the waiter after we read the turnId; clear the stale index entry
+            // and retry so a fresh ask is always backed by a registered waiter.
+            openAsksBySession.remove(session, turnId);
+        }
     }
 
     /**
@@ -152,6 +175,13 @@ public final class Rendezvous {
 
     /** Drop a reverse-rendezvous turn once its {@code bridge_ask} has resolved (answered or lapsed). */
     public void closeAsk(String turnId) {
+        AskWaiter w = asks.get(turnId);
+        if (w == null) {
+            return;
+        }
+        // Remove the session index first and only if it still points to this turn, so a concurrent
+        // fresh ask cannot inherit a waiter we are about to drop.
+        openAsksBySession.remove(w.session(), turnId);
         asks.remove(turnId);
     }
 

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

@@ -145,6 +145,44 @@ class MessageServiceTest {
         assertEquals("done", done.text());
     }
 
+    @Test
+    void duplicateAsksFromTheSameSessionCoalesceToOneTurn() 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
+
+        // A transport retry: two concurrent bridge_ask calls from the same worker session.
+        CompletableFuture<MessageService.AskResult> ask1 =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+        CompletableFuture<MessageService.AskResult> ask2 =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
+
+        // The primary's single blocked send surfaces exactly ONE question (one turnId).
+        MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.Outcome.QUESTION, q.outcome());
+        assertEquals("which config file?", q.text());
+        assertNotNull(q.turnId(), "only one turnId should be minted");
+
+        // The primary answers that one turnId; both asks unblock with the same answer.
+        CompletableFuture<MessageService.Reply> answer =
+                CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000));
+
+        MessageService.AskResult a1 = ask1.get(5, TimeUnit.SECONDS);
+        MessageService.AskResult a2 = ask2.get(5, TimeUnit.SECONDS);
+        assertEquals(MessageService.AskOutcome.ANSWERED, a1.outcome());
+        assertEquals("config.yaml", a1.answer());
+        assertEquals(MessageService.AskOutcome.ANSWERED, a2.outcome());
+        assertEquals("config.yaml", a2.answer());
+
+        // The resumed worker finishes with a structured reply, resolving the answering send.
+        awaitWaiting();
+        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);

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

@@ -21,12 +21,23 @@ class RendezvousTest {
     private final Rendezvous rendezvous = new Rendezvous();
 
     @Test
-    void openAskMintsAUniqueTurnScopedToItsSession() {
+    void openAskMintsAUniqueTurnScopedToItsSessionAndCoalescesDuplicates() {
         Rendezvous.AskTicket t1 = rendezvous.openAsk(W);
         Rendezvous.AskTicket t2 = rendezvous.openAsk(W);
-        assertNotEquals(t1.turnId(), t2.turnId(), "each ask gets its own turnId");
+        assertEquals(t1.turnId(), t2.turnId(), "duplicate asks from the same session coalesce onto one turn");
+        assertTrue(t1.fresh(), "the first ask freshly opens the turn");
+        assertFalse(t2.fresh(), "the coalesced ask rides the existing turn");
         assertTrue(t1.turnId().startsWith(W + "#"), "the turnId is scoped to the worker session");
         assertEquals(W, rendezvous.askSession(t1.turnId()));
+    }
+
+    @Test
+    void openAskAfterCloseMintsANewTurn() {
+        Rendezvous.AskTicket t1 = rendezvous.openAsk(W);
+        rendezvous.closeAsk(t1.turnId());
+        Rendezvous.AskTicket t2 = rendezvous.openAsk(W);
+        assertNotEquals(t1.turnId(), t2.turnId(), "after closing, a new ask gets a fresh turnId");
+        assertTrue(t2.fresh(), "the reopened ask is fresh");
         assertEquals(W, rendezvous.askSession(t2.turnId()));
     }