fleetd #201/#227 unit 4: durable BACKEND_ERROR member outcome

Adds MemberSession.State.BACKEND_ERROR with a nullable failureReason, surfaced in
rosterView, and SessionManager.onBackendError(target, reason). The CAS loop accepts
both sides of the completion race (BUSY and DONE); BACKEND_ERROR is terminal.

completeTurn now returns early when its CAS loses, so a stale DONE copy can no longer
release the pane or reset its context behind a member that just went BACKEND_ERROR.

Verified by the lead: 1130 tests green; mutating the completeTurn early return back to
the old fall-through turns losingCompletionDoesNotReleaseOrClearABackendErrorMember red
with 0 compile errors.

fleetd/src/main/java/dev/ltms/fleet/session/MemberSession.java

@@ -46,7 +46,8 @@ public record MemberSession(
         String worktree,
         String branch,
         CharterReceipt charterReceipt,
-        String agentSessionId) {
+        String agentSessionId,
+        String failureReason) {
 
     /** One-shot worker lifecycle states. */
     public enum State {
@@ -54,6 +55,7 @@ public record MemberSession(
         READY,
         BUSY,
         DONE,
+        BACKEND_ERROR,
         FAILED,
         RELEASED
     }
@@ -69,25 +71,35 @@ public record MemberSession(
                          long lastActivityAtNanos, int turnCount, State state,
                          String worktree, String branch) {
         this(paneId, terminalId, profile, role, cwd, ownerTerminal, spawnedAtNanos,
-                lastActivityAtNanos, turnCount, state, worktree, branch, null, null);
+                lastActivityAtNanos, turnCount, state, worktree, branch, null, null, null);
+    }
+
+    /** Backward-compatible shape without a backend failure reason. */
+    public MemberSession(String paneId, String terminalId, String profile, MemberRole role,
+                         String cwd, String ownerTerminal, long spawnedAtNanos,
+                         long lastActivityAtNanos, int turnCount, State state,
+                         String worktree, String branch, CharterReceipt charterReceipt,
+                         String agentSessionId) {
+        this(paneId, terminalId, profile, role, cwd, ownerTerminal, spawnedAtNanos,
+                lastActivityAtNanos, turnCount, state, worktree, branch, charterReceipt, agentSessionId, null);
     }
 
     /** Return a copy of this session in {@code state}. */
     public MemberSession withState(State state) {
         return new MemberSession(paneId, terminalId, profile, role, cwd, ownerTerminal, spawnedAtNanos,
-                lastActivityAtNanos, turnCount, state, worktree, branch, charterReceipt, agentSessionId);
+                lastActivityAtNanos, turnCount, state, worktree, branch, charterReceipt, agentSessionId, failureReason);
     }
 
     /** Return a copy with {@code lastActivityAtNanos} updated to {@code nowNanos}. */
     public MemberSession withActivity(long nowNanos) {
         return new MemberSession(paneId, terminalId, profile, role, cwd, ownerTerminal, spawnedAtNanos,
-                nowNanos, turnCount, state, worktree, branch, charterReceipt, agentSessionId);
+                nowNanos, turnCount, state, worktree, branch, charterReceipt, agentSessionId, failureReason);
     }
 
     /** Return a copy with the turn count incremented and activity timestamped at {@code nowNanos}. */
     public MemberSession bumpTurn(long nowNanos) {
         return new MemberSession(paneId, terminalId, profile, role, cwd, ownerTerminal, spawnedAtNanos,
-                nowNanos, turnCount + 1, state, worktree, branch, charterReceipt, agentSessionId);
+                nowNanos, turnCount + 1, state, worktree, branch, charterReceipt, agentSessionId, failureReason);
     }
 
     /**
@@ -98,6 +110,12 @@ public record MemberSession(
      */
     public MemberSession withAgentSessionId(String agentSessionId) {
         return new MemberSession(paneId, terminalId, profile, role, cwd, ownerTerminal, spawnedAtNanos,
-                lastActivityAtNanos, turnCount, state, worktree, branch, charterReceipt, agentSessionId);
+                lastActivityAtNanos, turnCount, state, worktree, branch, charterReceipt, agentSessionId, failureReason);
+    }
+
+    /** Return a copy with the durable backend failure detail. */
+    public MemberSession withFailureReason(String failureReason) {
+        return new MemberSession(paneId, terminalId, profile, role, cwd, ownerTerminal, spawnedAtNanos,
+                lastActivityAtNanos, turnCount, state, worktree, branch, charterReceipt, agentSessionId, failureReason);
     }
 }

