fleet/fleetd
3
0
Fork 0
Code Issues 73 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Merge cb578c-92885c-1

Browse Source
This commit is contained in:
Dai Ha
2026-08-15 13:07:50 +02:00
parent c29c3f063d 6725642274
commit a3842c873d
8 changed files with 432 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bridged/src/main/java/dev/ltms/bridged/Bridged.java
+11 -4
View File
@@ -428,10 +428,17 @@ public final class Bridged {
// 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");
replyInbox.release(terminal);
primaryRegistry.forgetDelegation(terminal); // CB-532: don't leak the lead binding
sessions.onRelease(detail -> {
// CB-578 stage C, acceptance criterion 10: a failed ticket's detail should tell a lead
// where to re-dispatch onto the same tree, not just that the worker vanished.
String reason = "the worker session was released before it replied";
if (detail.worktreePath() != null) {
reason += "; worktree=" + detail.worktreePath() + " branch=" + detail.branch()
+ " snapshot=" + (detail.snapshotRef() != null ? detail.snapshotRef() : "none");
}
messages.abandon(detail.terminalId(), reason);
replyInbox.release(detail.terminalId());
primaryRegistry.forgetDelegation(detail.terminalId()); // CB-532: don't leak the lead binding
});
// MCP server face (CB-105): bridge_send/bridge_reply/bridge_status, mounted at /mcp.
bridged/src/main/java/dev/ltms/bridged/session/GitWorktrees.java
+59 -1
View File
@@ -13,6 +13,8 @@ import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.security.SecureRandom;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
@@ -217,6 +219,53 @@ public final class GitWorktrees implements Worktrees {
return Path.of(out.trim()).toAbsolutePath().normalize().toString();
}
/**
* CB-578 stage C. Stages into a <em>temporary</em> index (never the worktree's real one, which
* the worker may still be writing to), writes that index to a tree, commits the tree on top of
* the worktree's current HEAD, and points {@code refs/wip/<branch>} at the result:
*
* <pre>
* GIT_INDEX_FILE=&lt;temp&gt; git -C worktree add -A
* tree=$(GIT_INDEX_FILE=&lt;temp&gt; git -C worktree write-tree)
* commit=$(git -C worktree commit-tree $tree -p HEAD -m message)
* git -C worktree update-ref refs/wip/branch $commit
* </pre>
*
* {@code add -A} (never {@code -f}) respects {@code .gitignore} exactly as it would in the real
* index — a gitignored file staying ignored is what keeps secrets and local config out of the
* snapshot's tree. The temporary index file is removed afterwards regardless of outcome.
*/
@Override
public Optional<String> snapshot(String worktreePath, String branch, String message) {
if (!Files.exists(Path.of(worktreePath))) {
log.debug("worktree {} already gone — nothing to snapshot", worktreePath);
return Optional.empty();
}
Path tempIndex;
try {
tempIndex = Files.createTempFile("bridged-wip-index-", ".tmp");
Files.delete(tempIndex); // git creates it fresh under GIT_INDEX_FILE; a stale empty
// file at that path is otherwise treated as a corrupt index.
} catch (IOException e) {
throw new WorktreeException("cannot create a temporary index for snapshot: " + e.getMessage(), e);
}
Map<String, String> indexEnv = Map.of("GIT_INDEX_FILE", tempIndex.toAbsolutePath().toString());
try {
exec(indexEnv, "git", "-C", worktreePath, "add", "-A");
String tree = exec(indexEnv, "git", "-C", worktreePath, "write-tree").trim();
String commit = exec("git", "-C", worktreePath, "commit-tree", tree, "-p", "HEAD", "-m", message).trim();
exec("git", "-C", worktreePath, "update-ref", "refs/wip/" + branch, commit);
log.info("snapshotted worktree {} to refs/wip/{} commit={}", worktreePath, branch, commit);
return Optional.of(commit);
} finally {
try {
Files.deleteIfExists(tempIndex);
} catch (IOException e) {
log.debug("could not delete temporary snapshot index {}: {}", tempIndex, e.getMessage());
}
}
}
/** Resolve the directory that will hold per-session worktree checkouts. */
private Path resolveRoot(String repoRoot) {
if (configuredRoot != null && !configuredRoot.isBlank()) {
@@ -239,11 +288,20 @@ public final class GitWorktrees implements Worktrees {
* stdout and stderr (merged by redirectErrorStream).
*/
private String exec(String... command) {
return exec(Map.of(), command);
}
/** Same as {@link #exec(String...)}, with extra environment variables set on the child process. */
private String exec(Map<String, String> extraEnv, String... command) {
String out;
int code;
Process p;
try {
p = new ProcessBuilder(command).redirectErrorStream(true).start();
ProcessBuilder pb = new ProcessBuilder(command).redirectErrorStream(true);
if (extraEnv != null && !extraEnv.isEmpty()) {
pb.environment().putAll(extraEnv);
}
p = pb.start();
} catch (IOException e) {
throw new WorktreeException("failed to start " + command[0] + ": " + e.getMessage(), e);
}
bridged/src/main/java/dev/ltms/bridged/session/SessionManager.java
+67 -11
View File
@@ -54,8 +54,8 @@ public final class SessionManager implements TurnListener {
/** CB-520: notified with a terminalId on every acquire; no-op until wired. */
private final List<Consumer<String>> acquireListeners = new java.util.concurrent.CopyOnWriteArrayList<>();
/** CB-516: notified with a terminalId on every release; no-op until wired. */
private final List<Consumer<String>> releaseListeners = new java.util.concurrent.CopyOnWriteArrayList<>();
/** CB-516: notified with a {@link ReleaseDetail} on every release; no-op until wired. */
private final List<Consumer<ReleaseDetail>> releaseListeners = new java.util.concurrent.CopyOnWriteArrayList<>();
/** Backward-compatible constructor: shared-tree sessions, production git seam. */
public SessionManager(PeerLauncher launcher) {
@@ -196,14 +196,16 @@ public final class SessionManager implements TurnListener {
private void release(String paneId, ReleaseCause cause) {
MemberSession removed = registry.remove(paneId);
boolean preserveWorktree = cause == ReleaseCause.SHUTDOWN;
String snapshotRef = null;
if (removed != null) {
try {
memberLifecycle.released(removed.terminalId());
log.debug("releasing session pane={} terminal={} state={} cause={}",
removed.paneId(), removed.terminalId(), removed.state(), cause);
boolean dirty = removed.worktree() != null && worktrees.hasUncommitted(removed.worktree());
if (preserveWorktree && removed.worktree() != null) {
logPreservedForShutdown(removed);
} else if (removed.worktree() != null && worktrees.hasUncommitted(removed.worktree())) {
} else if (dirty) {
// CB-576: a release that would otherwise remove the worktree finds it holding
// uncommitted work the bridge cannot see. A worker that ends a turn without
// committing (normally because it stopped to ask a question or refused the turn)
@@ -214,6 +216,13 @@ public final class SessionManager implements TurnListener {
+ "the worktree holds uncommitted changes that --force remove would destroy",
cause, removed.worktree(), removed.paneId(), removed.terminalId());
}
if (dirty) {
// CB-578 stage C: preserving on disk is not saving — the directory is one
// `worktree remove --force`, or an operator tidying up, away from gone. Commit
// its full state to a ref before the preserve-or-remove decision above can be
// undone by anything else, regardless of why this release fired.
snapshotRef = trySnapshot(removed, cause);
}
} catch (RuntimeException e) {
// CB-581: hasUncommitted shells out to `git status` and can throw on a non-zero
// exit. We can no longer tell whether the worktree holds uncommitted work, so fail
@@ -228,8 +237,10 @@ public final class SessionManager implements TurnListener {
// CB-516/CB-581: a send still waiting on this worker can never be answered now, no
// matter what happened above. 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());
// 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.
notifyReleased(new ReleaseDetail(removed.terminalId(), removed.worktree(),
removed.branch(), snapshotRef));
}
}
// CB-581: the pane must always stop, even if the dirty check above threw. A session removed
@@ -241,6 +252,51 @@ public final class SessionManager implements TurnListener {
}
}
/**
* Best-effort snapshot of a dirty worktree into {@code refs/wip/<branch>} (CB-578 stage C). A
* failure here must never escalate: the caller has already decided to preserve the worktree
* regardless of whether this succeeds, so the only cost of a failed snapshot is a WARN and a
* missing ref — never a lost pane stop or a lost release notification.
*/
private String trySnapshot(MemberSession session, ReleaseCause cause) {
if (session.worktree() == null || session.branch() == null) {
return null;
}
try {
Optional<String> ref = worktrees.snapshot(session.worktree(), session.branch(),
snapshotMessage(session, cause));
ref.ifPresent(sha -> log.info(
"snapshotted dirty worktree {} to refs/wip/{} commit={} for pane={} terminal={}",
session.worktree(), session.branch(), sha, session.paneId(), session.terminalId()));
return ref.orElse(null);
} catch (RuntimeException e) {
log.warn("snapshot of dirty worktree {} failed for pane={} terminal={} branch={}: the "
+ "worktree is still preserved on disk, just not committed to refs/wip/{}: {}",
session.worktree(), session.paneId(), session.terminalId(), session.branch(),
session.branch(), e.toString());
return null;
}
}
/** Commit message for a CB-578 stage C snapshot — names the member so an operator can tell runs apart. */
private String snapshotMessage(MemberSession session, ReleaseCause cause) {
return "CB-578 stage C: snapshot of a released worker\n\n"
+ "terminal: " + session.terminalId() + "\n"
+ "profile: " + session.profile() + "\n"
+ "branch: " + session.branch() + "\n"
+ "cause: " + cause;
}
/**
* Facts about a released session that a listener needs beyond the bare terminal id — enough
* for a caller to point a lead at where to re-dispatch onto the same tree after a failed
* release (CB-578 stage C, acceptance criterion 10). {@code worktreePath} and {@code branch}
* 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) {
}
/**
* Why a session is being released — governs whether its worktree is preserved or removed.
* Worktree removal is reserved for the one case that is genuinely finished; everything else
@@ -289,7 +345,7 @@ public final class SessionManager implements TurnListener {
* presence view this manager exposes). Wiring it at construction would require breaking that
* cycle for one callback.
*/
public void onRelease(Consumer<String> listener) {
public void onRelease(Consumer<ReleaseDetail> listener) {
if (listener != null) {
releaseListeners.add(listener);
}
@@ -315,15 +371,15 @@ public final class SessionManager implements TurnListener {
}
/** A listener failure must never prevent the teardown it is reacting to. */
private void notifyReleased(String terminalId) {
if (terminalId == null) {
private void notifyReleased(ReleaseDetail detail) {
if (detail.terminalId() == null) {
return;
}
for (Consumer<String> listener : releaseListeners) {
for (Consumer<ReleaseDetail> listener : releaseListeners) {
try {
listener.accept(terminalId);
listener.accept(detail);
} catch (RuntimeException e) {
log.warn("release listener failed for terminal {}: {}", terminalId, e.toString());
log.warn("release listener failed for terminal {}: {}", detail.terminalId(), e.toString());
}
}
}
bridged/src/main/java/dev/ltms/bridged/session/Worktrees.java
+27
View File
@@ -1,6 +1,7 @@
package dev.ltms.bridged.session;
import java.util.List;
import java.util.Optional;
/** Seam between {@link SessionManager} and git worktree operations. Tests use a recording fake. */
public interface Worktrees {
@@ -27,4 +28,30 @@ public interface Worktrees {
/** git -C <cwd> rev-parse --show-toplevel — the repo root that owns cwd. */
String repoRoot(String cwd);
/**
* Commit the worktree's full on-disk state — tracked and untracked, respecting
* {@code .gitignore} — to {@code refs/wip/<branch>}, so a release that would otherwise leave
* the work as loose, unprotected files has a durable git object to fall back on (CB-578 stage
* C). Built on a <em>temporary</em> index: the worker's own index, working tree, and HEAD are
* never touched, since the worker may still be mid-write. The commit is parented on the
* worktree's current HEAD.
*
* <p>Never writes under {@code refs/heads/} — the ref must not appear in {@code git branch},
* must not be pushed by default, and must not be swept by a later {@code git branch -d}.
*
* <p>Callers are expected to have already confirmed {@link #hasUncommitted} before reaching
* for this; it always stages and commits whatever {@code git add -A} finds, so calling it on
* a clean worktree still produces a (harmless, tree-identical-to-HEAD) commit rather than
* detecting cleanliness itself.
*
* @param worktreePath absolute path of the worktree to snapshot
* @param branch the worktree's own branch — keys {@code refs/wip/<branch>}
* @param message the commit message; should name the member, its branch, and the release
* cause so an operator can tell which run produced it
* @return the created commit's sha, or {@link Optional#empty()} if {@code worktreePath} does
* not exist (mirrors {@link #remove} and {@link #hasUncommitted}'s already-gone
* tolerance — a worktree that is gone holds nothing to snapshot)
*/
Optional<String> snapshot(String worktreePath, String branch, String message);
}