CB-574: retain async task correlation

A lead sent to sessionId "sol" — a profile name, not a terminal id. The
bridge accepted it, handed out a ticket, failed 60s later inside the
injector, and still reported the ticket as pending 20 minutes on. The
sender never learned anything and a whole delegation was lost.

bridge_send now rejects a target that exactly matches a configured
profile name, on both the blocking and the wait:false path, before any
ticket is issued. The error names the value and points at bridge_list.

The check is deliberately narrow. A target absent from the member roster
may still be a peer lead's terminal or a herdr-owned pane, so only a
value the bridge can prove is a profile is refused. profiles is a
required parameter on both send methods — the earlier revision kept
overloads that defaulted it to an empty set, which is the same silent
disable shape as CB-561.

bridged/src/main/java/dev/ltms/bridged/mcp/BridgeMcp.java

@@ -33,6 +33,7 @@ import jakarta.servlet.http.HttpServlet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -151,8 +152,8 @@ public final class BridgeMcp {
                     Runnable onAccepted = () -> primaryRegistry.recordDelegation(target, caller);
                     // wait defaults to true (block for the reply); wait:false is fire-and-poll.
                     return Boolean.FALSE.equals(a.get("wait"))
-                            ? sendAsync(messages, target, content, onAccepted)
-                            : send(messages, target, content, timeoutMs(a), onAccepted);
+                            ? sendAsync(messages, target, content, onAccepted, workers.profiles())
+                            : send(messages, target, content, timeoutMs(a), onAccepted, workers.profiles());
                 })
                 // bridge_reply's identity is the CONNECTION, never an argument — so the authz check
                 // is "is this caller a worker at all", and it can only ever reply as itself.
