diff --git a/bridged/src/main/java/dev/ltms/bridged/Bridged.java b/bridged/src/main/java/dev/ltms/bridged/Bridged.java
index 6364365..54094fb 100644
--- a/bridged/src/main/java/dev/ltms/bridged/Bridged.java
+++ b/bridged/src/main/java/dev/ltms/bridged/Bridged.java
@@ -209,6 +209,12 @@ public final class Bridged {
MessageService messages = new MessageService(agents, injector, rendezvous, replyInbox,
pushLoop, metrics);
+ // CB-516: releasing a worker must fail whatever send was waiting on it. Without this a
+ // torn-down delegation kept reporting PENDING until the 30-minute async timeout, and never
+ // reached /metrics — the delegation was unresolvable and nothing said so.
+ sessions.onRelease(terminal ->
+ messages.abandon(terminal, "the worker session was released before it replied"));
+
// MCP server face (CB-105): bridge_send/bridge_reply/bridge_status, mounted at /mcp.
// Caller identity is resolved from the connection (peer PID → herdr pane), not arguments.
ConnectionIdentity identity = new ConnectionIdentity(
diff --git a/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java b/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java
index 4cf9e79..befa426 100644
--- a/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java
+++ b/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java
@@ -255,6 +255,33 @@ public final class MessageService {
};
}
+ /**
+ * Abandon any send still waiting on {@code target} because its session has gone away (CB-516).
+ *
+ * Without this, tearing a worker down left its rendezvous waiter open: a blocking
+ * {@code bridge_send} kept blocking, and an async one kept reporting {@code PENDING} until
+ * {@link #ASYNC_TIMEOUT_MS} — thirty minutes — even though the worker provably no longer
+ * existed and the delegation could never complete. Worse, {@code poll} already had the evidence
+ * (it calls {@code liveStatus} to build its detail string and gets back {@code "unknown"}) and
+ * reported {@code PENDING} anyway.
+ *
+ *
Resolving the waiter as a failure — rather than letting it time out — also means the
+ * outcome is counted, so a torn-down delegation stops being invisible to {@code /metrics}.
+ *
+ * @return true if a live waiter was failed
+ */
+ public boolean abandon(String target, String reason) {
+ CompletableFuture waiter = rendezvous.currentWaiter(target);
+ if (waiter == null || waiter.isDone()) {
+ return false; // nobody is blocked on this worker — nothing to abandon
+ }
+ boolean failed = rendezvous.resolveFailure(waiter, reason);
+ if (failed) {
+ log.debug("abandoned send to {}: {}", target, reason);
+ }
+ return failed;
+ }
+
/**
* Acknowledge a specific reply by {@code msgId} for {@code target}. Removes it from the inbox
* so that a subsequent drain or peek no longer returns it.
diff --git a/bridged/src/main/java/dev/ltms/bridged/session/SessionManager.java b/bridged/src/main/java/dev/ltms/bridged/session/SessionManager.java
index 1ad2d2c..d60c24b 100644
--- a/bridged/src/main/java/dev/ltms/bridged/session/SessionManager.java
+++ b/bridged/src/main/java/dev/ltms/bridged/session/SessionManager.java
@@ -17,6 +17,7 @@ import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
import java.util.function.LongSupplier;
/**
@@ -47,6 +48,9 @@ public final class SessionManager implements TurnListener {
private final LongSupplier nowNanos;
private final int contextCap;
+ /** CB-516: notified with a terminalId on every release; no-op until wired. */
+ private volatile Consumer releaseListener = _ -> { };
+
/** Backward-compatible constructor: shared-tree sessions, production git seam. */
public SessionManager(PeerLauncher launcher) {
this(launcher, new GitWorktrees(), System::nanoTime, 0);
@@ -136,6 +140,10 @@ public final class SessionManager implements TurnListener {
if (removed != null) {
log.debug("releasing session pane={} terminal={} state={}",
removed.paneId(), removed.terminalId(), removed.state());
+ // CB-516: a send still waiting on this worker can never be answered now. Tell the
+ // listener BEFORE the pane is torn down, so a blocked caller fails fast with a real
+ // reason instead of sitting on a rendezvous nothing will ever resolve.
+ notifyReleased(removed.terminalId());
}
launcher.stop(paneId);
if (removed != null && removed.worktree() != null) {
@@ -143,6 +151,32 @@ public final class SessionManager implements TurnListener {
}
}
+ /**
+ * Register a callback invoked with a session's {@code terminalId} whenever it is released
+ * (CB-516). Every teardown path funnels through {@link #release}, so one hook covers the REST
+ * and MCP stop tools, the idle-TTL reaper, {@code recycle}, and shutdown drain alike.
+ *
+ * Set rather than injected because {@code MessageService} — the intended listener — is
+ * constructed after this manager (it needs the injector and rendezvous, which need the session
+ * presence view this manager exposes). Wiring it at construction would require breaking that
+ * cycle for one callback.
+ */
+ public void onRelease(Consumer listener) {
+ this.releaseListener = (listener == null) ? _ -> { } : listener;
+ }
+
+ /** A listener failure must never prevent the teardown it is reacting to. */
+ private void notifyReleased(String terminalId) {
+ if (terminalId == null) {
+ return;
+ }
+ try {
+ releaseListener.accept(terminalId);
+ } catch (RuntimeException e) {
+ log.warn("release listener failed for terminal {}: {}", terminalId, e.toString());
+ }
+ }
+
private WorkerSession acquireWithWorktree(String profile, String requestedCwd, String callerCwd,
String ownerTerminal, WorktreeRequest wt) {
String resolvedProfile = (profile == null || profile.isBlank())
diff --git a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java b/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java
index 2f4bb75..92f1b0c 100644
--- a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java
+++ b/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java
@@ -401,4 +401,67 @@ class MessageServiceTest {
injector.onStatus(T, AgentStatus.IDLE); // deliver the task
injector.onStatus(T, AgentStatus.WORKING); // worker picks it up
}
+
+ // --- CB-516: a released session must not leave a send hanging ------------------------------
+
+ /**
+ * The bug this fixes: tearing a worker down left its rendezvous waiter open, so a blocking send
+ * kept blocking and an async one kept reporting PENDING until the 30-minute async timeout —
+ * even though the worker provably no longer existed.
+ */
+ @Test
+ void abandonFailsASendThatIsStillWaitingOnAReleasedSession() throws Exception {
+ CompletableFuture send =
+ CompletableFuture.supplyAsync(() -> messages.send(T, "work", 30_000));
+ awaitWaiting();
+
+ assertTrue(messages.abandon(T, "session released"), "a live waiter is abandoned");
+
+ MessageService.Reply r = send.get(5, TimeUnit.SECONDS);
+ assertEquals(MessageService.Outcome.WORKER_FAILED, r.outcome(),
+ "an abandoned send fails rather than riding out its timeout");
+ assertEquals("session released", r.text(), "the caller is told why");
+ }
+
+ @Test
+ void abandonIsANoOpWhenNobodyIsWaiting() {
+ assertFalse(messages.abandon(T, "session released"),
+ "no open send ⇒ nothing to abandon");
+ }
+
+ @Test
+ void abandonDoesNotOverwriteAnAlreadyResolvedSend() throws Exception {
+ CompletableFuture send =
+ CompletableFuture.supplyAsync(() -> messages.send(T, "work", 30_000));
+ awaitWaiting();
+ assertTrue(rendezvous.resolve(T, "the real answer"));
+
+ assertFalse(messages.abandon(T, "session released"),
+ "a send already answered by the worker must not be clobbered");
+ MessageService.Reply r = send.get(5, TimeUnit.SECONDS);
+ assertEquals("the real answer", r.text());
+ }
+
+ /** The async path is the one that hung: poll must report FAILED, not PENDING forever. */
+ @Test
+ void anAbandonedAsyncTaskPollsAsFailedNotPending() throws Exception {
+ String ticket = messages.sendAsync(T, "long task");
+ awaitWaiting();
+ assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase());
+
+ messages.abandon(T, "session released");
+
+ MessageService.TaskView view = null;
+ long deadline = System.currentTimeMillis() + 3000;
+ while (System.currentTimeMillis() < deadline) {
+ view = messages.poll(ticket);
+ if (view.phase() != MessageService.Phase.PENDING) break;
+ Thread.sleep(10);
+ }
+ assertNotNull(view);
+ assertEquals(MessageService.Phase.FAILED, view.phase(),
+ "a delegation whose worker is gone must not keep reporting PENDING");
+ assertTrue(view.detail() != null && view.detail().contains("released"),
+ "and the detail says why, rather than 'worker unknown'");
+ }
}
diff --git a/bridged/src/test/java/dev/ltms/bridged/session/SessionManagerTest.java b/bridged/src/test/java/dev/ltms/bridged/session/SessionManagerTest.java
index fcece67..3338e7a 100644
--- a/bridged/src/test/java/dev/ltms/bridged/session/SessionManagerTest.java
+++ b/bridged/src/test/java/dev/ltms/bridged/session/SessionManagerTest.java
@@ -362,4 +362,46 @@ class SessionManagerTest {
assertTrue(sessions.roster().isEmpty(),
"no session is registered when spawn times out (roster empty)");
}
+
+ // --- CB-516: release must notify, so a blocked send can be failed --------------------------
+
+ @Test
+ void releaseNotifiesTheListenerWithTheReleasedTerminal() {
+ FakeHerdr herdr = new FakeHerdr();
+ SessionManager sessions = sessionManager(herdr);
+ java.util.List released = new java.util.concurrent.CopyOnWriteArrayList<>();
+ sessions.onRelease(released::add);
+
+ WorkerSession s = sessions.acquire("ltms-local", null, "/caller", null);
+ sessions.release(s.paneId());
+
+ assertEquals(java.util.List.of(s.terminalId()), released,
+ "every teardown path funnels through release, so one hook must see the terminal");
+ }
+
+ @Test
+ void releasingAnUnknownPaneNotifiesNobody() {
+ FakeHerdr herdr = new FakeHerdr();
+ SessionManager sessions = sessionManager(herdr);
+ java.util.List released = new java.util.concurrent.CopyOnWriteArrayList<>();
+ sessions.onRelease(released::add);
+
+ sessions.release("w9:p404"); // idempotent teardown of something already gone
+
+ assertTrue(released.isEmpty(), "no session removed ⇒ no send was waiting on it");
+ }
+
+ @Test
+ void aThrowingReleaseListenerDoesNotBlockTheTeardown() {
+ FakeHerdr herdr = new FakeHerdr();
+ SessionManager sessions = sessionManager(herdr);
+ sessions.onRelease(_ -> {
+ throw new IllegalStateException("listener blew up");
+ });
+
+ WorkerSession s = sessions.acquire("ltms-local", null, "/caller", null);
+ assertDoesNotThrow(() -> sessions.release(s.paneId()),
+ "a listener failure must never prevent the teardown it is reacting to");
+ assertTrue(sessions.get(s.paneId()).isEmpty(), "and the session is still deregistered");
+ }
}