CB-577: model async question ownership race

bridged/src/main/java/dev/ltms/bridged/Bridged.java

@@ -276,9 +276,9 @@ public final class Bridged {
             }
 
             @Override
-            public void onDelivered(String target, dev.ltms.bridged.msg.TurnToken token) {
-                completion.onDelivered(target, token);
-                sessions.onDelivered(target, token);
+            public void onDelivered(String target) {
+                completion.onDelivered(target);
+                sessions.onDelivered(target);
             }
 
             @Override

bridged/src/main/java/dev/ltms/bridged/inject/CompletionResolver.java

@@ -2,7 +2,6 @@ package dev.ltms.bridged.inject;
 
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.msg.Rendezvous;
-import dev.ltms.bridged.msg.TurnToken;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -84,17 +83,17 @@ public final class CompletionResolver implements TurnListener {
     }
 
     @Override
-    public void onDelivered(String target, TurnToken token) {
+    public void onDelivered(String target) {
         // Capture the exact waiter this turn belongs to (CB-116) and snapshot the pane's pre-turn
         // content — what it shows *before* the just-delivered turn produces output — as the staleness
         // reference (CB-115). Done synchronously (like the delivering send itself) so both are in
         // place before this turn's completion can fire.
-        captureBaseline(target, token);
+        captureBaseline(target);
     }
 
     /** Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link #onDelivered}). */
-    void captureBaseline(String target, TurnToken token) {
-        CompletableFuture<Rendezvous.Resolution> waiter = token.waiter();
+    void captureBaseline(String target) {
+        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
         if (waiter == null) {
             inFlight.remove(target); // no send is waiting on this delivery — nothing to resolve later
             return;

bridged/src/main/java/dev/ltms/bridged/inject/Injector.java

@@ -2,7 +2,6 @@ package dev.ltms.bridged.inject;
 
 import dev.ltms.bridged.herdr.AgentControl;
 import dev.ltms.bridged.herdr.AgentStatus;
-import dev.ltms.bridged.msg.TurnToken;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -130,7 +129,7 @@ public final class Injector {
     }
 
     /** A pending message and the future that completes when it has been delivered. */
-    private record Pending(String text, TurnToken token, CompletableFuture<Void> delivered) {
+    private record Pending(String text, CompletableFuture<Void> delivered) {
     }
 
     /** Per-worker delivery state, guarded by its own monitor (single writer per worker). */
@@ -160,9 +159,9 @@ public final class Injector {
      * <p>Uses an atomic map update so a concurrent {@link #drop} cannot slip between "find the
      * target" and "queue the message" and orphan it in a target it just removed.
      */
-    public CompletableFuture<Void> enqueue(String target, String text, TurnToken token) {
+    public CompletableFuture<Void> enqueue(String target, String text) {
         CompletableFuture<Void> delivered = new CompletableFuture<>();
-        Pending p = new Pending(text, token, delivered);
+        Pending p = new Pending(text, delivered);
         targets.compute(target, (_, existing) -> {
             Target t = (existing != null) ? existing : new Target();
             t.add(p); // synchronized on the Target monitor — atomic with a concurrent drop
@@ -357,7 +356,7 @@ public final class Injector {
             } else {
                 // Baseline the pane's pre-turn content so a misattributed completion (no new output)
                 // can't resolve this send with the previous turn's stale answer (CB-115).
-                turnListener.onDelivered(target, sent.token());
+                turnListener.onDelivered(target);
                 sent.delivered().complete(null);
             }
         }

bridged/src/main/java/dev/ltms/bridged/inject/TurnListener.java

@@ -1,7 +1,5 @@
 package dev.ltms.bridged.inject;
 
-import dev.ltms.bridged.msg.TurnToken;
-
 /**
  * Notified when a worker's delegated turn is observed to complete — a confirmed
  * {@code WORKING → IDLE} transition after a delivery. This is the CB-106 completion signal the
@@ -59,7 +57,7 @@ public interface TurnListener {
      * resolve the send with the previous turn's stale answer. A default no-op keeps the interface
      * functional for callers that don't scrape.
      */
-    default void onDelivered(String target, TurnToken token) {
+    default void onDelivered(String target) {
     }
 
     /** No-op default for callers that only need delivery, not completion signalling. */

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

@@ -173,6 +173,9 @@ public final class MessageService {
     private final ConcurrentHashMap<String, Task> tasks = new ConcurrentHashMap<>();
     /** Async tasks that have accepted delivery for a target. */
     private final ConcurrentHashMap<String, Set<Task>> asyncTasksByTarget = new ConcurrentHashMap<>();
+    /** Async task that owns each exact forward rendezvous waiter. */
+    private final ConcurrentHashMap<CompletableFuture<Rendezvous.Resolution>, Task> asyncTasksByWaiter =
+            new ConcurrentHashMap<>();
     /** Async tickets paused on a specific {@code bridge_ask} turn. */
     private final ConcurrentHashMap<String, Task> asyncTasksByTurn = new ConcurrentHashMap<>();
     private final AtomicLong ticketSeq = new AtomicLong();
@@ -385,14 +388,16 @@ public final class MessageService {
             // failed send leaves no stale waiter behind.
             CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target);
             try {
-                TurnToken token = new TurnToken(target, reply);
+                if (task != null) {
+                    asyncTasksByWaiter.put(reply, task);
+                }
                 // The send has won the lock; the accepted-delivery hook records delegator ownership
                 // here (CB-548). It runs BEFORE enqueue so a throwing hook — onAccepted is now a
                 // public callback — fails the send without queuing a message that would orphan.
                 if (onAccepted != null) {
                     onAccepted.run();
                 }
-                CompletableFuture<Void> delivered = injector.enqueue(target, content, token);
+                CompletableFuture<Void> delivered = injector.enqueue(target, content);
                 try {
                     Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
                     return recorded(new Reply(outcomeOf(r.kind()), r.text(), r.turnId()));
@@ -409,6 +414,7 @@ public final class MessageService {
                     throw new IllegalStateException("interrupted awaiting reply from " + target, e);
                 }
             } finally {
+                asyncTasksByWaiter.remove(reply);
                 rendezvous.close(target, reply);
             }
         } finally {
@@ -434,11 +440,15 @@ public final class MessageService {
         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.
+            CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(workerSession);
+            Task task = markAsyncQuestion(waiter, question, ticket.turnId());
             if (!rendezvous.resolveQuestion(workerSession, question, ticket.turnId())) {
+                if (task != null) {
+                    clearAsyncQuestion(ticket.turnId(), true);
+                }
                 rendezvous.closeAsk(ticket.turnId());
                 return new AskResult(AskOutcome.NO_WAITER, null); // no primary is blocked on this worker
             }
-            markAsyncQuestion(workerSession, question, ticket.turnId());
         }
         try {
             String answer = ticket.answer().get(timeoutMillis, TimeUnit.MILLISECONDS);
@@ -614,17 +624,14 @@ public final class MessageService {
     }
 
     /** Record the active question for an async ticket; blocking sends have no entry and stay unchanged. */
-    private void markAsyncQuestion(String target, String text, String turnId) {
-        Set<Task> targetTasks = asyncTasksByTarget.get(target);
-        Task task = targetTasks == null ? null : targetTasks.stream()
-                .filter(candidate -> !candidate.future.isDone())
-                .findFirst()
-                .orElse(null);
+    private Task markAsyncQuestion(CompletableFuture<Rendezvous.Resolution> waiter, String text, String turnId) {
+        Task task = waiter == null ? null : asyncTasksByWaiter.get(waiter);
         if (task != null) {
             task.question = new Reply(Outcome.QUESTION, text, turnId);
             task.turnId = turnId;
             asyncTasksByTurn.put(turnId, task);
         }
+        return task;
     }
 
     /** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
@@ -635,6 +642,7 @@ public final class MessageService {
             if (forgetTurn) {
                 asyncTasksByTurn.remove(turnId, task);
                 task.turnId = null;
+                untrackAsyncTarget(task);
             }
         }
     }

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

@@ -1,20 +0,0 @@
-package dev.ltms.bridged.msg;
-
-import java.util.concurrent.CompletableFuture;
-
-/**
- * Identity for one accepted send. The session turn is deliberately absent: CompletionResolver's
- * delivery callback runs before SessionManager.onDelivered, so binding it needs a later ordering design.
- */
-public final class TurnToken {
-    private final String target;
-    private final CompletableFuture<Rendezvous.Resolution> waiter;
-
-    public TurnToken(String target, CompletableFuture<Rendezvous.Resolution> waiter) {
-        this.target = target;
-        this.waiter = waiter;
-    }
-
-    public String target() { return target; }
-    public CompletableFuture<Rendezvous.Resolution> waiter() { return waiter; }
-}

bridged/src/main/java/dev/ltms/bridged/session/SessionManager.java

@@ -4,7 +4,6 @@ import dev.ltms.bridged.auth.MemberLifecycle;
 import dev.ltms.bridged.herdr.Agent;
 import dev.ltms.bridged.inject.TurnListener;
 import dev.ltms.bridged.inject.MemberPresence;
-import dev.ltms.bridged.msg.TurnToken;
 import dev.ltms.bridged.peer.MemberRole;
 import dev.ltms.bridged.peer.PeerHandle;
 import dev.ltms.bridged.peer.PeerLauncher;
@@ -426,7 +425,7 @@ public final class SessionManager implements TurnListener {
      * can be re-delivered for multi-turn reuse until it is released.
      */
     @Override
-    public void onDelivered(String target, TurnToken token) {
+    public void onDelivered(String target) {
         MemberSession current = findByTerminal(target);
         if (current == null) return;
         if (current.state() != MemberSession.State.READY && current.state() != MemberSession.State.DONE) {

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

@@ -10,6 +10,9 @@ import dev.ltms.bridged.inject.Injector;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -676,6 +679,53 @@ class MessageServiceTest {
         assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
     }
 
+    @Test
+    void unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget() throws Exception {
+        String ticket = messages.sendAsync(T, "task that asks");
+        awaitWaiting();
+        injectDelivery();
+
+        assertEquals(MessageService.AskOutcome.TIMED_OUT,
+                messages.ask(T, "which config?", 200).outcome());
+        assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase(),
+                "only the question wait ended; the delegated turn may still finish");
+
+        String next = messages.sendAsync(T, "next task");
+        awaitWaiting();
+        assertTrue(rendezvous.resolve(T, "done"));
+        assertEquals(MessageService.Phase.DONE, awaitTicketPhase(next, MessageService.Phase.DONE).phase());
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    void asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter() throws Exception {
+        String first = messages.sendAsync(T, "first task");
+        awaitWaiting();
+
+        // Model the resolveQuestion/markAsyncQuestion race: another accepted task reached the target set.
+        Field byTargetField = MessageService.class.getDeclaredField("asyncTasksByTarget");
+        byTargetField.setAccessible(true);
+        Map<String, Set<Object>> byTarget = (Map<String, Set<Object>>) byTargetField.get(messages);
+        Set<Object> targetTasks = byTarget.get(T);
+        Class<?> taskClass = Class.forName(MessageService.class.getName() + "$Task");
+        var constructor = taskClass.getDeclaredConstructor(String.class);
+        constructor.setAccessible(true);
+        targetTasks.clear();
+        targetTasks.add(constructor.newInstance(T));
+
+        CompletableFuture<MessageService.AskResult> ask =
+                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
+        assertEquals(MessageService.Phase.ASKING, awaitTicketPhase(first, MessageService.Phase.ASKING).phase());
+
+        MessageService.TaskView asking = messages.poll(first);
+        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
+                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
+        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
+        awaitWaiting();
+        assertTrue(rendezvous.resolve(T, "done"));
+        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
+    }
+
     private void assertFailedTicket(String ticket, String reason) throws Exception {
         MessageService.TaskView view = awaitTicketPhase(ticket, MessageService.Phase.FAILED);
         assertEquals(reason, view.detail());