@@ -379,21 +380,22 @@ public final class BridgeMcp {
 
     // --- tool logic (thin adapters over the services; unit-testable) ---------------------------
 
-    /** {@code bridge_send}: delegate {@code content} to a worker session and block for its reply. */
-    static McpSchema.CallToolResult send(MessageService messages, String sessionId, String content, Long timeoutMs) {
-        return send(messages, sessionId, content, timeoutMs, null);
-    }
-
     /**
-     * As {@link #send(MessageService, String, String, Long)}, wiring an accepted-delivery hook
+     * {@code bridge_send}: delegate {@code content} to a worker session and block for its reply.
+     * The configured profiles are required so a profile name can never bypass target validation.
+     *
      * (CB-548): {@code onAccepted} records delegator ownership the instant the send is accepted, so
      * a BUSY interloper never claims a turn it did not win. {@code null} disables recording.
      */
     static McpSchema.CallToolResult send(MessageService messages, String sessionId, String content,
-                                         Long timeoutMs, Runnable onAccepted) {
+                                         Long timeoutMs, Runnable onAccepted, Set<String> profiles) {
         if (isBlank(sessionId) || isBlank(content)) {
             return error("sessionId and content are required");
         }
+        McpSchema.CallToolResult targetError = profileTargetError(sessionId, profiles);
+        if (targetError != null) {
+            return targetError;
+        }
         long timeout = clamp(timeoutMs == null ? DEFAULT_TIMEOUT_MS : timeoutMs);
         try {
             return formatReply(messages.send(sessionId, content, timeout, onAccepted), timeout);
@@ -463,24 +465,33 @@ public final class BridgeMcp {
     /**
      * {@code bridge_send} with {@code wait:false}: delegate {@code content} and return a ticket
      * immediately (fire-and-poll), so a long task isn't cut off by the caller's MCP call timeout.
-     */
-    static McpSchema.CallToolResult sendAsync(MessageService messages, String sessionId, String content) {
-        return sendAsync(messages, sessionId, content, null);
-    }
-
-    /**
-     * As {@link #sendAsync(MessageService, String, String)}, wiring the accepted-delivery hook
+     * The configured profiles are required so a profile name can never bypass target validation.
+     *
+     * This wires the accepted-delivery hook
      * (CB-548) so an async flooding send records delegator ownership exactly once it is accepted.
      */
     static McpSchema.CallToolResult sendAsync(MessageService messages, String sessionId, String content,
-                                              Runnable onAccepted) {
+                                              Runnable onAccepted, Set<String> profiles) {
         if (isBlank(sessionId) || isBlank(content)) {
             return error("sessionId and content are required");
         }
+        McpSchema.CallToolResult targetError = profileTargetError(sessionId, profiles);
+        if (targetError != null) {
+            return targetError;
+        }
         String ticket = messages.sendAsync(sessionId, content, onAccepted);
         return text("accepted — task delegated. Poll bridge_poll with ticket=" + ticket);
     }
 
+    /** A configured profile is never a send target; other unknown values may be herdr-owned panes. */
+    private static McpSchema.CallToolResult profileTargetError(String sessionId, Set<String> profiles) {
+        if (profiles.contains(sessionId)) {
+            return error("unknown send target \"" + sessionId + "\": it is a configured profile name, not a "
+                    + "session id. Call bridge_list to find a member or lead sessionId.");
+        }
+        return null;
+    }
+
     /** {@code bridge_poll}: check an async delegation by ticket, or drain a worker's inbox by target. */
     static McpSchema.CallToolResult poll(MessageService messages, String ticket, String target) {
         if (!isBlank(target)) {
@@ -502,6 +513,9 @@ public final class BridgeMcp {
                     ? "[done — worker finished without a structured bridge_reply; transcript tail follows]\n" + v.reply()
                     : v.reply());
             case PENDING -> text("[pending — " + v.detail() + "]");
+            case ASKING -> text("[question — worker is waiting for your answer]\n" + v.reply()
+                    + "\n\nAnswer it by calling bridge_send again with turnId=\"" + v.turnId()
+                    + "\" and content set to your answer; the worker resumes the same turn.");
             case FAILED -> text("[failed — " + v.detail() + "]");
         };
     }

bridged/src/main/java/dev/ltms/bridged/member/ClaudeCodeLauncher.java

@@ -181,8 +181,8 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
         // id via -r and passes no --session-id (the two conflict). Both are injected before the
         // model flag so --model keeps outranking the operator's own argv.
         // mutableArgv: argvWithBridge may hand back the profile's own (immutable) List.of when it
-        // has no MCP — session flags must be added into a list we own.
-        List<String> argv = mutableArgv(argvWithBridge(cfg));
+        // has neither MCP nor a charter — session flags must be added into a list we own.
+        List<String> argv = mutableArgv(argvWithBridge(cfg, spec.charter()));
         String agentSessionId = applySessionIdentity(argv, spec.sessionName(), spec.resumeSessionId());
         return new Launch(workerEnv, argvWithModel(argv, cfg), agentSessionId);
     }
@@ -217,22 +217,26 @@ public final class ClaudeCodeLauncher extends HerdrPeerLauncher {
     }
 
     /**
-     * The launch argv, plus — when {@code worker.mcpUrl} is set — inline {@code --mcp-config} for
-     * the bridge server and {@code --append-system-prompt} for the {@link #REPLY_CHARTER}. Neither
-     * touches the profile's config; both are pure command-line flags. This inline-flag mount is
-     * Claude Code specific — other adapters mount MCP and instructions their own way.
+     * The launch argv, plus an inline {@code --mcp-config} when {@code worker.mcpUrl} is set and
+     * {@code --append-system-prompt} when the base composed a charter. Neither touches the profile's
+     * config; both are pure command-line flags. This inline-flag mount is Claude Code specific —
+     * other adapters mount MCP and instructions their own way.
      */
-    private List<String> argvWithBridge(BridgedConfig.Profile cfg) {
-        if (!cfg.hasMcp()) {
+    private List<String> argvWithBridge(BridgedConfig.Profile cfg, String charter) {
+        if (!cfg.hasMcp() && charter == null) {
             return cfg.argv();
         }
-        String mcpJson = "{\"mcpServers\":{\"bridge\":{\"type\":\"http\",\"url\":\""
-                + cfg.mcpUrl() + "\"}}}";
         List<String> argv = mutableArgv(cfg.argv());
-        argv.add("--mcp-config");
-        argv.add(mcpJson);
-        argv.add("--append-system-prompt");
-        argv.add(REPLY_CHARTER);
+        if (cfg.hasMcp()) {
+            String mcpJson = "{\"mcpServers\":{\"bridge\":{\"type\":\"http\",\"url\":\""
+                    + cfg.mcpUrl() + "\"}}}";
+            argv.add("--mcp-config");
+            argv.add(mcpJson);
+        }
+        if (charter != null) {
+            argv.add("--append-system-prompt");
+            argv.add(charter);
+        }
         return argv;
     }
 

bridged/src/main/java/dev/ltms/bridged/member/OpenCodeLauncher.java

@@ -163,10 +163,10 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
      * {@inheritDoc}
      *
      * <p>Builds the opencode launch: no {@code ANTHROPIC_*} and no guard (opencode reads its own
-     * provider credentials); when the profile mounts the bridge MCP or has a member charter, generate an ephemeral
-     * {@code opencode.json} (remote MCP server + member-charter instructions) and point the worker at
-     * it via {@code OPENCODE_CONFIG}; carry the parity-neutral git-forge grant; and select the model
-     * with {@code -m}.
+     * provider credentials); when the profile mounts the bridge MCP or has a member charter,
+     * generate an ephemeral {@code opencode.json} (remote MCP server + member-charter instructions)
+     * and point the worker at it via {@code OPENCODE_CONFIG}; carry the parity-neutral git-forge
+     * grant; and select the model with {@code -m}.
      */
     @Override
     protected Launch buildLaunch(BridgedConfig.Profile cfg, LaunchSpec spec) {

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

@@ -127,6 +127,8 @@ public final class MessageService {
     public enum Phase {
         /** Delegated and in flight — queued for the worker or being worked. */
         PENDING,
+        /** The worker is paused in {@code bridge_ask}; {@link TaskView#reply} and {@link TaskView#turnId} identify it. */
+        ASKING,
         /** The worker's turn finished; {@link TaskView#reply} holds the answer. */
         DONE,
         /** The delegation could not complete (timed out, worker gone, or busy). */
@@ -136,16 +138,28 @@ public final class MessageService {
     /**
      * A poll snapshot of an async delegation.
      *
-     * @param reply       the answer when {@link #phase} is {@link Phase#DONE}, else {@code null}
+     * @param reply       the answer when {@link #phase} is {@link Phase#DONE}, or the question when
+     *                    {@link #phase} is {@link Phase#ASKING}; otherwise {@code null}
      * @param replySource {@code "reply"} (structured {@code bridge_reply}) or {@code "transcript"}
      *                    (completion scrape) when {@link Phase#DONE}, else {@code null}
-     * @param detail      a human note (live worker status while pending, or the failure reason)
+     * @param detail      a human note (live worker status while pending, ask state, or failure reason)
+     * @param turnId      correlation id for an {@link Phase#ASKING} ticket, else {@code null}
      */
-    public record TaskView(String ticket, Phase phase, String reply, String replySource, String detail) {
+    public record TaskView(String ticket, Phase phase, String reply, String replySource, String detail,
+                           String turnId) {
     }
 
     /** An in-flight or finished async delegation, keyed by its ticket. */
-    private record Task(String target, CompletableFuture<Reply> future, long createdNanos) {
+    private static final class Task {
+        private final String target;
+        private final CompletableFuture<Reply> future = new CompletableFuture<>();
+        private final long createdNanos = System.nanoTime();
+        private volatile Reply question;
+        private volatile String turnId;
+
+        private Task(String target) {
+            this.target = target;
+        }
     }
 
     private final AgentControl agents;
@@ -156,6 +170,10 @@ public final class MessageService {
     private final Metrics metrics; // CB-502: nullable — no registry in unit tests
     private final ConcurrentHashMap<String, ReentrantLock> sessionLocks = new ConcurrentHashMap<>();
     private final ConcurrentHashMap<String, Task> tasks = new ConcurrentHashMap<>();
+    /** The async task that currently owns a target's send lock. */
+    private final ConcurrentHashMap<String, Task> asyncTasksByTarget = 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();
     private final ExecutorService asyncExecutor = Executors.newThreadPerTaskExecutor(
             Thread.ofVirtual().name("bridge-async-", 0).factory());
@@ -333,6 +351,9 @@ public final class MessageService {
             return new Reply(Outcome.BUSY, null); // another send held the session the whole window
         }
         try {
+            if (hasAsyncQuestion(target)) {
+                return new Reply(Outcome.BUSY, null); // the worker's current turn is paused for its lead
+            }
             // Open the waiter BEFORE queueing delivery (CB-548). A fast reply — the worker already
             // injectable the instant we enqueue — otherwise arrives before the waiter is registered
             // and orphans into the inbox while this send blocks to the timeout (the enqueue-before-
@@ -393,12 +414,14 @@ public final class MessageService {
                 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);
             return new AskResult(AskOutcome.ANSWERED, answer);
         } catch (TimeoutException e) {
             log.debug("bridge_ask from {} went unanswered in {}ms", workerSession, timeoutMillis);
+            clearAsyncQuestion(ticket.turnId(), true);
             return new AskResult(AskOutcome.TIMED_OUT, null);
         } catch (ExecutionException e) {
             Throwable cause = e.getCause();
@@ -441,9 +464,12 @@ public final class MessageService {
                 rendezvous.close(workerSession, reply);
                 return new Reply(Outcome.STALE_TURN, null); // lapsed between the lookup and the unblock
             }
+            clearAsyncQuestion(turnId, false);
             try {
                 Rendezvous.Resolution r = reply.get(remainingMillis(deadlineNanos), TimeUnit.MILLISECONDS);
-                return new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
+                Reply result = new Reply(outcomeOf(r.kind()), r.text(), r.turnId());
+                finishAsyncTask(turnId, result);
+                return result;
             } catch (TimeoutException e) {
                 // The worker resumed but hasn't replied yet — no completion fallback arms an answered
                 // turn (it never re-entered the injector), so a silent worker rides out the window.
@@ -483,9 +509,27 @@ public final class MessageService {
      */
     public String sendAsync(String target, String content, Runnable onAccepted) {
         String ticket = "task-" + ticketSeq.incrementAndGet();
-        CompletableFuture<Reply> future = CompletableFuture.supplyAsync(
-                () -> send(target, content, ASYNC_TIMEOUT_MS, onAccepted), asyncExecutor);
-        tasks.put(ticket, new Task(target, future, System.nanoTime()));
+        Task task = new Task(target);
+        tasks.put(ticket, task);
+        asyncExecutor.submit(() -> {
+            try {
+                Runnable trackingAccepted = () -> {
+                    if (onAccepted != null) {
+                        onAccepted.run();
+                    }
+                    asyncTasksByTarget.put(target, task);
+                };
+                Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted);
+                if (result.outcome() == Outcome.QUESTION) {
+                    asyncTasksByTarget.remove(target, task);
+                } else {
+                    finishAsyncTask(task, result);
+                }
+            } catch (Throwable t) {
+                task.future.completeExceptionally(t);
+                asyncTasksByTarget.remove(target, task);
+            }
+        });
         pruneTerminalTickets();
         log.debug("async send {} -> {}", ticket, target);
         return ticket;
@@ -501,27 +545,32 @@ public final class MessageService {
         if (task == null) {
             return null;
         }
-        CompletableFuture<Reply> f = task.future();
+        CompletableFuture<Reply> f = task.future;
         if (!f.isDone()) {
-            return new TaskView(ticket, Phase.PENDING, null, null, "worker " + liveStatus(task.target()));
+            Reply question = task.question;
+            if (question != null) {
+                return new TaskView(ticket, Phase.ASKING, question.text(), null,
+                        "worker is waiting for your answer", question.turnId());
+            }
+            return new TaskView(ticket, Phase.PENDING, null, null, "worker " + liveStatus(task.target), null);
         }
         Reply r;
         try {
             r = f.getNow(null);
         } catch (CompletionException | java.util.concurrent.CancellationException e) {
             Throwable cause = (e instanceof CompletionException ce && ce.getCause() != null) ? ce.getCause() : e;
-            return new TaskView(ticket, Phase.FAILED, null, null, cause.getMessage());
+            return new TaskView(ticket, Phase.FAILED, null, null, cause.getMessage(), null);
         }
         if (r.completed()) {
             String source = r.outcome() == Outcome.REPLIED ? "reply" : "transcript";
-            return new TaskView(ticket, Phase.DONE, r.text(), source, null);
+            return new TaskView(ticket, Phase.DONE, r.text(), source, null, null);
         }
         // A wedged worker (CB-109) carries the error context as its reason; the timeout/busy
         // outcomes carry none, so fall back to the outcome name.
         String detail = r.outcome() == Outcome.WORKER_FAILED && r.text() != null
                 ? r.text()
                 : "no reply — " + r.outcome().name().toLowerCase();
-        return new TaskView(ticket, Phase.FAILED, null, null, detail);
+        return new TaskView(ticket, Phase.FAILED, null, null, detail, null);
     }
 
     /** Best-effort live worker status for a pending poll; never throws (a lookup error is just noise). */
@@ -536,7 +585,51 @@ public final class MessageService {
     /** Drop finished tickets older than the TTL so the registry cannot grow without bound. */
     private void pruneTerminalTickets() {
         long cutoff = System.nanoTime() - TICKET_TTL_NANOS;
-        tasks.values().removeIf(t -> t.future().isDone() && t.createdNanos() < cutoff);
+        tasks.values().removeIf(t -> t.future.isDone() && t.createdNanos < cutoff);
+    }
+
+    /** 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) {
+        Task task = asyncTasksByTarget.get(target);
+        if (task != null) {
+            task.question = new Reply(Outcome.QUESTION, text, turnId);
+            task.turnId = turnId;
+            asyncTasksByTurn.put(turnId, task);
+        }
+    }
+
+    /** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
+    private void clearAsyncQuestion(String turnId, boolean forgetTurn) {
+        Task task = asyncTasksByTurn.get(turnId);
+        if (task != null && turnId.equals(task.turnId)) {
+            task.question = null;
+            if (forgetTurn) {
+                asyncTasksByTurn.remove(turnId, task);
+                task.turnId = null;
+            }
+        }
+    }
+
+    /** Complete and detach an async ticket after its worker's actual terminal reply. */
+    private void finishAsyncTask(Task task, Reply result) {
+        task.future.complete(result);
+        asyncTasksByTarget.remove(task.target, task);
+        if (task.turnId != null) {
+            asyncTasksByTurn.remove(task.turnId, task);
+        }
+    }
+
+    /** Complete the async ticket correlated to a specific answered turn. */
+    private void finishAsyncTask(String turnId, Reply result) {
+        Task task = asyncTasksByTurn.get(turnId);
+        if (task != null) {
+            finishAsyncTask(task, result);
+        }
+    }
+
+    /** A new send must not open a waiter while an async ticket owns this worker's paused turn. */
+    private boolean hasAsyncQuestion(String target) {
+        return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));
     }
 
     /** Release the async executor. */

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

@@ -53,6 +53,18 @@ class BridgeMcpTest {
         return ((McpSchema.TextContent) r.content().getFirst()).text();
     }
 
+    private void assertSendRoundTrips(String target, Set<String> profiles) throws Exception {
+        CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.send(messages, target, "hi", 4000L, null, profiles));
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting(target) && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting(target), "send should be accepted for " + target);
+        BridgeMcp.reply(messages, target, "received");
+        assertEquals("received", textOf(send.get(6, TimeUnit.SECONDS)));
+    }
+
     private static ClaudeCodeLauncher workerService(FakeHerdr h, String baseUrl, Set<String> allow) {
         BridgedConfig.Profile cfg = new BridgedConfig.Profile(
                 "ltms-local", baseUrl, "coder", null, "BRIDGED_WORKER_TOKEN", null,
@@ -69,7 +81,7 @@ class BridgeMcpTest {
     void sendThenReplyRoundTrips() throws Exception {
         // bridge_send blocks; bridge_reply resolves it with the worker's structured answer.
         CompletableFuture<McpSchema.CallToolResult> send = CompletableFuture.supplyAsync(
-                () -> BridgeMcp.send(messages, "term_a", "review this", 4000L));
+                () -> BridgeMcp.send(messages, "term_a", "review this", 4000L, null, Set.of()));
 
         // Wait until the send has opened its waiter so the reply resolves it (CB-307: reply now
         // queues in the inbox if no waiter is open, which would break the round-trip).
@@ -91,7 +103,7 @@ class BridgeMcpTest {
     @Test
     void asyncSendReturnsATicketThenPollReportsTheReply() throws Exception {
         // wait:false parity — a ticket is issued, resolved by a reply, and surfaced by bridge_poll.
-        McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it");
+        McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
         assertNotEquals(Boolean.TRUE, accepted.isError());
         String out = textOf(accepted);
         assertTrue(out.contains("ticket="), out);
@@ -120,6 +132,131 @@ class BridgeMcpTest {
         assertEquals("async LGTM", textOf(polled));
     }
 
+    @Test
+    void asyncSendSurfacesAnAskThenKeepsTheTicketForTheFinalReply() throws Exception {
+        McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
+        String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
+
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"));
+
+        CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.ask(messages, "term_a", "which config?", 5000L));
+        McpSchema.CallToolResult question = BridgeMcp.poll(messages, ticket, null);
+        deadline = System.currentTimeMillis() + 3000;
+        while (!textOf(question).contains("[question") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            question = BridgeMcp.poll(messages, ticket, null);
+        }
+        assertTrue(textOf(question).contains("which config?"), textOf(question));
+        String questionText = textOf(question);
+        String afterTurnId = questionText.substring(questionText.indexOf("turnId=\"") + "turnId=\"".length());
+        String turnId = afterTurnId.substring(0, afterTurnId.indexOf('"'));
+
+        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.answer(messages, turnId, "config.yaml", 5000L));
+        assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
+
+        deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"));
+        BridgeMcp.reply(messages, "term_a", "done");
+        assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
+
+        McpSchema.CallToolResult done = BridgeMcp.poll(messages, ticket, null);
+        deadline = System.currentTimeMillis() + 3000;
+        while (!"done".equals(textOf(done)) && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            done = BridgeMcp.poll(messages, ticket, null);
+        }
+        assertEquals("done", textOf(done));
+    }
+
+    @Test
+    void unansweredAsyncAskReturnsTheTicketToPending() throws Exception {
+        McpSchema.CallToolResult accepted = BridgeMcp.sendAsync(messages, "term_a", "do it", null, Set.of());
+        String ticket = textOf(accepted).substring(textOf(accepted).indexOf("ticket=") + "ticket=".length()).trim();
+
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"));
+
+        McpSchema.CallToolResult ask = BridgeMcp.ask(messages, "term_a", "still there?", 50L);
+        assertTrue(textOf(ask).contains("no answer"), textOf(ask));
+        assertTrue(textOf(BridgeMcp.poll(messages, ticket, null)).startsWith("[pending"));
+
+        BridgeMcp.reply(messages, "term_a", "finished after timeout");
+        assertEquals("finished after timeout", messages.drainReplies("term_a").getFirst().content());
+    }
+
+    @Test
+    void asyncSendFailureDoesNotLeaveItsTicketPending() throws Exception {
+        String ticket = messages.sendAsync("term_a", "do it", () -> {
+            throw new IllegalStateException("accept failed");
+        });
+
+        long deadline = System.currentTimeMillis() + 3000;
+        MessageService.TaskView view = messages.poll(ticket);
+        while (view.phase() == MessageService.Phase.PENDING && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            view = messages.poll(ticket);
+        }
+        assertEquals(MessageService.Phase.FAILED, view.phase());
+        assertEquals("accept failed", view.detail());
+    }
+
+    @Test
+    void anotherAsyncTicketCannotCaptureAReplyWhileTheFirstTicketIsAsking() throws Exception {
+        McpSchema.CallToolResult firstAccepted = BridgeMcp.sendAsync(messages, "term_a", "first", null, Set.of());
+        String firstTicket = textOf(firstAccepted).substring(textOf(firstAccepted).indexOf("ticket=") + "ticket=".length()).trim();
+
+        long deadline = System.currentTimeMillis() + 3000;
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        assertTrue(rendezvous.isWaiting("term_a"));
+
+        CompletableFuture<McpSchema.CallToolResult> ask = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.ask(messages, "term_a", "which config?", 5000L));
+        MessageService.TaskView first = messages.poll(firstTicket);
+        deadline = System.currentTimeMillis() + 3000;
+        while (first.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            first = messages.poll(firstTicket);
+        }
+        assertEquals(MessageService.Phase.ASKING, first.phase());
+        String firstTurnId = first.turnId();
+
+        McpSchema.CallToolResult secondAccepted = BridgeMcp.sendAsync(messages, "term_a", "second", null, Set.of());
+        String secondTicket = textOf(secondAccepted).substring(textOf(secondAccepted).indexOf("ticket=") + "ticket=".length()).trim();
+        MessageService.TaskView second = messages.poll(secondTicket);
+        deadline = System.currentTimeMillis() + 3000;
+        while (second.phase() == MessageService.Phase.PENDING && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+            second = messages.poll(secondTicket);
+        }
+        assertEquals(MessageService.Phase.FAILED, second.phase());
+
+        BridgeMcp.reply(messages, "term_a", "late reply");
+        assertEquals("late reply", messages.drainReplies("term_a").getFirst().content());
+
+        CompletableFuture<McpSchema.CallToolResult> answer = CompletableFuture.supplyAsync(
+                () -> BridgeMcp.answer(messages, firstTurnId, "config.yaml", 5000L));
+        assertEquals("config.yaml", textOf(ask.get(6, TimeUnit.SECONDS)));
+        while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5);
+        }
+        BridgeMcp.reply(messages, "term_a", "done");
+        assertEquals("done", textOf(answer.get(6, TimeUnit.SECONDS)));
+    }
+
     @Test
     void pollUnknownTicketIsAnError() {
         McpSchema.CallToolResult res = BridgeMcp.poll(messages, "task-999", null);
@@ -129,15 +266,40 @@ class BridgeMcpTest {
 
     @Test
     void sendTimesOutWithAWorkingNote() {
-        McpSchema.CallToolResult res = BridgeMcp.send(messages, "term_a", "hi", 120L);
+        McpSchema.CallToolResult res = BridgeMcp.send(messages, "term_a", "hi", 120L, null, Set.of());
         assertNotEquals(Boolean.TRUE, res.isError(), "a timeout is informational, not a tool error");
         assertTrue(textOf(res).contains("no reply"), "got: " + textOf(res));
     }
 
     @Test
     void sendRejectsMissingArgs() {
-        assertTrue(BridgeMcp.send(messages, null, "hi", null).isError());
-        assertTrue(BridgeMcp.send(messages, "term_a", "  ", null).isError());
+        assertTrue(BridgeMcp.send(messages, null, "hi", null, null, Set.of()).isError());
+        assertTrue(BridgeMcp.send(messages, "term_a", "  ", null, null, Set.of()).isError());
+    }
+
+    @Test
+    void sendRejectsAConfiguredProfileNameBeforeAcceptingIt() {
+        McpSchema.CallToolResult blocking = BridgeMcp.send(messages, "sol", "hi", 100L, null, Set.of("sol"));
+        McpSchema.CallToolResult async = BridgeMcp.sendAsync(messages, "sol", "hi", null, Set.of("sol"));
+
+        assertTrue(blocking.isError());
+        assertTrue(async.isError());
+        assertTrue(textOf(blocking).contains("sol"));
+        assertTrue(textOf(blocking).contains("configured profile name"));
+        assertTrue(textOf(blocking).contains("bridge_list"));
+        assertFalse(textOf(async).contains("ticket="));
+    }
+
+    @Test
+    void sendAllowsPeerLeadMemberAndUnclassifiedTargets() throws Exception {
+        Set<String> profiles = Set.of("sol");
+
+        assertSendRoundTrips("term_peer_lead", profiles);
+        assertSendRoundTrips("term_live_member", profiles);
+
+        // A herdr-owned pane outside the bridge roster cannot be classified at accept time.
+        McpSchema.CallToolResult result = BridgeMcp.send(messages, "external-pane", "hi", 10L, null, profiles);
+        assertFalse(result.isError(), "an unclassified target must not be rejected at acceptance time");
     }
 
     @Test
@@ -173,7 +335,7 @@ class BridgeMcpTest {
     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));
+                () -> BridgeMcp.send(messages, "term_a", "do X", 5000L, null, Set.of()));
         long deadline = System.currentTimeMillis() + 3000;
         while (!rendezvous.isWaiting("term_a") && System.currentTimeMillis() < deadline) {
             //noinspection BusyWait

bridged/src/test/java/dev/ltms/bridged/member/ClaudeCodeLauncherTest.java

@@ -102,6 +102,53 @@ class ClaudeCodeLauncherTest {
                 "the operator's own args are preserved, in order, ahead of the model flag");
     }
 
+    @Test
+    void appendsTheBaseComposedRoleAndReplyCharter() {
+        FakeHerdr herdr = new FakeHerdr();
+        String roleCharter = "You review changes.";
+        BridgedConfig.Profile cfg = new BridgedConfig.Profile(
+                "sonnet", "http://gx00.gw:8000", null, null, "BRIDGED_WORKER_TOKEN",
+                List.of("claude"), "tab", "bridged-workers", "w #{n}",
+                "http://127.0.0.1:8765/mcp", null, null);
+        ClaudeCodeLauncher svc = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null,
+                0, 0L, () -> fleet(Map.of("reviewer", roleCharter), null));
+
+        svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.REVIEWER));
+
+        List<String> args = spawnedArgs(herdr);
+        int flag = args.indexOf("--append-system-prompt");
+        assertEquals(1, args.stream().filter("--append-system-prompt"::equals).count(),
+                "the composed charter is passed once");
+        assertEquals(roleCharter + "\n\n" + HerdrPeerLauncher.REPLY_CHARTER, args.get(flag + 1),
+                "the role charter comes first and the reply rule comes last");
+    }
+
+    @Test
+    void profileWithoutMcpOrRoleCharterGetsNoSystemPrompt() {
+        FakeHerdr herdr = new FakeHerdr();
+        ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(Map.of(), null));
+
+        svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.DEV));
+
+        assertFalse(spawnedArgs(herdr).contains("--append-system-prompt"));
+    }
+
+    @Test
+    void profileWithoutMcpStillGetsItsRoleCharter() {
+        FakeHerdr herdr = new FakeHerdr();
+        String roleCharter = "You design changes.";
+        ClaudeCodeLauncher svc = labelService(herdr, () -> fleet(Map.of("architect", roleCharter), null));
+
+        svc.spawn(new SpawnRequest("sonnet", null, null, null, null, MemberRole.ARCHITECT));
+
+        List<String> args = spawnedArgs(herdr);
+        int flag = args.indexOf("--append-system-prompt");
+        assertTrue(flag >= 0, "a role charter does not need an MCP mount");
+        assertEquals(roleCharter, args.get(flag + 1));
+        assertFalse(args.contains("--mcp-config"));
+    }
+
     private ClaudeCodeLauncher multiProfile(FakeHerdr herdr) {
         BridgedConfig.Profile gx10 = new BridgedConfig.Profile("gx10", "http://gx10.gw:8000", "coder",
                 null, "BRIDGED_WORKER_TOKEN", List.of("claude"), "tab", "bridged-workers", "w #{n}", null, null, null);