CB-604: reject an unrecognized profile kind at config load

Most of issue #65 was already shipped in 5d5b3bd - MemberSession
records agentSessionId, PeerHandle.agentSessionId() has no default, the
roster exposes it, and bridge_spawn accepts sessionName and
resumeSessionId. That commit left one item for follow-up: carrying the
id on ReleaseDetail.

This does that item. CB-578 stage C already lets a lead re-dispatch onto
the same worktree after a failure; without the session id that is a cold
start. With it, the work and the thread both survive.

Also updates the bridge_spawn and bridge_list rows in the CLAUDE.md
intent table, which the earlier commit missed.

Verified here: trial merge onto main builds 830 tests BUILD SUCCESS,
unpiped. Confirmed against the code that items 1-4 really were already
on main, so the scope-down is correct rather than work skipped.

Closes #65

CLAUDE.md

@@ -105,8 +105,8 @@ the merge — and merging on a reviewer's word is delegating it by proxy.
 |---|---|
 | Confirm your own role | `bridge_whoami` |
 | See backends available | `bridge_profiles` |
-| Start a member | `bridge_spawn{role?, profile?, cwd?, worktree?, ticket?}` → `sessionId` + `paneId` |
-| See the fleet | `bridge_list` → `leads` (your peers) + `members` · one peer's state: `bridge_status{sessionId}` |
+| Start a member | `bridge_spawn{role?, profile?, cwd?, worktree?, ticket?, sessionName?, resumeSessionId?}` → `sessionId` + `paneId` |
+| See the fleet | `bridge_list` → `leads` (your peers) + `members` (each carries `agentSessionId` when its backend knows one) · one peer's state: `bridge_status{sessionId}` |
 | Delegate (blocking) | `bridge_send{sessionId, content}` |
 | Delegate (long task) | `bridge_send{sessionId, content, wait:false}` → ticket → `bridge_poll{ticket}` |
 | Answer a member's `bridge_ask` | `bridge_send{turnId, content}` — **not** `sessionId` |

bridged/bridged.example.yaml

@@ -275,6 +275,26 @@ profiles:
   #     argv: ["opencode"]
 # How an unqualified spawn chooses a profile: fixed (default, reproduces pre-CB-518 behaviour),
 # round-robin, or weighted. Omitting this key is a strict no-op for existing configs.
+#
+# `weighted` IS NOT "cheapest first" — read this before you set weights (CB-589).
+# It is smooth weighted round-robin: it spreads spawns across EVERY profile that has a free slot,
+# in weight ratio. It has no idea which profile costs money. So with local:10 / paid:2 you do not
+# get "use local, overflow to paid" — you get roughly one spawn in six going to the paid profile
+# while the local box still has a free slot.
+#
+# There is a sharper second effect. The policy's running score map lives for the daemon's whole
+# life. While a profile is at maxLoad it is filtered out and its score FREEZES, so the paid
+# profiles keep accumulating against it. When the local slot frees up it returns with a stale
+# score and can LOSE the next pick — a paid spawn while the free box sits idle.
+#
+# Until a real cost-first policy exists, the workaround is to make the ratio decisive rather than
+# proportional: give the free profile a weight so large that it wins every pick it is eligible
+# for, and paid profiles only ever take genuine overflow. On this host that is local weight 100
+# against paid weights of ~1.
+#
+# The gotcha with that workaround: it expresses a PREFERENCE ORDER through a RATIO knob. Add a
+# future profile at weight 150 and it silently outranks the free box, with nothing to warn you.
+# Re-check the weights whenever you add a profile.
 placement: weighted
 
 # How long a credential sits out after a BACKEND_EXHAUSTED classification (CB-578 stage B), in

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

@@ -441,6 +441,11 @@ public final class Bridged {
                 reason += "; worktree=" + detail.worktreePath() + " branch=" + detail.branch()
                         + " snapshot=" + (detail.snapshotRef() != null ? detail.snapshotRef() : "none");
             }
+            // CB-584 (issue #65 criterion 5): also name the agent session, so a lead can resume the
+            // member's conversation instead of only re-dispatching a fresh one onto the same files.
+            if (detail.agentSessionId() != null) {
+                reason += " agentSessionId=" + detail.agentSessionId();
+            }
             messages.abandon(detail.terminalId(), reason);
             replyInbox.release(detail.terminalId());
             primaryRegistry.forgetDelegation(detail.terminalId()); // CB-532: don't leak the lead binding

bridged/src/main/java/dev/ltms/bridged/config/BridgedConfig.java

@@ -986,6 +986,7 @@ public record BridgedConfig(
             warnUnknownTopLevelKeys(yaml, path);
             rejectDuplicateMemberSlots(yaml);
             rejectNegativeMaxLoad(yaml);
+            rejectUnknownKind(yaml);
             BridgedConfig cfg = YAML.readValue(yaml, BridgedConfig.class);
             return cfg.withDefaults();
         } catch (IOException e) {
@@ -1280,6 +1281,53 @@ public record BridgedConfig(
         }
     }
 