fleetd/src/main/java/dev/ltms/fleet/session/SessionManager.java

@@ -658,6 +658,9 @@ public final class SessionManager implements TurnListener {
         m.put("profile", session.profile());
         m.put("role", session.role() == null ? "dev" : session.role().wireName());
         m.put("state", session.state().name().toLowerCase());
+        if (session.failureReason() != null) {
+            m.put("failureReason", session.failureReason());
+        }
         if (session.worktree() != null) {
             m.put("worktree", session.worktree());
         }
@@ -718,6 +721,36 @@ public final class SessionManager implements TurnListener {
         completeTurn(target, false);
     }
 
+    /**
+     * Record that a member backend failed after a delegated ticket expired. This CAS loop accepts
+     * either side of the completion race: {@code BUSY -> BACKEND_ERROR} or
+     * {@code DONE -> BACKEND_ERROR}. A released or unknown session is never recreated.
+     *
+     * @return {@code true} when the error is recorded on a live session
+     */
+    public boolean onBackendError(String target, String reason) {
+        while (true) {
+            MemberSession current = findByTerminal(target);
+            if (current == null) {
+                log.warn("backend error for unknown member terminal={}", target);
+                return false;
+            }
+            if (current.state() == MemberSession.State.RELEASED) {
+                return false;
+            }
+            if (current.state() == MemberSession.State.BACKEND_ERROR) {
+                return true;
+            }
+            MemberSession updated = current.withState(MemberSession.State.BACKEND_ERROR)
+                    .withFailureReason(reason).withActivity(nowNanos.getAsLong());
+            if (replace(current, updated)) {
+                log.warn("member terminal={} pane={} transitioned {} -> BACKEND_ERROR: {}",
+                        target, current.paneId(), current.state(), reason);
+                return true;
+            }
+        }
+    }
+
     @Override
     public boolean hasPostTurnAction(String target) {
         if (!clearAfterTurn) return false;
@@ -736,10 +769,9 @@ public final class SessionManager implements TurnListener {
         if (current == null || current.state() != MemberSession.State.BUSY) return false;
         long now = nowNanos.getAsLong();
         MemberSession updated = current.withState(MemberSession.State.DONE).withActivity(now);
-        if (replace(current, updated)) {
-            log.debug("session transitioned terminal={} pane={} BUSY -> DONE turn={}",
-                    target, current.paneId(), updated.turnCount());
-        }
+        if (!replace(current, updated)) return false;
+        log.debug("session transitioned terminal={} pane={} BUSY -> DONE turn={}",
+                target, current.paneId(), updated.turnCount());
         if (contextCap > 0 && updated.turnCount() >= contextCap) {
             release(current.paneId());
             return false;

fleetd/src/test/java/dev/ltms/fleet/session/SessionManagerTest.java

@@ -300,6 +300,151 @@ class SessionManagerTest {
         assertTrue(sessions.roster().contains(updated), "FAILED is still in acquired-minus-released roster");
     }
 
+    @Test
+    void backendErrorWinsBeforeNormalCompletionFromBusy() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.asPresence().markPresent(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
+
+        assertTrue(sessions.onBackendError(session.terminalId(), "backend exited"));
+        sessions.onTurnComplete(session.terminalId());
+
+        MemberSession updated = sessions.get(session.paneId()).orElseThrow();
+        assertEquals(MemberSession.State.BACKEND_ERROR, updated.state(),
+                "normal completion must not restore DONE after a backend error from BUSY");
+        assertEquals("backend exited", updated.failureReason());
+    }
+
+    @Test
+    void backendErrorWinsAfterNormalCompletionFromDone() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.asPresence().markPresent(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
+        sessions.onTurnComplete(session.terminalId());
+
+        assertTrue(sessions.onBackendError(session.terminalId(), "backend exited"));
+
+        MemberSession updated = sessions.get(session.paneId()).orElseThrow();
+        assertEquals(MemberSession.State.BACKEND_ERROR, updated.state(),
+                "backend error must replace DONE when normal completion won first");
+        assertEquals("backend exited", updated.failureReason());
+    }
+
+    @Test
+    void backendErrorMemberCannotReceiveAnotherDelivery() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.asPresence().markPresent(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
+        assertTrue(sessions.onBackendError(session.terminalId(), "backend exited"));
+
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
+
+        assertEquals(MemberSession.State.BACKEND_ERROR, sessions.get(session.paneId()).orElseThrow().state(),
+                "onDelivered must refuse a backend-error member");
+    }
+
+    @Test
+    void losingCompletionDoesNotReleaseOrClearABackendErrorMember() {
+        assertLosingCompletionLeavesBackendErrorIntact(1, false,
+                "a stale completion must not release a backend-error member at the context cap");
+        assertLosingCompletionLeavesBackendErrorIntact(0, true,
+                "a stale completion must not clear the context of a backend-error member");
+    }
+
+    private void assertLosingCompletionLeavesBackendErrorIntact(int contextCap, boolean clearAfterTurn,
+                                                                  String protectedAction) {
+        FakeHerdr herdr = new FakeHerdr();
+        FleetConfig.Profile cfg = new FleetConfig.Profile(
+                "ltms-local", "http://gx00.gw:8000", "coder", null, "FLEETD_WORKER_TOKEN",
+                List.of("ccs", "ltms-local"), "tab", "fleetd-workers",
+                "worker: {profile} #{n}", null, null, null);
+        ClaudeCodeLauncher delegate = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
+                new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
+        ClearContextSpyLauncher launcher = new ClearContextSpyLauncher(delegate);
+        java.util.concurrent.atomic.AtomicReference<SessionManager> manager = new java.util.concurrent.atomic.AtomicReference<>();
+        java.util.concurrent.atomic.AtomicReference<String> terminal = new java.util.concurrent.atomic.AtomicReference<>();
+        java.util.concurrent.atomic.AtomicBoolean armed = new java.util.concurrent.atomic.AtomicBoolean();
+        SessionManager sessions = new SessionManager(launcher, new RecordingWorktrees(), () -> {
+            if (armed.compareAndSet(true, false)) {
+                manager.get().onBackendError(terminal.get(), "backend exited during completion");
+            }
+            return 1;
+        }, contextCap, clearAfterTurn);
+        manager.set(sessions);
+        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        terminal.set(session.terminalId());
+        sessions.asPresence().markPresent(session.terminalId());
+        sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
+        armed.set(true);
+
+        assertFalse(sessions.onTurnCompleteWithPostAction(session.terminalId()),
+                "the completion CAS loses after the injected backend error");
+
+        assertTrue(sessions.get(session.paneId()).isPresent(), protectedAction);
+        MemberSession updated = sessions.get(session.paneId()).orElseThrow();
+        assertEquals(MemberSession.State.BACKEND_ERROR, updated.state());
+        assertEquals("backend exited during completion", updated.failureReason());
+        assertFalse(herdr.called("pane.close"), protectedAction);
+        assertEquals(0, launcher.clearContextCalls(), protectedAction);
+    }
+
+    @Test
+    void backendErrorForUnknownTargetIsWarnedAndDoesNotCreateASession() {
+        LoggerContext ctx = (LoggerContext) LoggerFactory.getILoggerFactory();
+        ch.qos.logback.classic.Logger sessionLog =
+                (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(SessionManager.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>();
+        appender.setContext(ctx);
+        appender.start();
+        sessionLog.addAppender(appender);
+        try {
+            SessionManager sessions = sessionManager(new FakeHerdr());
+
+            assertFalse(sessions.onBackendError("term_missing", "backend exited"));
+
+            assertTrue(sessions.roster().isEmpty(), "unknown target must not create a session");
+            assertTrue(appender.list.stream().anyMatch(e -> e.getLevel().equals(Level.WARN)
+                            && e.getFormattedMessage().contains("term_missing")),
+                    "unknown target is logged at WARN");
+        } finally {
+            sessionLog.detachAppender(appender);
+        }
+    }
+
+    @Test
+    void backendErrorForReleasedTargetDoesNotRecreateTheSession() {
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager sessions = sessionManager(herdr);
+        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.release(session.paneId());
+
+        assertFalse(sessions.onBackendError(session.terminalId(), "backend exited"));
+        assertTrue(sessions.get(session.paneId()).isEmpty(), "released session must stay absent");
+    }
+
+    @Test
+    void rosterViewIncludesFailureReasonOnlyForBackendError() {
+        MemberSession failed = new MemberSession("pane-error", "term-error", "prof", MemberRole.DEV,
+                "/cwd", null, 0, 0, 1, MemberSession.State.BACKEND_ERROR, null, null,
+                null, null, "backend exited");
+        MemberSession ordinary = new MemberSession("pane-ready", "term-ready", "prof", MemberRole.DEV,
+                "/cwd", null, 0, 0, 0, MemberSession.State.READY, null, null);
+
+        Map<String, Object> failedView = SessionManager.rosterView(failed, null);
+        Map<String, Object> ordinaryView = SessionManager.rosterView(ordinary, null);
+
+        assertEquals("backend_error", failedView.get("state"));
+        assertEquals("backend exited", failedView.get("failureReason"));
+        assertFalse(ordinaryView.containsKey("failureReason"),
+                "ordinary rows must not gain a blank failureReason field");
+    }
+
     @Test
     void onTurnFailedIsLoggedAtWarnWithThePriorState() {
         // CB-564: this transition used to be a bare DEBUG "session marked failed" — a symptom with no
@@ -1004,6 +1149,76 @@ class SessionManagerTest {
         }
     }
 
+    /** Delegates all launcher work while recording context clears. */
+    private static final class ClearContextSpyLauncher implements PeerLauncher {
+        private final PeerLauncher delegate;
+        private int clearContextCalls;
+
+        ClearContextSpyLauncher(PeerLauncher delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public Set<Capability> capabilities() {
+            return delegate.capabilities();
+        }
+
+        @Override
+        public Set<Capability> capabilitiesFor(String profileName) {
+            return delegate.capabilitiesFor(profileName);
+        }
+
+        @Override
+        public PeerHandle spawn(SpawnRequest req) {
+            return delegate.spawn(req);
+        }
+
+        @Override
+        public Set<String> profiles() {
+            return delegate.profiles();
+        }
+
+        @Override
+        public String defaultProfile() {
+            return delegate.defaultProfile();
+        }
+
+        @Override
+        public String effectiveCwd(SpawnRequest req) {
+            return delegate.effectiveCwd(req);
+        }
+
+        @Override
+        public List<String> parityOverlay(String profileName) {
+            return delegate.parityOverlay(profileName);
+        }
+
+        @Override
+        public List<?> list() {
+            return delegate.list();
+        }
+
+        @Override
+        public int reapOrphanWorkers() {
+            return delegate.reapOrphanWorkers();
+        }
+
+        @Override
+        public void stop(String id) {
+            delegate.stop(id);
+        }
+
+        @Override
+        public boolean clearContext(String id) {
+            clearContextCalls++;
+            return delegate.clearContext(id);
+        }
+
+        int clearContextCalls() {
+            return clearContextCalls;
+        }
+    }
+
     @Test
     void acquireRecordsTheAgentSessionIdFromTheHandle() {
         FakeHerdr herdr = new FakeHerdr();