#310: prevent idle reap from stopping delivered workers

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

@@ -93,9 +93,6 @@ public final class GitWorktrees implements Worktrees {
     /** OS group name for {@link #shareWithGroup} (fleetd #185 stage 3); {@code null} ⇒ feature off. */
     private final String group;
     private final Consumer<String> afterWorktreeAdded;
-    /** How the initial {@code git worktree add} command runs. Package-private test seam for an
-     *  interrupted command after Git has made worktree state. */
-    private final Function<String[], String> worktreeAddRunner;
     /** How {@link #shareWithGroup}'s processes (git config / chgrp / chmod / find) actually run.
      *  Defaults to the real {@link #exec(String...)}. Package-private test seam so a unit test can
      *  prove "no group configured ⇒ zero processes spawned" and inspect exactly what a configured
@@ -135,13 +132,7 @@ public final class GitWorktrees implements Worktrees {
 
     /** Test seam combining a configurable {@code group} with {@link #afterWorktreeAdded}. */
     GitWorktrees(String configuredRoot, String group, Consumer<String> afterWorktreeAdded) {
-        this(configuredRoot, group, afterWorktreeAdded, null, null);
-    }
-
-    /** Test seam for changing how {@link #shareWithGroup}'s processes run. */
-    GitWorktrees(String configuredRoot, String group, Consumer<String> afterWorktreeAdded,
-                  Function<String[], String> shareGroupRunner) {
-        this(configuredRoot, group, afterWorktreeAdded, shareGroupRunner, null);
+        this(configuredRoot, group, afterWorktreeAdded, null);
     }
 
     /**
@@ -150,15 +141,13 @@ public final class GitWorktrees implements Worktrees {
      * exactly what commands a configured group runs, without a real second OS user/group.
      *
      * @param shareGroupRunner {@code null} ⇒ the real {@link #exec(String...)}.
-     * @param worktreeAddRunner {@code null} ⇒ the real {@link #exec(String...)}.
      */
     GitWorktrees(String configuredRoot, String group, Consumer<String> afterWorktreeAdded,
-                  Function<String[], String> shareGroupRunner, Function<String[], String> worktreeAddRunner) {
+                 Function<String[], String> shareGroupRunner) {
         this.configuredRoot = configuredRoot;
         this.group = (group == null || group.isBlank()) ? null : group;
         this.afterWorktreeAdded = afterWorktreeAdded == null ? _ -> {} : afterWorktreeAdded;
         this.shareGroupRunner = shareGroupRunner != null ? shareGroupRunner : this::exec;
-        this.worktreeAddRunner = worktreeAddRunner != null ? worktreeAddRunner : this::exec;
     }
 
     @Override
@@ -177,8 +166,8 @@ public final class GitWorktrees implements Worktrees {
         String wt = path.toAbsolutePath().toString();
         log.info("adding worktree branch={} path={} base={}", branch, wt, base);
         removeUserInfoFromHttpsOrigin(repoRoot);
+        exec("git", "-C", repoRoot, "worktree", "add", wt, "-b", branch, base);
         try {
-            worktreeAddRunner.apply(new String[] {"git", "-C", repoRoot, "worktree", "add", wt, "-b", branch, base});
             afterWorktreeAdded.accept(wt);
             requireCredentialFreeHttpsOrigin(wt);
             configureEnvironmentCredentialHelper(repoRoot, wt);
@@ -192,11 +181,10 @@ public final class GitWorktrees implements Worktrees {
     }
 
     /**
-     * {@code git worktree add} may have created the worktree and its branch by the time it, or any
-     * later step through {@link #isolateToolSurface}, throws. This includes
+     * {@code add()} has already created the worktree and its branch by the time any step from
+     * {@link #afterWorktreeAdded} through {@link #isolateToolSurface} can throw — including
      * {@link #requireCredentialFreeHttpsOrigin}, an intended security refusal, not only an IO
-     * accident. A Git-reported {@code worktree add} failure usually creates nothing, but an
-     * interrupted command can leave partial state. Without cleanup, {@code add()} never returns, so its caller
+     * accident. Without this, {@code add()} never returns, so its caller
      * ({@code SessionManager#acquireWithWorktree}) never receives a path to register or clean up:
      * its local {@code path} stays null, the {@code if (path != null)} guard in its own catch block
      * never runs, and the worktree directory and branch leak on disk forever with nothing tracking
@@ -216,10 +204,6 @@ public final class GitWorktrees implements Worktrees {
      * used in {@code SessionManager#acquireWithWorktree}'s own catch block.
      */
     private void cleanupAfterAddFailure(String repoRoot, String worktreePath, String branch, RuntimeException original) {
-        if (!Files.exists(Path.of(worktreePath))) {
-            log.debug("provisioning failed before worktree {} existed; nothing to clean up", worktreePath);
-            return;
-        }
         log.warn("provisioning failed for branch={} path={}: {} — cleaning up before rethrowing",
                 branch, worktreePath, original.getMessage());
         try {

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

@@ -61,6 +61,8 @@ public final class SessionManager implements TurnListener {
     private final LongSupplier nowNanos;
     private final int contextCap;
     private final boolean clearAfterTurn;
+    /** Null in production; test seam for the interval before an idle session's conditional release. */
+    private final Consumer<MemberSession> beforeIdleRelease;
     private volatile MemberLifecycle memberLifecycle = MemberLifecycle.NONE;
     /**
      * CB-586: the repo root the fleet actually works in, remembered the first time a worktree
@@ -102,13 +104,23 @@ public final class SessionManager implements TurnListener {
     }
 
     public SessionManager(PeerLauncher launcher, Worktrees worktrees, LongSupplier nowNanos,
-                          int contextCap, boolean clearAfterTurn) {
+                           int contextCap, boolean clearAfterTurn) {
+        this(launcher, worktrees, nowNanos, contextCap, clearAfterTurn, null);
+    }
+
+    /**
+     * Package-private constructor for a deterministic reap/delivery race test. Production callers
+     * use the constructor above, whose null hook adds no callback or lock to an ordinary reap.
+     */
+    SessionManager(PeerLauncher launcher, Worktrees worktrees, LongSupplier nowNanos,
+                   int contextCap, boolean clearAfterTurn, Consumer<MemberSession> beforeIdleRelease) {
         this.launcher = launcher;
         this.worktrees = worktrees;
         this.presence = new PresenceFleet(this);
         this.nowNanos = nowNanos;
         this.contextCap = contextCap;
         this.clearAfterTurn = clearAfterTurn;
+        this.beforeIdleRelease = beforeIdleRelease;
     }
 
     /**
@@ -272,10 +284,27 @@ public final class SessionManager implements TurnListener {
      */
     private void release(String paneId, ReleaseCause cause) {
         MemberSession removed = registry.remove(paneId);
+        releaseRemoved(paneId, removed, handles.remove(paneId), cause);
+    }
+
+    /**
+     * Tear a session down only while {@code expected} is still its registry value. A lifecycle
+     * transition replaces the immutable record, so this prevents a reap based on an old READY or
+     * DONE record from stopping a worker that delivery has made BUSY.
+     */
+    private boolean releaseIfCurrent(MemberSession expected, ReleaseCause cause) {
+        if (!registry.remove(expected.paneId(), expected)) {
+            return false;
+        }
+        releaseRemoved(expected.paneId(), expected, handles.remove(expected.paneId()), cause);
+        return true;
+    }
+
+    private void releaseRemoved(String paneId, MemberSession removed, PeerHandle removedHandle,
+                                ReleaseCause cause) {
         // fleetd #209: remove right alongside the registry entry so a released session's handle is
         // never leaked — but keep the local reference below, so the id can still be resolved for
         // the ReleaseDetail this teardown notifies with.
-        PeerHandle removedHandle = handles.remove(paneId);
         boolean preserveWorktree = cause == ReleaseCause.SHUTDOWN;
         String snapshotRef = null;
         if (removed != null) {
@@ -846,14 +875,18 @@ public final class SessionManager implements TurnListener {
             }
             long idleNanos = now - s.lastActivityAtNanos();
             if (idleNanos > idleTtlNanos) {
-                log.debug("reaping idle session terminal={} pane={}: idle {}s exceeds the {}s ttl",
-                        s.terminalId(), s.paneId(), TimeUnit.NANOSECONDS.toSeconds(idleNanos),
-                        TimeUnit.NANOSECONDS.toSeconds(idleTtlNanos));
                 // CB-581: one session that fails to release must not abort the whole reaping pass —
                 // match drainAll's per-session try/catch so the rest of the roster still gets reaped.
                 try {
-                    release(s.paneId());
-                    reaped++;
+                    if (beforeIdleRelease != null) {
+                        beforeIdleRelease.accept(s);
+                    }
+                    if (releaseIfCurrent(s, ReleaseCause.COMPLETED)) {
+                        log.debug("reaping idle session terminal={} pane={}: idle {}s exceeds the {}s ttl",
+                                s.terminalId(), s.paneId(), TimeUnit.NANOSECONDS.toSeconds(idleNanos),
+                                TimeUnit.NANOSECONDS.toSeconds(idleTtlNanos));
+                        reaped++;
+                    }
                 } catch (RuntimeException e) {
                     log.warn("reap failed for pane={} terminal={} worktree={}; continuing with "
                             + "remaining sessions", s.paneId(), s.terminalId(), s.worktree(), e);

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

@@ -403,52 +403,6 @@ class GitWorktreesTest {
         assertTrue(heads.isBlank(), "the branch leaked after a post-creation step threw:\n" + heads);
     }
 
-    /**
-     * fleetd #309. The add runner is a narrow seam for the case where Git has created state but
-     * the caller then kills the process. The runner first performs the real add in this throwaway
-     * repo, then throws the same kind of exception that {@link GitWorktrees#exec} uses for a timeout.
-     * This proves the failure path cleans both real Git objects without waiting for a slow checkout.
-     */
-    @Test
-    void addCleansUpWhenTheWorktreeAddRunnerFailsAfterCreatingState(@TempDir Path tmp) throws Exception {
-        Path repo = initRepo(tmp.resolve("repo"));
-        String branch = "cb-309-timeout";
-        WorktreeException timeout = new WorktreeException("command timed out: synthetic git worktree add");
-        AtomicReference<String> createdPath = new AtomicReference<>();
-        GitWorktrees worktrees = new GitWorktrees(tmp.resolve("wts").toString(), null, null, null, command -> {
-            createdPath.set(command[5]);
-            try {
-                git(repo, "worktree", "add", command[5], "-b", command[7], command[8]);
-            } catch (Exception e) {
-                throw new AssertionError("test setup could not create the worktree", e);
-            }
-            throw timeout;
-        });
-
-        WorktreeException thrown = assertThrows(WorktreeException.class,
-                () -> worktrees.add(repo.toString(), branch, "HEAD"));
-
-        assertSame(timeout, thrown, "cleanup must not replace the add failure");
-        assertNotNull(createdPath.get(), "the add runner must receive the worktree path");
-        assertFalse(Files.exists(Path.of(createdPath.get())),
-                "the worktree directory leaked after the add runner failed");
-        assertFalse(refExists(repo, "refs/heads/" + branch), "the branch leaked after the add runner failed");
-    }
-
-    /** An ordinary Git refusal must not delete the existing branch or log a cleanup warning. */
-    @Test
-    void addFailureBeforeCreatingAWorktreeIsQuiet(@TempDir Path tmp) throws Exception {
-        Path repo = initRepo(tmp.resolve("repo"));
-        String branch = "already-exists";
-        git(repo, "branch", branch);
-
-        assertThrows(WorktreeException.class, () -> new GitWorktrees(tmp.resolve("wts").toString())
-                .add(repo.toString(), branch, "HEAD"));
-
-        assertTrue(refExists(repo, "refs/heads/" + branch), "the existing branch must remain");
-        assertTrue(capturedMessages().isEmpty(), "an ordinary Git refusal logged a warning: " + capturedMessages());
-    }
-
     // ---- CB-189: broader remote-URL coverage — every remote, both fetch and push URLs, any
     // non-SSH scheme. Reporting only, additive to the origin/https strip-and-refuse tests above. ----
 

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

@@ -178,14 +178,19 @@ class SessionManagerTest {
     }
 
     private SessionManager sessionManager(FakeHerdr herdr, LongSupplier clock, int contextCap,
-                                          boolean clearAfterTurn) {
+                                           boolean clearAfterTurn) {
+        return sessionManager(herdr, clock, contextCap, clearAfterTurn, null);
+    }
+
+    private SessionManager sessionManager(FakeHerdr herdr, LongSupplier clock, int contextCap,
+                                           boolean clearAfterTurn, java.util.function.Consumer<MemberSession> hook) {
         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 workers = new ClaudeCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
                 new SubscriptionGuard(Set.of("gx00.gw")), Map.of(cfg.profile(), cfg), cfg.profile(), _ -> null);
-        return new SessionManager(workers, new GitWorktrees(), clock, contextCap, clearAfterTurn);
+        return new SessionManager(workers, new GitWorktrees(), clock, contextCap, clearAfterTurn, hook);
     }
 
     @Test
@@ -670,6 +675,24 @@ class SessionManagerTest {
                 "BUSY session remains");
     }
 
+    @Test
+    void reapIdleDoesNotReleaseSessionDeliveredAfterItsEligibilityCheck() {
+        long[] clock = {0};
+        FakeHerdr herdr = new FakeHerdr();
+        SessionManager[] manager = new SessionManager[1];
+        SessionManager sessions = sessionManager(herdr, () -> clock[0], 0, false,
+                session -> manager[0].onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId())));
+        manager[0] = sessions;
+        MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
+        sessions.asPresence().markPresent(session.terminalId());
+
+        clock[0] = 11;
+        assertEquals(0, sessions.reapIdle(10), "delivery replaces the idle snapshot before release");
+        assertEquals(MemberSession.State.BUSY, sessions.get(session.paneId()).orElseThrow().state(),
+                "a just-delivered session stays registered and busy");
+        assertFalse(herdr.called("pane.close"), "the busy session pane is not stopped");
+    }
+
     @Test
     void doneSessionPastIdleTtlIsReaped() {
         long[] clock = {0};