+    /** The peer kinds this build has an adapter for — {@link Profile#kind()}'s only valid values. */
+    private static final Set<String> KNOWN_KINDS = Set.of(Profile.KIND_CLAUDE_CODE, Profile.KIND_OPENCODE);
+
+    /**
+     * Reject a profile whose {@code kind:} is not one of {@link #KNOWN_KINDS} (CB-604), naming the
+     * profile, the value it set, and the accepted set.
+     *
+     * <p>{@link Profile}'s compact constructor only lower-cases {@code kind} and compares it against
+     * {@code KIND_OPENCODE} — anything else, including a typo like {@code opencod}, silently falls
+     * into the claude-code bucket ({@link dev.ltms.bridged.member.CompositePeerLauncher} routes by
+     * exact adapter claim, not by membership in a known set). With {@code argv:} also unset, the argv
+     * default special-cases only the exact string {@code "claude-code"}, so the launch command falls
+     * back to {@code List.of(kind)} — the daemon then tries to run a program literally named after the
+     * typo. {@code CompositePeerLauncher}'s constructor already treats a profile claimed by two
+     * adapters as fatal (CB-402); an unrecognized kind is the same class of adapter-routing mistake
+     * and gets the same treatment here, at config load, rather than surfacing later as a failed spawn.
+     *
+     * @param yaml the raw config text
+     * @throws IllegalStateException when any profile's {@code kind} is a non-blank value not in
+     *                                {@link #KNOWN_KINDS} (case-insensitive)
+     */
+    static void rejectUnknownKind(String yaml) {
+        Map<?, ?> raw;
+        try {
+            raw = YAML.readValue(yaml, Map.class);
+        } catch (IOException | IllegalArgumentException e) {
+            return; // a malformed file is reported by the real parse, not here
+        }
+        if (raw == null || !(raw.get("profiles") instanceof Map<?, ?> profiles)) {
+            return;
+        }
+        List<String> bad = profiles.entrySet().stream()
+                .filter(e -> e.getValue() instanceof Map<?, ?> p
+                        && p.get("kind") instanceof String k && !k.isBlank()
+                        && !KNOWN_KINDS.contains(k.toLowerCase()))
+                .map(e -> String.valueOf(e.getKey()) + "=" + ((Map<?, ?>) e.getValue()).get("kind"))
+                .sorted()
+                .toList();
+        if (!bad.isEmpty()) {
+            throw new IllegalStateException("refusing to start: profile(s) [" + String.join(", ", bad)
+                    + "] set an unrecognized kind — accepted values are "
+                    + String.join(", ", KNOWN_KINDS.stream().sorted().toList())
+                    + " (case-insensitive); an unrecognized kind would otherwise fall back to the"
+                    + " claude-code adapter and try to launch a program named after the typo.");
+        }
+    }
+
     static List<String> unknownTopLevelKeys(String yaml) {
         Map<?, ?> raw;
         try {

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

@@ -577,25 +577,13 @@ public final class BridgeMcp {
         return text("acknowledged " + msgId);
     }
 
-    /**
-     * {@code bridge_status}: the live lifecycle status of a worker session, plus — when the worker
-     * is paused mid-turn in an async {@code bridge_ask} (CB-582) — the open question and how to
-     * answer it, so a lead on its normal poll cadence does not need the ticket to notice.
-     */
+    /** {@code bridge_status}: the live lifecycle status of a worker session. */
     static McpSchema.CallToolResult status(MessageService messages, String sessionId) {
         if (isBlank(sessionId)) {
             return error("sessionId is required");
         }
         try {
-            String base = messages.status(sessionId).name().toLowerCase();
-            MessageService.PendingAsk ask = messages.pendingAsk(sessionId);
-            if (ask == null) {
-                return text(base);
-            }
-            return text(base + "\n\n[question — worker is waiting for your answer]\n" + ask.question()
-                    + "\n\nAnswer it by calling bridge_send again with turnId=\"" + ask.turnId()
-                    + "\" and content set to your answer; the worker resumes the same turn."
-                    + " (ticket " + ask.ticket() + ")");
+            return text(messages.status(sessionId).name().toLowerCase());
         } catch (HerdrException e) {
             return error("herdr error for session " + sessionId + ": " + e.getMessage());
         }

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

@@ -164,30 +164,18 @@ public final class MessageService {
 
     /** An in-flight or finished async delegation, keyed by its ticket. */
     private static final class Task {
-        private final String ticket;
         private final String target;
         private final CompletableFuture<Reply> future = new CompletableFuture<>();
         private final long createdNanos;
         private volatile Reply question;
         private volatile String turnId;
 
-        private Task(String ticket, String target, long createdNanos) {
-            this.ticket = ticket;
+        private Task(String target, long createdNanos) {
             this.target = target;
             this.createdNanos = createdNanos;
         }
     }
 
-    /**
-     * A worker session's currently-open {@code bridge_ask} question, surfaced so {@code bridge_status}
-     * can show it without the caller needing the ticket first (CB-582). Only covers async
-     * (fire-and-poll) delegations, which track the question on their {@link Task}; a blocking
-     * ({@code wait:true}) send already hands the question straight back to its own caller, so there is
-     * nothing hidden left for {@code bridge_status} to surface in that case.
-     */
-    public record PendingAsk(String ticket, String question, String turnId) {
-    }
-
     private final AgentControl agents;
     private final Injector injector;
     private final Rendezvous rendezvous;
@@ -212,11 +200,9 @@ public final class MessageService {
      * Create with an explicit {@link ReplyInbox} and optional {@link ReplyPushLoop}.
      *
      * @param pushLoop nullable — when non-null, the push loop is notified on the no-waiter reply
-     *                 branch ({@link #reply}) so it can nudge the primary to drain the inbox,
+     *                 branch ({@link #reply}) so it can nudge the primary to drain the inbox, and
      *                 (CB-588) whenever an async ticket started by {@link #sendAsync} reaches a
-     *                 terminal phase, whenever {@link #poll} hands a terminal ticket to its caller,
-     *                 and (CB-582) whenever an async ticket's worker pauses mid-turn in
-     *                 {@code bridge_ask} or that pause ends (answered or lapsed)
+     *                 terminal phase, and whenever {@link #poll} hands a terminal ticket to its caller
      */
     public MessageService(AgentControl agents, Injector injector, Rendezvous rendezvous,
                           ReplyInbox inbox, ReplyPushLoop pushLoop) {
@@ -509,15 +495,6 @@ public final class MessageService {
                 rendezvous.closeAsk(ticket.turnId());
                 return new AskResult(AskOutcome.NO_WAITER, null); // no primary is blocked on this worker
             }
-            // CB-582: the question just became visible via bridge_poll (Phase.ASKING) for an async
-            // (wait:false) delegation — nudge the lead's own pane the same way a terminal ticket does
-            // (CB-588), since the lead's normal poll cadence is minutes away and the reverse-rendezvous
-            // window (~55s, see BridgeMcp/BridgedApp) is far shorter. A blocking (wait:true) send has
-            // no Task and gets the question directly in its own reply, so task == null there — nothing
-            // to nudge.
-            if (task != null && pushLoop != null) {
-                pushLoop.onQuestionOpened(task.ticket, workerSession, ticket.turnId(), question);
-            }
         }
         try {
             String answer = ticket.answer().get(timeoutMillis, TimeUnit.MILLISECONDS);
@@ -612,7 +589,7 @@ public final class MessageService {
      */
     public String sendAsync(String target, String content, Runnable onAccepted) {
         String ticket = "task-" + ticketSeq.incrementAndGet();
-        Task task = new Task(ticket, target, nowNanos.getAsLong());
+        Task task = new Task(target, nowNanos.getAsLong());
         tasks.put(ticket, task);
         if (pushLoop != null) {
             // CB-588: task.future only ever completes on a terminal phase (DONE or a failure) — a
@@ -738,12 +715,6 @@ public final class MessageService {
 
     /** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
     private void clearAsyncQuestion(String turnId, boolean forgetTurn) {
-        // CB-582: tell the push loop first — like ticketCollected, a removal for a turnId it never
-        // nudged about (or already dropped) is a harmless no-op, so this is safe to call unconditionally
-        // rather than threading the guard below through it.
-        if (pushLoop != null) {
-            pushLoop.questionClosed(turnId);
-        }
         Task task = asyncTasksByTurn.get(turnId);
         if (task != null && turnId.equals(task.turnId)) {
             task.question = null;
@@ -775,23 +746,6 @@ public final class MessageService {
         return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));
     }
 
-    /**
-     * The question {@code workerSession} is currently paused on via {@code bridge_ask}, if any
-     * (CB-582) — {@code bridge_status} uses this to show a pending question without the caller
-     * needing the ticket. {@code null} when the session has no open async question (including a
-     * session mid a <em>blocking</em> {@code bridge_ask}, which has no {@link Task} to look up — see
-     * {@link PendingAsk}).
-     */
-    public PendingAsk pendingAsk(String workerSession) {
-        for (Task task : tasks.values()) {
-            Reply q = task.question;
-            if (q != null && workerSession.equals(task.target)) {
-                return new PendingAsk(task.ticket, q.text(), q.turnId());
-            }
-        }
-        return null;
-    }
-
     /** Release the async executor. */
     public void close() {
         asyncExecutor.shutdown();

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

@@ -19,33 +19,30 @@ import java.util.stream.Collectors;
 
 /**
  * A status-gated push loop that nudges a lead's own herdr pane when it has uncollected work
- * waiting: a worker reply queued with no live {@code bridge_send} to resolve it (CB-307), an
+ * waiting: a worker reply queued with no live {@code bridge_send} to resolve it (CB-307), or an
  * async delegation ticket ({@code bridge_send(wait:false)}) that reached a terminal phase
- * (CB-588), or an async ticket's worker pausing mid-turn in {@code bridge_ask} to await an answer
- * (CB-582).
+ * (CB-588).
  *
- * <p><strong>CB-590: one schedule per lead.</strong> All three kinds of work are triggered
- * through their own entry point — {@link #onReplyQueued(String)},
- * {@link #onTicketTerminal(String, String, boolean)}, and
- * {@link #onQuestionOpened(String, String, String, String)} — but each resolves the lead that
- * should be nudged and coalesces onto a single per-lead reminder schedule, tracked in
- * {@link #activeLeads}. Earlier this was two independent schedules (one keyed by worker target
- * for replies, one keyed by lead for tickets) that could both decide to inject into the same pane
- * in the same window — a race, not routine behaviour, but the expensive kind: it interrupts the
- * lead's live turn twice. Collapsing to one schedule per lead makes that structurally impossible:
- * at most one scheduled tick chain is ever live for a given lead (guarded by {@link #activeLeads}'
+ * <p><strong>CB-590: one schedule per lead.</strong> Both kinds of work are triggered through
+ * their own entry point — {@link #onReplyQueued(String)} and
+ * {@link #onTicketTerminal(String, String, boolean)} — but both resolve the lead that should be
+ * nudged and coalesce onto a single per-lead reminder schedule, tracked in {@link #activeLeads}.
+ * Earlier this was two independent schedules (one keyed by worker target for replies, one keyed
+ * by lead for tickets) that could both decide to inject into the same pane in the same window —
+ * a race, not routine behaviour, but the expensive kind: it interrupts the lead's live turn
+ * twice. Collapsing to one schedule per lead makes that structurally impossible: at most one
+ * scheduled tick chain is ever live for a given lead (guarded by {@link #activeLeads}'
  * compare-and-set), so at most one {@code agents.send} to that lead's pane is ever in flight.
  *
  * <p>Each tick examines <em>everything</em> pending for that lead — reply targets whose inbox
- * still holds an unacked message ({@link #pendingReplies}), tickets not yet collected
- * ({@link #pendingTickets}), and open questions not yet answered or lapsed
- * ({@link #pendingQuestions}) — and sends at most one combined nudge per tick
- * ({@link #injectNudge(String, int, int, int)}). Work that arrives while the lead is busy is
- * never lost: it is re-read fresh on every tick until the lead is injectable or its own reminder
- * cap ({@link #maxReminders}) is reached — each source spends from its own budget, so one source
- * exhausting its cap does not stop nudges about the others (post-CB-590 regression fix; see
- * {@link #decide}) — whichever the durable inbox / pending set doesn't already answer via
- * {@code STOP}.
+ * still holds an unacked message ({@link #pendingReplies}) and tickets not yet collected
+ * ({@link #pendingTickets}) — and sends at most one combined nudge per tick
+ * ({@link #injectNudge(String, int, int)}). Work that arrives while the lead is busy is never
+ * lost: it is re-read fresh on every tick until the lead is injectable or its own reminder cap
+ * ({@link #maxReminders}) is reached — reply and ticket work each spend from their own budget, so
+ * one source exhausting its cap does not stop nudges about the other (post-CB-590 regression fix;
+ * see {@link #decide}) — whichever the durable inbox / pending-ticket set doesn't already answer
+ * via {@code STOP}.
  */
 public final class ReplyPushLoop {
 
@@ -60,14 +57,6 @@ public final class ReplyPushLoop {
     /** Coalesced form, several uncollected tickets for the same lead. */
     static final String TICKETS_NUDGE_FORMAT =
             "%d tickets finished%s — run bridge_poll(ticket=...) for each to collect them: %s";
-    /** Singular form, one worker paused mid-turn in bridge_ask (CB-582) — names the answer call directly. */
-    static final String QUESTION_NUDGE_FORMAT =
-            "Worker %s asked a question (ticket %s) — answer it with bridge_send(turnId=\"%s\", "
-                    + "content=...) to resume its turn:\n%s";
-    /** Coalesced form, several open questions for the same lead. */
-    static final String QUESTIONS_NUDGE_FORMAT =
-            "%d workers are paused on a question — run bridge_poll(ticket=...) for each, then answer "
-                    + "with bridge_send(turnId=..., content=...): %s";
 
     private final PrimaryRegistry primaryRegistry;
     private final AgentControl agents;
@@ -86,14 +75,7 @@ public final class ReplyPushLoop {
     private final ConcurrentHashMap<String, ReplyEntry> pendingReplies = new ConcurrentHashMap<>();
     /** Tickets that have gone terminal but not yet been polled, keyed by ticket. */
     private final ConcurrentHashMap<String, PendingTicket> pendingTickets = new ConcurrentHashMap<>();
-    /**
-     * Open {@code bridge_ask} questions not yet answered or lapsed, keyed by {@code turnId}
-     * (CB-582). A question's own nudge count is tracked the same per-item way as
-     * {@link #pendingTickets} (CB-598): a fresh question keeps its source eligible regardless of
-     * how depleted an older, still-open question's count is.
-     */
-    private final ConcurrentHashMap<String, PendingQuestion> pendingQuestions = new ConcurrentHashMap<>();
-    /** CB-590: leads with an active combined reminder schedule (replies and/or tickets and/or questions). */
+    /** CB-590: leads with an active combined reminder schedule (replies and/or tickets). */
     private final ConcurrentHashMap<String, Boolean> activeLeads = new ConcurrentHashMap<>();
 
     public ReplyPushLoop(PrimaryRegistry primaryRegistry, AgentControl agents, ReplyInbox inbox,
@@ -172,26 +154,6 @@ public final class ReplyPushLoop {
                 .collect(Collectors.toUnmodifiableSet());
     }
 
-    /**
-     * An open question awaiting the lead's answer: which ticket it belongs to, which worker asked,
-     * which lead to nudge, the question text, and how many nudges have named it so far (CB-598 —
-     * tracked per question, not per lead per source).
-     */
-    private record PendingQuestion(String turnId, String ticket, String target, String lead,
-                                   String question, int nudgeCount) {
-    }
-
-    /** Questions still open for {@code lead}, snapshotted fresh for one tick. */
-    private List<PendingQuestion> pendingQuestionsFor(String lead) {
-        return pendingQuestions.values().stream().filter(q -> lead.equals(q.lead())).toList();
-    }
-
-    /** Question turnIds still open for {@code lead} — a plain snapshot for race comparison. */
-    private Set<String> pendingQuestionTurnIdsFor(String lead) {
-        return pendingQuestionsFor(lead).stream().map(PendingQuestion::turnId)
-                .collect(Collectors.toUnmodifiableSet());
-    }
-
     /**
      * The reply-source reminder count {@link #decide} should see for {@code lead} on this tick:
      * the <em>minimum</em> nudge count among the reply targets currently pending for it (CB-598).
@@ -227,15 +189,6 @@ public final class ReplyPushLoop {
         return min == Integer.MAX_VALUE ? 0 : min;
     }
 
-    /** As {@link #minReplyNudgeCountFor}, for the question source (CB-582). */
-    private int minQuestionNudgeCountFor(String lead) {
-        int min = Integer.MAX_VALUE;
-        for (PendingQuestion q : pendingQuestionsFor(lead)) {
-            min = Math.min(min, q.nudgeCount());
-        }
-        return min == Integer.MAX_VALUE ? 0 : min;
-    }
-
     /**
      * Pure decision function: examine everything pending for {@code lead} — reply targets and
      * tickets alike — and return what the loop should do.
@@ -260,29 +213,15 @@ public final class ReplyPushLoop {
      * @return the action the caller should take
      */
     Action decide(String lead, int replyReminderCount, int ticketReminderCount) {
-        return decide(lead, replyReminderCount, ticketReminderCount, minQuestionNudgeCountFor(lead));
-    }
-
-    /**
-     * As {@link #decide(String, int, int)}, with the question source (CB-582) folded in on the
-     * same footing as replies and tickets: its own eligibility (has open questions AND under its
-     * own {@link #maxReminders} budget) is enough on its own to {@link Action#INJECT}, exactly like
-     * the other two.
-     *
-     * @param questionReminderCount the lowest nudge count among questions open for this lead
-     */
-    Action decide(String lead, int replyReminderCount, int ticketReminderCount, int questionReminderCount) {
         boolean hasReplyWork = !pendingReplyTargetsFor(lead).isEmpty();
         boolean hasTicketWork = !pendingTicketIdsFor(lead).isEmpty();
-        boolean hasQuestionWork = !pendingQuestionTurnIdsFor(lead).isEmpty();
-        if (!hasReplyWork && !hasTicketWork && !hasQuestionWork) {
+        if (!hasReplyWork && !hasTicketWork) {
             log.debug("push: nothing pending for lead {}, stopping reminder", lead);
             return Action.STOP;
         }
         boolean replyEligible = hasReplyWork && replyReminderCount < maxReminders;
         boolean ticketEligible = hasTicketWork && ticketReminderCount < maxReminders;
-        boolean questionEligible = hasQuestionWork && questionReminderCount < maxReminders;
-        if (!replyEligible && !ticketEligible && !questionEligible) {
+        if (!replyEligible && !ticketEligible) {
             log.debug("push: reminder cap ({}) reached for lead {} on every source with pending work, stopping",
                     maxReminders, lead);
             countNudge("exhausted");
@@ -361,39 +300,6 @@ public final class ReplyPushLoop {
         pendingTickets.remove(ticket);
     }
 
-    /**
-     * Called when an async ticket's worker pauses mid-turn in {@code bridge_ask} (CB-582): the
-     * question is now visible via {@code bridge_poll} (Phase.ASKING), but the reverse-rendezvous
-     * window it opened with (~55s default, see {@code BridgeMcp}/{@code BridgedApp}) is far shorter
-     * than a lead's normal minutes-long poll cadence — exactly the gap this closes. Resolves the
-     * delegating lead the same way {@link #onTicketTerminal} does and coalesces onto the same
-     * per-lead schedule (CB-590).
-     *
-     * @param ticket   the async ticket the question belongs to (for {@code bridge_poll})
-     * @param target   the worker session that asked
-     * @param turnId   correlation id the lead answers with ({@code bridge_send turnId=...})
-     * @param question the question text
-     */
-    public void onQuestionOpened(String ticket, String target, String turnId, String question) {
-        var lead = primaryRegistry.nudgeTargetFor(target);
-        if (lead.isEmpty()) {
-            log.debug("push: no lead is known to be waiting on {}'s question (turnId {}), skipping nudge",
-                    target, turnId);
-            return;
-        }
-        pendingQuestions.put(turnId, new PendingQuestion(turnId, ticket, target, lead.get(), question, 0));
-        startOrCoalesce(lead.get());
-    }
-
-    /**
-     * Called when a worker's {@code bridge_ask} resolves — answered or lapsed unanswered — so a
-     * scheduled tick never nudges about a question the lead already handled. A {@code turnId} that
-     * was never pending (never nudged, or already closed) is a no-op.
-     */
-    public void questionClosed(String turnId) {
-        pendingQuestions.remove(turnId);
-    }
-
     // --- the schedule ----------------------------------------------------------------------------
 
     /** Start a reminder schedule for {@code lead}, or join the one already running. */
@@ -424,19 +330,17 @@ public final class ReplyPushLoop {
     void tick(String lead) {
         Set<String> repliesBefore = pendingReplyTargetsFor(lead);
         Set<String> ticketsBefore = pendingTicketIdsFor(lead);
-        Set<String> questionsBefore = pendingQuestionTurnIdsFor(lead);
         int replyReminderCount = minReplyNudgeCountFor(lead);
         int ticketReminderCount = minTicketNudgeCountFor(lead);
-        int questionReminderCount = minQuestionNudgeCountFor(lead);
-        var action = decide(lead, replyReminderCount, ticketReminderCount, questionReminderCount);
+        var action = decide(lead, replyReminderCount, ticketReminderCount);
         switch (action) {
             case INJECT -> {
-                injectNudge(lead, replyReminderCount, ticketReminderCount, questionReminderCount);
+                injectNudge(lead, replyReminderCount, ticketReminderCount);
                 scheduleNext(lead);
             }
             // Re-check after the configured backoff; the lead may become injectable soon.
             case WAIT_BUSY -> scheduleNext(lead);
-            case STOP -> stopOrRestart(lead, repliesBefore, ticketsBefore, questionsBefore);
+            case STOP -> stopOrRestart(lead, repliesBefore, ticketsBefore);
         }
     }
 
@@ -471,20 +375,9 @@ public final class ReplyPushLoop {
      * side always wins; neither can miss the other, so this never loops on its own account.
      */
     void stopOrRestart(String lead, Set<String> repliesBefore, Set<String> ticketsBefore) {
-        stopOrRestart(lead, repliesBefore, ticketsBefore, pendingQuestionTurnIdsFor(lead));
-    }
-
-    /**
-     * As {@link #stopOrRestart(String, Set, Set)}, with the question source's (CB-582) own "before"
-     * snapshot folded into the same race check: a question that raced in during the
-     * decision-to-release window reclaims the schedule slot exactly like a raced-in reply or ticket.
-     */
-    void stopOrRestart(String lead, Set<String> repliesBefore, Set<String> ticketsBefore,
-                       Set<String> questionsBefore) {
         activeLeads.remove(lead);
         boolean racedIn = pendingReplyTargetsFor(lead).stream().anyMatch(t -> !repliesBefore.contains(t))
-                || pendingTicketIdsFor(lead).stream().anyMatch(t -> !ticketsBefore.contains(t))
-                || pendingQuestionTurnIdsFor(lead).stream().anyMatch(t -> !questionsBefore.contains(t));
+                || pendingTicketIdsFor(lead).stream().anyMatch(t -> !ticketsBefore.contains(t));
         if (racedIn && activeLeads.putIfAbsent(lead, Boolean.TRUE) == null) {
             log.debug("push: new work for lead {} raced the reminder loop's stop — restarting", lead);
             scheduleNext(lead);
@@ -494,44 +387,36 @@ public final class ReplyPushLoop {
     }
 
     /** Send one combined nudge covering everything currently pending for {@code lead}. */
-    private void injectNudge(String lead, int replyReminderCount, int ticketReminderCount,
-                             int questionReminderCount) {
-        // Re-read rather than threading it down from decide(): a reply can drain, a ticket be
-        // collected, or a question be answered (or another arrive), between the decision and the
-        // injection.
+    private void injectNudge(String lead, int replyReminderCount, int ticketReminderCount) {
+        // Re-read rather than threading it down from decide(): a reply can drain, or a ticket be
+        // collected (or another arrive), between the decision and the injection.
         Set<String> replyTargets = pendingReplyTargetsFor(lead);
         List<PendingTicket> tickets = pendingTicketsFor(lead);
-        List<PendingQuestion> questions = pendingQuestionsFor(lead);
-        if (replyTargets.isEmpty() && tickets.isEmpty() && questions.isEmpty()) {
+        if (replyTargets.isEmpty() && tickets.isEmpty()) {
             log.debug("push: pending work for lead {} drained before the nudge could be sent", lead);
             return;
         }
-        String nudge = formatNudge(replyTargets, tickets, questions);
+        String nudge = formatNudge(replyTargets, tickets);
         try {
             agents.send(lead, nudge);
-            log.debug("push: nudge sent to lead {} (reply {}/{}, ticket {}/{}, question {}/{}; "
-                            + "{} reply target(s), {} ticket(s), {} question(s))",
+            log.debug("push: nudge sent to lead {} (reply {}/{}, ticket {}/{}; {} reply target(s), {} ticket(s))",
                     lead, replyReminderCount + 1, maxReminders, ticketReminderCount + 1, maxReminders,
-                    questionReminderCount + 1, maxReminders,
-                    replyTargets.size(), tickets.size(), questions.size());
+                    replyTargets.size(), tickets.size());
             countNudge("delivered");
         } catch (RuntimeException e) {
-            log.warn("push: failed to nudge lead {} (reply {}/{}, ticket {}/{}, question {}/{}): {}",
-                    lead, replyReminderCount + 1, maxReminders, ticketReminderCount + 1, maxReminders,
-                    questionReminderCount + 1, maxReminders, e.toString());
+            log.warn("push: failed to nudge lead {} (reply {}/{}, ticket {}/{}): {}",
+                    lead, replyReminderCount + 1, maxReminders, ticketReminderCount + 1, maxReminders, e.toString());
         }
         // Bump every item actually named in this nudge, not just whatever the shared source-level
         // eligibility used to gate (CB-598) — each item's own count is what the next tick's
-        // minReplyNudgeCountFor / minTicketNudgeCountFor / minQuestionNudgeCountFor will read. An
-        // item already at or over the cap keeps riding along in the text (still pending, still
-        // named) but its extra bumps here are inert: decide() already treats it as ineligible once
-        // its count reaches maxReminders.
-        bumpNudgeCounts(replyTargets, tickets, questions);
+        // minReplyNudgeCountFor / minTicketNudgeCountFor will read. An item already at or over the
+        // cap keeps riding along in the text (still pending, still named) but its extra bumps here
+        // are inert: decide() already treats it as ineligible once its count reaches maxReminders.
+        bumpNudgeCounts(replyTargets, tickets);
     }
 
     /** Record that every one of these items was just named in a sent (or attempted) nudge. */
-    private void bumpNudgeCounts(Set<String> replyTargets, List<PendingTicket> tickets,
-                                 List<PendingQuestion> questions) {
+    private void bumpNudgeCounts(Set<String> replyTargets, List<PendingTicket> tickets) {
         for (String target : replyTargets) {
             pendingReplies.computeIfPresent(target, (t, e) -> new ReplyEntry(e.lead(), e.nudgeCount() + 1));
         }
@@ -539,11 +424,6 @@ public final class ReplyPushLoop {
             pendingTickets.computeIfPresent(ticket.ticket(),
                     (id, e) -> new PendingTicket(e.ticket(), e.lead(), e.failed(), e.nudgeCount() + 1));
         }
-        for (PendingQuestion question : questions) {
-            pendingQuestions.computeIfPresent(question.turnId(), (id, e) ->
-                    new PendingQuestion(e.turnId(), e.ticket(), e.target(), e.lead(), e.question(),
-                            e.nudgeCount() + 1));
-        }
     }
 
     /** Schedule the next tick on the scheduler thread pool. */
@@ -555,8 +435,7 @@ public final class ReplyPushLoop {
     // --- nudge formatting ------------------------------------------------------------------------
 
     /** Render everything pending for one lead as a single nudge line. */
-    private static String formatNudge(Set<String> replyTargets, List<PendingTicket> tickets,
-                                      List<PendingQuestion> questions) {
+    private static String formatNudge(Set<String> replyTargets, List<PendingTicket> tickets) {
         List<String> parts = new ArrayList<>();
         if (!replyTargets.isEmpty()) {
             parts.add(formatRepliesNudge(replyTargets));
@@ -564,9 +443,6 @@ public final class ReplyPushLoop {
         if (!tickets.isEmpty()) {
             parts.add(formatTicketsNudge(tickets));
         }
-        if (!questions.isEmpty()) {
-            parts.add(formatQuestionsNudge(questions));
-        }
         return String.join(" | ", parts);
     }
 
@@ -594,18 +470,6 @@ public final class ReplyPushLoop {
         return TICKETS_NUDGE_FORMAT.formatted(pending.size(), failedNote, ids);
     }
 
-    /** Render one or several open questions (CB-582). */
-    private static String formatQuestionsNudge(List<PendingQuestion> pending) {
-        if (pending.size() == 1) {
-            PendingQuestion q = pending.get(0);
-            return QUESTION_NUDGE_FORMAT.formatted(q.target(), q.ticket(), q.turnId(), q.question());
-        }
-        String ids = pending.stream()
-                .map(q -> q.ticket() + " (turnId=" + q.turnId() + ")")
-                .collect(Collectors.joining(", "));
-        return QUESTIONS_NUDGE_FORMAT.formatted(pending.size(), ids);
-    }
-
     // --- lifecycle -----------------------------------------------------------------------------
 
     /**
@@ -613,8 +477,8 @@ public final class ReplyPushLoop {
      * uses this to stand aside: while the push loop is actively nudging a lead, a concurrent
      * heartbeat injection would start a second competing turn in the same pane — racing loops
      * multiply turns and context burn. "Active" means a schedule exists in {@link #activeLeads},
-     * which now covers reply-queued (CB-307), ticket-terminal (CB-588), and question-open (CB-582)
-     * work (CB-590) — bounded by what has been triggered, not by any persistent state.
+     * which now covers both reply-queued (CB-307) and ticket-terminal (CB-588) work (CB-590) —
+     * bounded by what has been triggered, not by any persistent state.
      */
     public boolean isActive() {
         return !activeLeads.isEmpty();
@@ -626,7 +490,6 @@ public final class ReplyPushLoop {
         activeLeads.clear();
         pendingReplies.clear();
         pendingTickets.clear();
-        pendingQuestions.clear();
     }
 
     /** @see #stop() */

bridged/src/main/java/dev/ltms/bridged/rest/BridgedApp.java

@@ -509,20 +509,10 @@ public final class BridgedApp {
             return;
         }
         try {
-            Map<String, Object> body = new LinkedHashMap<>();
-            body.put("sessionId", id);
-            body.put("status", messages.status(id).name().toLowerCase());
-            body.put("ready", presence.isPresent(id));
-            // CB-582: a worker paused mid-turn in an async bridge_ask is otherwise invisible to a
-            // status poll — surface the open question and how to answer it, same as bridge_poll's
-            // Phase.ASKING view.
-            MessageService.PendingAsk ask = messages.pendingAsk(id);
-            if (ask != null) {
-                body.put("question", ask.question());
-                body.put("turnId", ask.turnId());
-                body.put("ticket", ask.ticket());
-            }
-            ctx.status(200).json(body);
+            ctx.status(200).json(Map.of(
+                    "sessionId", id,
+                    "status", messages.status(id).name().toLowerCase(),
+                    "ready", presence.isPresent(id)));
         } catch (HerdrException e) {
             herdrError(ctx, e);
         }
@@ -548,11 +538,6 @@ public final class BridgedApp {
         if (v.detail() != null) {
             body.put("detail", v.detail());
         }
-        // CB-582: Phase.ASKING carries the question in v.reply() (handled above) and its answer-
-        // correlation id here — a REST caller polling this ticket otherwise has no way to answer it.
-        if (v.turnId() != null) {
-            body.put("turnId", v.turnId());
-        }
         ctx.status(200).json(body);
     }
 

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

@@ -293,8 +293,10 @@ public final class SessionManager implements TurnListener {
                 // blocked caller fails fast with a real reason instead of sitting on a rendezvous
                 // nothing will ever resolve. CB-578 stage C: carry the worktree/branch/snapshot ref
                 // too, so a failed ticket's detail can point a lead at the same tree to re-dispatch.
+                // CB-584 (issue #65 criterion 5): carry agentSessionId alongside them, so a lead can
+                // also resume the member's conversation, not just re-dispatch onto its files.
                 notifyReleased(new ReleaseDetail(removed.terminalId(), removed.worktree(),
-                        removed.branch(), snapshotRef));
+                        removed.branch(), snapshotRef, removed.agentSessionId()));
             }
         }
         // CB-581: the pane must always stop, even if the dirty check above threw. A session removed
@@ -348,7 +350,8 @@ public final class SessionManager implements TurnListener {
      * are {@code null} for a shared-tree session; {@code snapshotRef} is {@code null} unless this
      * release snapshotted a dirty worktree into {@code refs/wip/<branch>}.
      */
-    public record ReleaseDetail(String terminalId, String worktreePath, String branch, String snapshotRef) {
+    public record ReleaseDetail(String terminalId, String worktreePath, String branch, String snapshotRef,
+                                String agentSessionId) {
     }
 
     /**

bridged/src/test/java/dev/ltms/bridged/config/BridgedConfigTest.java

@@ -1040,6 +1040,44 @@ class BridgedConfigTest {
                 "an opencode worker with no argv defaults to the opencode binary, never claude");
     }
 
+    /**
+     * CB-604: an unrecognized {@code kind:} used to silently fall into the claude-code bucket — not
+     * matching {@code "opencode"} was the only check. With {@code argv:} also unset that meant the
+     * daemon tried to launch a program literally named after the typo.
+     */
+    @Test
+    void unknownKindIsRefusedAtLoadNamingTheValueAndTheAcceptedSet(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("kind-typo.yaml");
+        Files.writeString(f, """
+                profiles:
+                  gemini:
+                    kind: opencod
+                    model: google/gemini-2.5-pro
+                """);
+
+        IllegalStateException e = assertThrows(IllegalStateException.class, () -> BridgedConfig.load(f));
+        assertTrue(e.getMessage().contains("gemini"), "error names the profile: " + e.getMessage());
+        assertTrue(e.getMessage().contains("opencod"), "error names the bad value: " + e.getMessage());
+        assertTrue(e.getMessage().contains("claude-code") && e.getMessage().contains("opencode"),
+                "error names the accepted set: " + e.getMessage());
+    }
+
+    @Test
+    void blankKindStillDefaultsToClaudeCode(@TempDir Path dir) throws Exception {
+        Path f = dir.resolve("kind-blank.yaml");
+        Files.writeString(f, """
+                profiles:
+                  gx10:
+                    baseUrl: http://gx10.gw:8000
+                    kind: ""
+                    argv: ["ccs", "gx10"]
+                """);
+
+        BridgedConfig cfg = BridgedConfig.load(f);
+        assertEquals(BridgedConfig.Profile.KIND_CLAUDE_CODE, cfg.profiles().get("gx10").kind(),
+                "a blank kind: is documented to behave exactly like an absent one");
+    }
+
     @Test
     void authDefaultsToLoopbackTrustSoExistingConfigsBehaveAsBefore(@TempDir Path dir) throws Exception {
         Path f = dir.resolve("no-auth-block.yaml");

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

@@ -773,52 +773,6 @@ class BridgeMcpTest {
         assertEquals("blocked", textOf(res));
     }
 
-    /**
-     * CB-582: a lead polling {@code bridge_status} on its normal cadence — not {@code bridge_poll}
-     * — must also see a worker's open async {@code bridge_ask} question, since the reverse-rendezvous
-     * window it opened with is far shorter than that cadence.
-     */
-    @Test
-    void statusReportsAnOpenQuestionWhenTheWorkerIsMidAsk() throws Exception {
-        String ticket = messages.sendAsync(T, "task that asks");
-        long deadline = System.currentTimeMillis() + 3000;
-        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
-            Thread.sleep(5);
-        }
-        assertTrue(rendezvous.isWaiting(T), "sendAsync should have opened its rendezvous waiter");
-
-        CompletableFuture<MessageService.AskResult> ask =
-                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
-
-        MessageService.TaskView asking;
-        deadline = System.currentTimeMillis() + 3000;
-        do {
-            asking = messages.poll(ticket);
-            Thread.sleep(5);
-        } while (asking.phase() != MessageService.Phase.ASKING && System.currentTimeMillis() < deadline);
-        assertEquals(MessageService.Phase.ASKING, asking.phase());
-
-        McpSchema.CallToolResult res = BridgeMcp.status(messages, T);
-        assertNotEquals(Boolean.TRUE, res.isError());
-        String out = textOf(res);
-        assertTrue(out.startsWith("idle"), "the live status must still lead the text: " + out);
-        assertTrue(out.contains("which config file?"), "the question text must be shown: " + out);
-        assertTrue(out.contains("turnId=\"" + asking.turnId() + "\""), "the turnId must be shown: " + out);
-        assertTrue(out.contains("(ticket " + ticket + ")"), "the ticket must be shown: " + out);
-
-        // Clean up the still-open ask so the background thread does not linger past the test.
-        String turnId = asking.turnId();
-        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(turnId, "config.yaml", 5000));
-        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-        deadline = System.currentTimeMillis() + 3000;
-        while (!rendezvous.isWaiting(T) && System.currentTimeMillis() < deadline) {
-            Thread.sleep(5);
-        }
-        assertTrue(rendezvous.resolve(T, "done"));
-        answer.get(5, TimeUnit.SECONDS);
-    }
-
     // --- bridge_whoami: the caller's own identity, so an agent never has to guess its role -------
 
     @Test

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

@@ -714,47 +714,6 @@ class MessageServiceTest {
         assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
     }
 
-    // --- CB-582: bridge_status pendingAsk() ------------------------------------------------------
-
-    @Test
-    void pendingAskReturnsNullWhenNoQuestionIsOpen() throws Exception {
-        assertNull(messages.pendingAsk(T), "no async ticket at all -> no pending ask");
-
-        String ticket = messages.sendAsync(T, "long task");
-        awaitWaiting();
-        assertNull(messages.pendingAsk(T), "a plain pending delegation is not a question");
-
-        injectDelivery();
-        assertTrue(rendezvous.resolve(T, "done"));
-        awaitTicketPhase(ticket, MessageService.Phase.DONE);
-        assertNull(messages.pendingAsk(T), "a finished ticket carries no open question either");
-    }
-
-    @Test
-    void pendingAskReturnsTheOpenQuestionForAnAsyncTicket() throws Exception {
-        String ticket = messages.sendAsync(T, "task that asks");
-        awaitWaiting();
-        injectDelivery();
-
-        CompletableFuture<MessageService.AskResult> ask =
-                CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
-        MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
-
-        MessageService.PendingAsk pending = messages.pendingAsk(T);
-        assertNotNull(pending, "bridge_status should see the open question");
-        assertEquals(ticket, pending.ticket());
-        assertEquals("which config file?", pending.question());
-        assertEquals(asking.turnId(), pending.turnId());
-
-        CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                () -> messages.answer(asking.turnId(), "config.yaml", 5000));
-        assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-        assertNull(messages.pendingAsk(T), "an answered question is no longer pending");
-        awaitWaiting();
-        assertTrue(rendezvous.resolve(T, "done"));
-        assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
-    }
-
     // --- CB-588: async ticket terminal nudges ---------------------------------------------------
     //
     // MessageService.reply's rendezvous fast path is exactly what an async ticket always takes
@@ -874,71 +833,6 @@ class MessageServiceTest {
         }
     }
 
-    // --- CB-582: bridge_ask question-open nudges --------------------------------------------------
-
-    @Test
-    void anAsyncTicketThatPausesOnAQuestionNudgesTheLeadWithNoPriorPollCall() throws Exception {
-        try (var wiring = wireWithPushLoop(5, 50)) {
-            String ticket = wiring.service().sendAsync(T, "task that asks");
-            awaitWaiting();
-            injectDelivery();
-
-            CompletableFuture<MessageService.AskResult> ask = CompletableFuture.supplyAsync(
-                    () -> wiring.service().ask(T, "which config file?", 5000));
-            MessageService.TaskView asking = awaitTicketPhaseOn(wiring.service(), ticket, MessageService.Phase.ASKING);
-
-            awaitNudge(wiring.leadHerdr());
-            String nudge = wiring.leadHerdr().lastCall("agent.prompt").params().toString();
-            assertTrue(nudge.contains(ticket), "the nudge should name the ticket: " + nudge);
-            assertTrue(nudge.contains(asking.turnId()), "the nudge should name the turnId: " + nudge);
-            assertTrue(nudge.contains("bridge_send(turnId="),
-                    "the nudge should name the exact answer call: " + nudge);
-            assertTrue(nudge.contains("which config file?"), "the nudge should include the question: " + nudge);
-
-            // Clean up the still-open ask so the background thread does not linger past the test.
-            CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                    () -> wiring.service().answer(asking.turnId(), "config.yaml", 5000));
-            assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-            awaitWaiting();
-            assertTrue(rendezvous.resolve(T, "done"));
-            answer.get(5, TimeUnit.SECONDS);
-        }
-    }
-
-    @Test
-    void answeringAQuestionStopsFurtherNudgesAboutIt() throws Exception {
-        try (var wiring = wireWithPushLoop(5, 50)) {
-            String ticket = wiring.service().sendAsync(T, "task that asks");
-            awaitWaiting();
-            injectDelivery();
-
-            CompletableFuture<MessageService.AskResult> ask = CompletableFuture.supplyAsync(
-                    () -> wiring.service().ask(T, "which config file?", 5000));
-            MessageService.TaskView asking = awaitTicketPhaseOn(wiring.service(), ticket, MessageService.Phase.ASKING);
-
-            awaitNudge(wiring.leadHerdr());
-            long callsBeforeAnswer = wiring.leadHerdr().calls.stream()
-                    .filter(c -> c.method().equals("agent.prompt")).count();
-
-            CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
-                    () -> wiring.service().answer(asking.turnId(), "config.yaml", 5000));
-            assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
-            awaitWaiting();
-            assertTrue(rendezvous.resolve(T, "done"));
-            answer.get(5, TimeUnit.SECONDS);
-
-            // Let several more ticks (and the ticket's own now-legitimate terminal nudge) fire —
-            // none of them may still name the question's turnId, which is closed.
-            Thread.sleep(300);
-            boolean anyNamesClosedQuestion = wiring.leadHerdr().calls.stream()
-                    .filter(c -> c.method().equals("agent.prompt"))
-                    .skip(callsBeforeAnswer)
-                    .anyMatch(c -> c.params().toString().contains(asking.turnId()));
-            assertFalse(anyNamesClosedQuestion,
-                    "no nudge sent after the answer may still name the now-closed turnId " + asking.turnId());
-        }
-    }
-
     @Test
     void aFleetWithNoPushLoopConfiguredBehavesExactlyAsToday() throws Exception {
         // `messages` (the shared field) uses the no-pushLoop constructor — poll() must not throw,

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

@@ -657,132 +657,6 @@ class ReplyPushLoopTest {
         assertTrue(loop.isActive(), "the schedule must stay active after nudging the fresh ticket");
     }
 
-    // --- CB-582: bridge_ask question-open nudges -------------------------------------------------
-
-    @Test
-    void onQuestionOpenedWithNoKnownLeadNeverStartsASchedule() throws Exception {
-        var rec = recordingClient();
-        agents = new AgentControl(rec);
-        var loop = new ReplyPushLoop(new PrimaryRegistry(null), agents, inbox, scheduler, 5, 50);
-
-        loop.onQuestionOpened("task-1", WORKER, "term_worker#1", "which config?");
-
-        assertFalse(loop.isActive(), "no lead known means nothing to nudge yet");
-        Thread.sleep(150);
-        assertEquals(0, rec.sendCount(), "must not nudge when no lead is known to be waiting");
-    }
-
-    @Test
-    void decideQuestionsWithNothingPendingIsStop() {
-        agents = agentWithStatus("idle");
-        assertEquals(ReplyPushLoop.Action.STOP, loop().decide(PRIMARY, 0, 0, 0));
-    }
-
-    @Test
-    void decideQuestionsAtCapIsStop() {
-        agents = agentWithStatus("idle");
-        var loop = loop(2, 100_000);
-        loop.onQuestionOpened("task-1", WORKER, "term_worker#1", "which config?");
-        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0, 0, 2));
-    }
-
-    @Test
-    void decideQuestionsUnderCapWithInjectableLeadIsInject() {
-        agents = agentWithStatus("idle");
-        var loop = loop(5, 100_000);
-        loop.onQuestionOpened("task-1", WORKER, "term_worker#1", "which config?");
-        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0, 0, 0));
-    }
-
-    @Test
-    void onQuestionOpenedCausesExactlyOneNudgeNamingTheTicketAndTurnId() throws Exception {
-        var rec = recordingClient();
-        agents = new AgentControl(rec);
-
-        loop(1, 50).onQuestionOpened("task-1", WORKER, "term_worker#1", "which config file?");
-
-        assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS), "one question nudge should have been sent");
-        assertEquals(1, rec.sendCount());
-        String nudge = rec.sentParams().getFirst().getValue().toString();
-        assertTrue(nudge.contains("task-1"), "nudge should name the ticket: " + nudge);
-        assertTrue(nudge.contains("term_worker#1"), "nudge should name the turnId: " + nudge);
-        assertTrue(nudge.contains("bridge_send(turnId="), "nudge should name the exact answer call: " + nudge);
-        assertTrue(nudge.contains("which config file?"), "nudge should include the question text: " + nudge);
-    }
-
-    @Test
-    void questionClosedPreventsFurtherNudging() throws Exception {
-        var rec = recordingClient();
-        agents = new AgentControl(rec);
-        var loop = loop(1, 100);
-
-        loop.onQuestionOpened("task-1", WORKER, "term_worker#1", "which config?");
-        loop.questionClosed("term_worker#1"); // answered/lapsed before the first tick fired
-
-        Thread.sleep(300); // let the scheduled tick run
-        assertEquals(0, rec.sendCount(), "an already-closed question must never be nudged");
-    }
-
-    @Test
-    void questionNudgesSendUpToCapThenStop() throws Exception {
-        int cap = 2;
-        var rec = recordingClient();
-        agents = new AgentControl(rec);
-        rec.sendLatch = new CountDownLatch(cap);
-
-        loop(cap, 50).onQuestionOpened("task-1", WORKER, "term_worker#1", "which config?");
-
-        assertTrue(rec.sendLatch.await(5, TimeUnit.SECONDS), cap + " question nudges should have fired");
-        Thread.sleep(300);
-        assertEquals(cap, rec.sendCount(), "exactly " + cap + " question nudges (cap=" + cap + ")");
-    }
-
-    @Test
-    void questionAndTicketForTheSameLeadCoalesceIntoOneSend() throws Exception {
-        var rec = recordingClient();
-        agents = new AgentControl(rec);
-        var loop = loop(1, 300); // backoff wide enough that both entry points land before the first tick
-
-        loop.onTicketTerminal("task-1", WORKER, false);
-        loop.onQuestionOpened("task-2", WORKER, "term_worker#1", "which config?");
-
-        assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS), "one combined nudge should have been sent");
-        Thread.sleep(300);
-        assertEquals(1, rec.sendCount(),
-                "a ticket and a question for the same lead must coalesce onto ONE schedule");
-        String nudge = rec.sentParams().getFirst().getValue().toString();
-        assertTrue(nudge.contains("task-1"), "the combined nudge must still mention the ticket: " + nudge);
-        assertTrue(nudge.contains("term_worker#1"), "the combined nudge must still mention the question: " + nudge);
-    }
-
-    @Test
-    void oneExhaustedQuestionSourceDoesNotBlockANudgeForTheOtherSources() {
-        // Mirrors oneExhaustedSourceDoesNotBlockANudgeForTheOtherSource for the question source:
-        // the question source is at its cap (2/2), but the ticket source has never been nudged
-        // (0/2) — decide() must still INJECT so the ticket is not stranded.
-        agents = agentWithStatus("idle");
-        var loop = loop(2, 100_000);
-        loop.onQuestionOpened("task-1", WORKER, "term_worker#1", "which config?");
-        loop.onTicketTerminal("task-2", WORKER, false);
-
-        assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0, 0, 2),
-                "the question source is exhausted (2/2), but the ticket source has never been "
-                        + "nudged (0/2) — the lead must still be injected so the ticket is not lost");
-    }
-
-    @Test
-    void questionNudgeFormatIsCorrect() {
-        String single = ReplyPushLoop.QUESTION_NUDGE_FORMAT.formatted(
-                WORKER, "task-1", "term_worker#1", "which config?");
-        assertTrue(single.contains("Worker term_worker"));
-        assertTrue(single.contains("bridge_send(turnId=\"term_worker#1\""));
-        assertTrue(single.contains("which config?"));
-
-        String multi = ReplyPushLoop.QUESTIONS_NUDGE_FORMAT.formatted(2, "task-1 (turnId=t1), task-2 (turnId=t2)");
-        assertTrue(multi.contains("2 workers"));
-        assertTrue(multi.contains("bridge_poll(ticket=...)"));
-    }
-
     // --- metrics (CB-512) ----------------------------------------------------------------------
 
     @Test

bridged/src/test/java/dev/ltms/bridged/rest/BridgedAppTest.java

@@ -484,67 +484,6 @@ class BridgedAppTest {
         assertTrue(mapper.readTree(req(port, "GET", "/sessions/term_a/status").body()).get("ready").asBoolean());
     }
 
-    /**
-     * CB-582: a lead polling {@code GET /sessions/{id}/status} on its normal cadence — not the
-     * ticket-scoped {@code /tasks/{ticket}} — must also see a worker's open async {@code bridge_ask}
-     * question, since the reverse-rendezvous window it opened with is far shorter than that cadence.
-     */
-    @Test
-    void sessionStatusReportsAnOpenQuestionWhenTheWorkerIsMidAsk() throws Exception {
-        FakeHerdr herdr = new FakeHerdr().agentStatus("idle"); // poller delivers the injection
-        int port = start(herdr, "http://gx00.gw:8000", Set.of("gx00.gw"));
-
-        HttpResponse<String> accepted = postMessage(port, "{\"content\":\"do it\",\"wait\":false}");
-        assertEquals(202, accepted.statusCode());
-        String ticket = mapper.readTree(accepted.body()).get("ticket").asText();
-
-        Thread.sleep(200); // let the background async send open its rendezvous waiter
-
-        var ask = java.util.concurrent.CompletableFuture.supplyAsync(() -> {
-            try {
-                return postJson(port, "/sessions/term_a/ask",
-                        "{\"question\":\"which config file?\",\"timeoutMs\":5000}");
-            } catch (Exception e) {
-                throw new RuntimeException(e);
-            }
-        });
-
-        JsonNode task;
-        long deadline = System.currentTimeMillis() + 3000;
-        do {
-            task = mapper.readTree(req(port, "GET", "/tasks/" + ticket).body());
-            if ("asking".equals(task.path("phase").asText())) break;
-            //noinspection BusyWait
-            Thread.sleep(10);
-        } while (System.currentTimeMillis() < deadline);
-        assertEquals("asking", task.get("phase").asText());
-        String turnId = task.get("turnId").asText();
-
-        HttpResponse<String> status = req(port, "GET", "/sessions/term_a/status");
-        assertEquals(200, status.statusCode());
-        JsonNode body = mapper.readTree(status.body());
-        assertEquals("idle", body.get("status").asText(), "the live status must still be reported");
-        assertEquals("which config file?", body.get("question").asText());
-        assertEquals(turnId, body.get("turnId").asText());
-        assertEquals(ticket, body.get("ticket").asText());
-
-        // Answer it — via the same /message route bridge_send uses, keyed by turnId — so the
-        // background ask thread does not linger past the test.
-        var answer = java.util.concurrent.CompletableFuture.supplyAsync(() -> {
-            try {
-                return postJson(port, "/sessions/term_a/message",
-                        "{\"content\":\"config.yaml\",\"turnId\":\"" + turnId + "\",\"timeoutMs\":4000}");
-            } catch (Exception e) {
-                throw new RuntimeException(e);
-            }
-        });
-        HttpResponse<String> askResponse = ask.get(6, java.util.concurrent.TimeUnit.SECONDS);
-        assertEquals(200, askResponse.statusCode());
-        assertEquals("config.yaml", mapper.readTree(askResponse.body()).get("answer").asText());
-        postJson(port, "/sessions/term_a/reply", "{\"content\":\"done\"}");
-        answer.get(6, java.util.concurrent.TimeUnit.SECONDS);
-    }
-
     @Test
     void stopWorkerInPanePlacementClosesOnlyThePane() throws Exception {
         FakeHerdr herdr = new FakeHerdr();

bridged/src/test/java/dev/ltms/bridged/session/WorktreeSessionManagerTest.java

@@ -422,6 +422,28 @@ class WorktreeSessionManagerTest {
                 "acceptance criterion 6: a failed ticket's detail must carry the snapshot ref");
     }
 
+    @Test
+    void releaseNotifiesTheListenerWithTheAgentSessionId() {
+        // CB-584 (issue #65 criterion 5): a failed ticket's detail must also name the agent
+        // session, alongside worktree/branch/snapshot, so a lead can resume the conversation
+        // rather than only re-dispatch a fresh member onto the same files.
+        FakeHerdr herdr = new FakeHerdr();
+        FakeWorktrees worktrees = new FakeWorktrees().withRepoRoot("/repo").withPrefix("/wt")
+                .withDirty(true);
+        SessionManager sessions = new SessionManager(workerService(herdr), worktrees);
+        java.util.List<SessionManager.ReleaseDetail> released = new java.util.concurrent.CopyOnWriteArrayList<>();
+        sessions.onRelease(released::add);
+        MemberSession s = sessions.acquire("ltms-local", MemberRole.DEV, null, "/caller/proj", null,
+                new WorktreeRequest("cb-584-e", null), "cb-584-session", null);
+
+        assertNotNull(s.agentSessionId(), "a named session must mint an agent session id to assert on");
+        sessions.release(s.paneId());
+
+        assertEquals(1, released.size());
+        SessionManager.ReleaseDetail detail = released.getFirst();
+        assertEquals(s.agentSessionId(), detail.agentSessionId());
+    }
+
     @Test
     void aFailingSnapshotStillPreservesTheWorktreeStopsThePaneAndNotifies() {
         FakeHerdr herdr = new FakeHerdr();