diff --git a/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java b/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java index 4b98a40..2c63c5c 100644 --- a/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java +++ b/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()); + } } } diff --git a/bridged/src/main/java/dev/ltms/bridged/msg/Rendezvous.java b/bridged/src/main/java/dev/ltms/bridged/msg/Rendezvous.java index 7163f49..f3895ec 100644 --- a/bridged/src/main/java/dev/ltms/bridged/msg/Rendezvous.java +++ b/bridged/src/main/java/dev/ltms/bridged/msg/Rendezvous.java @@ -56,8 +56,11 @@ public final class Rendezvous { private record AskWaiter(String session, CompletableFuture answer) { } - /** Handle to a just-opened reverse-rendezvous turn: the minted {@code turnId} and its answer future. */ - public record AskTicket(String turnId, CompletableFuture 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 answer, boolean fresh) { } private final ConcurrentHashMap> 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 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 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 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 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); } diff --git a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java b/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java index bd74066..550481c 100644 --- a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java +++ b/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 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 ask1 = + CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000)); + CompletableFuture 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 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); diff --git a/bridged/src/test/java/dev/ltms/bridged/msg/RendezvousTest.java b/bridged/src/test/java/dev/ltms/bridged/msg/RendezvousTest.java index 6848a75..61e1a58 100644 --- a/bridged/src/test/java/dev/ltms/bridged/msg/RendezvousTest.java +++ b/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())); }