CB-573: carry accepted turn tokens on delivery

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

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

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

@@ -2,6 +2,7 @@ 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;
 
@@ -83,17 +84,17 @@ public final class CompletionResolver implements TurnListener {
     }
 
     @Override
-    public void onDelivered(String target) {
+    public void onDelivered(String target, TurnToken token) {
         // 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);
+        captureBaseline(target, token);
     }
 
     /** Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link #onDelivered}). */
-    void captureBaseline(String target) {
-        CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(target);
+    void captureBaseline(String target, TurnToken token) {
+        CompletableFuture<Rendezvous.Resolution> waiter = token.waiter();
         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,6 +2,7 @@ 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;
 
@@ -129,7 +130,7 @@ public final class Injector {
     }
 
     /** A pending message and the future that completes when it has been delivered. */
-    private record Pending(String text, CompletableFuture<Void> delivered) {
+    private record Pending(String text, TurnToken token, CompletableFuture<Void> delivered) {
     }
 
     /** Per-worker delivery state, guarded by its own monitor (single writer per worker). */
@@ -159,9 +160,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) {
+    public CompletableFuture<Void> enqueue(String target, String text, TurnToken token) {
         CompletableFuture<Void> delivered = new CompletableFuture<>();
-        Pending p = new Pending(text, delivered);
+        Pending p = new Pending(text, token, 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
@@ -356,7 +357,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);
+                turnListener.onDelivered(target, sent.token());
                 sent.delivered().complete(null);
             }
         }

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

@@ -1,5 +1,7 @@
 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
@@ -57,7 +59,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) {
+    default void onDelivered(String target, TurnToken token) {
     }
 
     /** No-op default for callers that only need delivery, not completion signalling. */

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

@@ -173,9 +173,6 @@ 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();
@@ -388,16 +385,14 @@ public final class MessageService {
             // failed send leaves no stale waiter behind.
             CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target);
             try {
-                if (task != null) {
-                    asyncTasksByWaiter.put(reply, task);
-                }
+                TurnToken token = new TurnToken(target, reply);
                 // 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);
+                CompletableFuture<Void> delivered = injector.enqueue(target, content, token);
                 try {
                     Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
                     return recorded(new Reply(outcomeOf(r.kind()), r.text(), r.turnId()));
@@ -414,7 +409,6 @@ public final class MessageService {
                     throw new IllegalStateException("interrupted awaiting reply from " + target, e);
                 }
             } finally {
-                asyncTasksByWaiter.remove(reply);
                 rendezvous.close(target, reply);
             }
         } finally {
@@ -440,15 +434,11 @@ 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);
@@ -624,14 +614,17 @@ public final class MessageService {
     }
 
     /** Record the active question for an async ticket; blocking sends have no entry and stay unchanged. */
-    private Task markAsyncQuestion(CompletableFuture<Rendezvous.Resolution> waiter, String text, String turnId) {
-        Task task = waiter == null ? null : asyncTasksByWaiter.get(waiter);
+    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);
         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. */
@@ -642,7 +635,6 @@ 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

@@ -0,0 +1,20 @@
+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,6 +4,7 @@ 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;
@@ -425,7 +426,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) {
+    public void onDelivered(String target, TurnToken token) {
         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,9 +10,6 @@ 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;
 
@@ -679,53 +676,6 @@ 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());