Dai Ha
427 lines
23 KiB
Java
427 lines
23 KiB
Java
package dev.ltms.bridged;
|
|
|
|
import dev.ltms.bridged.config.BridgedConfig;
|
|
import dev.ltms.bridged.guard.SubscriptionGuard;
|
|
import dev.ltms.bridged.herdr.AgentControl;
|
|
import dev.ltms.bridged.herdr.HerdrClient;
|
|
import dev.ltms.bridged.herdr.HerdrException;
|
|
import dev.ltms.bridged.herdr.LeadTabScanner;
|
|
import dev.ltms.bridged.herdr.PaneLocator;
|
|
import dev.ltms.bridged.herdr.UnixSocketHerdrClient;
|
|
import dev.ltms.bridged.herdr.WorkspaceControl;
|
|
import dev.ltms.bridged.inject.CompletionResolver;
|
|
import dev.ltms.bridged.inject.Injector;
|
|
import dev.ltms.bridged.inject.StatusPoller;
|
|
import dev.ltms.bridged.inject.TurnListener;
|
|
import dev.ltms.bridged.inject.WorkerPresence;
|
|
import dev.ltms.bridged.auth.ArchitectRegistry;
|
|
import dev.ltms.bridged.auth.CallerResolver;
|
|
import dev.ltms.bridged.mcp.BridgeMcp;
|
|
import dev.ltms.bridged.mcp.ConnectionIdentity;
|
|
import dev.ltms.bridged.metrics.BridgedMetrics;
|
|
import dev.ltms.bridged.metrics.Metrics;
|
|
import dev.ltms.bridged.mcp.PrimaryRegistry;
|
|
import dev.ltms.bridged.mcp.LsofPeerPidLookup;
|
|
import dev.ltms.bridged.mcp.LsofProcessCwdLookup;
|
|
import dev.ltms.bridged.msg.AmqpReplyInbox;
|
|
import dev.ltms.bridged.msg.InMemoryReplyInbox;
|
|
import dev.ltms.bridged.msg.MessageService;
|
|
import dev.ltms.bridged.msg.Rendezvous;
|
|
import dev.ltms.bridged.msg.ReplyInbox;
|
|
import dev.ltms.bridged.msg.ReplyPushLoop;
|
|
import dev.ltms.bridged.rest.BridgedApp;
|
|
import dev.ltms.bridged.session.GitWorktrees;
|
|
import dev.ltms.bridged.session.SessionManager;
|
|
import dev.ltms.bridged.peer.PeerLauncher;
|
|
import dev.ltms.bridged.session.SessionReaper;
|
|
import dev.ltms.bridged.worker.ClaudeCodeLauncher;
|
|
import dev.ltms.bridged.placement.PlacementPolicies;
|
|
import dev.ltms.bridged.worker.CompositePeerLauncher;
|
|
import dev.ltms.bridged.worker.HerdrPeerLauncher;
|
|
import dev.ltms.bridged.worker.OpenCodeLauncher;
|
|
import io.javalin.Javalin;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
import java.nio.file.Path;
|
|
import java.util.ArrayList;
|
|
import java.util.LinkedHashMap;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.Objects;
|
|
import java.util.Set;
|
|
import java.util.concurrent.Executors;
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.concurrent.atomic.AtomicReference;
|
|
import java.util.function.Function;
|
|
import java.util.function.Predicate;
|
|
import java.util.function.Supplier;
|
|
import java.util.stream.Collectors;
|
|
|
|
/**
|
|
* {@code bridged} entry point. Wires the real herdr socket client to the REST app and
|
|
* starts listening. Before anything else it asserts its own environment is clean —
|
|
* {@code bridged} is not a Claude process and must never carry a base_url.
|
|
*/
|
|
public final class Bridged {
|
|
|
|
private static final Logger log = LoggerFactory.getLogger(Bridged.class);
|
|
|
|
/** How often the injector samples a busy worker's status while it has queued work. */
|
|
private static final long INJECT_POLL_MILLIS = 250;
|
|
|
|
/** CB-504: how long to wait at startup for herdr's socket before serving degraded. */
|
|
private static final long HERDR_WAIT_SECONDS = 30;
|
|
private static final long HERDR_WAIT_POLL_MILLIS = 500;
|
|
|
|
static void main(String[] args) {
|
|
Path configPath = Path.of(args.length > 0 ? args[0] : "bridged.yaml");
|
|
BridgedConfig cfg = BridgedConfig.load(configPath);
|
|
|
|
// The primary/host env that launched bridged must not be tainted.
|
|
SubscriptionGuard guard = new SubscriptionGuard(cfg.guard().hostSet());
|
|
guard.assertPrimaryClean(System.getenv());
|
|
|
|
// CB-501: refuse to start if the bind is wider than the auth mode can defend. Under
|
|
// loopback-trust, "not a known worker" means "the primary" — sound only because the OS
|
|
// refuses remote connections to a loopback socket. This throws rather than warns so the
|
|
// dangerous configuration cannot be reached by ignoring a log line.
|
|
cfg.validateAuthExposure();
|
|
cfg.validateLeadScan();
|
|
// CB-542: a subscription:true profile whose env: reseats ANTHROPIC_BASE_URL/AUTH_TOKEN would
|
|
// reach an unguarded endpoint (the launcher skips SubscriptionGuard for it). Refuse at load.
|
|
cfg.validateSubscriptionProfiles();
|
|
// CB-548: every architect slot must name a configured workers: profile — the strong-model
|
|
// backend the future spawn lifecycle would read. A stale reference dies here, not later.
|
|
cfg.validateArchitects();
|
|
|
|
Path socket = cfg.herdrSocket() != null && !cfg.herdrSocket().isBlank()
|
|
? Path.of(cfg.herdrSocket())
|
|
: UnixSocketHerdrClient.defaultSocketPath();
|
|
|
|
UnixSocketHerdrClient herdr = UnixSocketHerdrClient.connect(socket, new com.fasterxml.jackson.databind.ObjectMapper());
|
|
|
|
AgentControl agents = new AgentControl(herdr);
|
|
WorkspaceControl spaces = new WorkspaceControl(herdr);
|
|
// CB-402: one adapter per configured peer kind, fronted by a composite router. A profile's
|
|
// `kind:` selects its adapter — claude-code (the default) and opencode partition the profile
|
|
// set — and the composite dispatches each SPI call to the adapter that owns the profile/pane.
|
|
Map<String, BridgedConfig.Worker> claudeProfiles = new LinkedHashMap<>();
|
|
Map<String, BridgedConfig.Worker> opencodeProfiles = new LinkedHashMap<>();
|
|
cfg.workerProfiles().forEach((name, w) -> {
|
|
if (w.isOpenCode()) {
|
|
opencodeProfiles.put(name, w);
|
|
} else {
|
|
claudeProfiles.put(name, w);
|
|
}
|
|
});
|
|
List<HerdrPeerLauncher> adapters = new ArrayList<>();
|
|
// The claude-code adapter is the always-present default; keep it even with no profiles (so a
|
|
// bridge configured with no workers, or opencode-only, still has a well-defined base adapter)
|
|
// unless opencode is the only kind configured.
|
|
if (!claudeProfiles.isEmpty() || opencodeProfiles.isEmpty()) {
|
|
adapters.add(new ClaudeCodeLauncher(agents, spaces, guard,
|
|
claudeProfiles, cfg.defaultProfile(), System::getenv,
|
|
cfg.spawnReadyTimeoutMs(), cfg.spawnReadyPollMs()));
|
|
}
|
|
if (!opencodeProfiles.isEmpty()) {
|
|
adapters.add(new OpenCodeLauncher(agents, spaces,
|
|
opencodeProfiles, cfg.defaultProfile(), System::getenv,
|
|
cfg.spawnReadyTimeoutMs(), cfg.spawnReadyPollMs()));
|
|
}
|
|
AtomicReference<Function<String, Integer>> liveCountRef = new AtomicReference<>(_ -> 0);
|
|
PeerLauncher workers = new CompositePeerLauncher(
|
|
adapters,
|
|
cfg.defaultProfile(),
|
|
cfg.workerProfiles(),
|
|
PlacementPolicies.fromName(cfg.placement()),
|
|
profileName -> liveCountRef.get().apply(profileName));
|
|
// CB-504: under supervision (launchd/systemd) bridged can start before herdr's socket
|
|
// exists. The client itself is lazy — it connects per call — but the orphan reap below is
|
|
// the first thing that actually talks to herdr, so without this wait a boot-order race
|
|
// would crash the daemon into a restart loop. Wait, then degrade rather than die: serving
|
|
// with /healthz reporting "degraded" is strictly more useful than exiting.
|
|
if (awaitHerdr(herdr)) {
|
|
// CB-117: herdr keeps worker panes alive across a daemon restart, and their ids died
|
|
// with the previous process — reap those leaked orphans now, before we start serving.
|
|
workers.reapOrphanWorkers();
|
|
} else {
|
|
log.warn("herdr did not answer within {}s — starting anyway; /healthz will report "
|
|
+ "degraded until it comes up. Orphaned worker panes (if any) were NOT reaped.",
|
|
HERDR_WAIT_SECONDS);
|
|
}
|
|
|
|
// CB-301: authoritative session registry + lifecycle FSM on top of ClaudeCodeLauncher.
|
|
// CB-301-ext: worktree provisioning seam, optionally rooted at a configured directory.
|
|
// CB-303 part 2: context cap is opt-in and disabled (0) when absent/null.
|
|
int contextCap = 0;
|
|
if (cfg.lifecycle() != null && cfg.lifecycle().contextCap() != null
|
|
&& cfg.lifecycle().contextCap() > 0) {
|
|
contextCap = cfg.lifecycle().contextCap();
|
|
}
|
|
boolean clearAfterTurn = cfg.lifecycle() != null && cfg.lifecycle().clearAfterTurn();
|
|
SessionManager sessions = new SessionManager(workers, new GitWorktrees(cfg.worktreeRoot()),
|
|
System::nanoTime, contextCap, clearAfterTurn);
|
|
liveCountRef.set(profileName -> (int) sessions.roster().stream()
|
|
.filter(s -> profileName.equals(s.profile()))
|
|
.count());
|
|
|
|
// CB-303 part 1: idle-ttl reaper — only when configured, defaults to disabled.
|
|
final SessionReaper reaper;
|
|
if (cfg.lifecycle() != null
|
|
&& cfg.lifecycle().idleTtlSeconds() != null
|
|
&& cfg.lifecycle().idleTtlSeconds() > 0) {
|
|
reaper = new SessionReaper(sessions, cfg.lifecycle().idleTtlSeconds());
|
|
reaper.start();
|
|
} else {
|
|
reaper = null;
|
|
}
|
|
|
|
// CB-530: every pane the config names as a lead, merged from `leaders:` and the legacy
|
|
// singular pin. PrimaryRegistry below still tracks ONE terminal — it addresses the push
|
|
// loop's nudges, which need a single destination — so it keeps the legacy pin.
|
|
Map<String, String> leadTerminals = cfg.leaderTerminals();
|
|
if (leadTerminals.size() > 1) {
|
|
log.info("leads: {} panes recognised {}", leadTerminals.size(), leadTerminals.values());
|
|
}
|
|
// CB-531: on top of the static registry, discover leads by the tab labels the operator
|
|
// writes. Opt-in, so a config with no `leadScan:` block resolves exactly as it did under
|
|
// CB-530 — the supplier is then a constant and never touches herdr.
|
|
final Supplier<Map<String, String>> leads;
|
|
if (cfg.leadScan() != null) {
|
|
var scan = cfg.leadScan();
|
|
Set<String> workerSpaces = cfg.workerProfiles().values().stream()
|
|
.map(BridgedConfig.Worker::workspace)
|
|
.filter(Objects::nonNull)
|
|
.collect(Collectors.toSet());
|
|
leads = new LeadTabScanner(herdr, scan.tabPrefix(), workerSpaces, leadTerminals,
|
|
TimeUnit.SECONDS.toNanos(scan.intervalSeconds()), System::nanoTime);
|
|
log.info("lead scan: tabs labelled '{}…' host a lead (rescan every {}s, worker spaces {} "
|
|
+ "excluded)",
|
|
scan.tabPrefix(), scan.intervalSeconds(), workerSpaces);
|
|
} else {
|
|
leads = () -> leadTerminals;
|
|
}
|
|
|
|
// CB-548: config-declared architect slots. Config supplies only the stable name → profile
|
|
// map; the terminal → slot binding is owned by the registry and is empty at startup, so no
|
|
// pane resolves to an architect until the later spawn lifecycle binds one. The registry is
|
|
// what CallerResolver resolves against and what that lifecycle will read profiles from;
|
|
// nothing here spawns a slot.
|
|
ArchitectRegistry architects = new ArchitectRegistry(
|
|
cfg.architects() == null ? Map.of() : cfg.architects());
|
|
if (!architects.slots().isEmpty()) {
|
|
log.info("architect slots: {} configured {} — none bound yet (a slot is idle until the "
|
|
+ "spawn lifecycle binds a live terminal to it)",
|
|
architects.slots().size(), architects.slots().keySet());
|
|
}
|
|
|
|
// Status-gated injector (CB-103): the single writer into workers, fed by a poller.
|
|
// The blocking message endpoint (CB-104) is the producer; the poller is inert until then.
|
|
// CB-106: a confirmed turn completion resolves a blocked send whose worker never replied.
|
|
Rendezvous rendezvous = new Rendezvous();
|
|
CompletionResolver completion = new CompletionResolver(agents, rendezvous);
|
|
// CB-113: deliver only to an available worker (its MCP is connected), never its boot window.
|
|
// CB-301: the manager's presence bridge records availability and drives SPAWNING → READY.
|
|
WorkerPresence presence = sessions.asPresence();
|
|
TurnListener turnListener = new TurnListener() {
|
|
@Override
|
|
public void onTurnComplete(String target) {
|
|
completion.onTurnComplete(target);
|
|
sessions.onTurnComplete(target);
|
|
}
|
|
|
|
@Override
|
|
public boolean hasPostTurnAction(String target) {
|
|
return sessions.hasPostTurnAction(target);
|
|
}
|
|
|
|
@Override
|
|
public boolean onTurnCompleteWithPostAction(String target) {
|
|
completion.resolveBeforePostAction(target);
|
|
return sessions.onTurnCompleteWithPostAction(target);
|
|
}
|
|
|
|
@Override
|
|
public void onDelivered(String target) {
|
|
completion.onDelivered(target);
|
|
sessions.onDelivered(target);
|
|
}
|
|
|
|
@Override
|
|
public void onTurnFailed(String target) {
|
|
completion.onTurnFailed(target);
|
|
sessions.onTurnFailed(target);
|
|
}
|
|
};
|
|
Injector injector = new Injector(agents, turnListener, deliverableTo(presence, leads),
|
|
presence::forget);
|
|
StatusPoller poller = new StatusPoller(agents, injector, INJECT_POLL_MILLIS);
|
|
poller.start();
|
|
|
|
// CB-307: reply inbox. A broker: block (with a uri) selects the AMQP-backed durable adapter;
|
|
// absent, bridged stays soft-state on the in-memory inbox. The AMQP inbox owns a broker
|
|
// connection, so keep the reference to close it in the ordered shutdown hook.
|
|
final ReplyInbox replyInbox;
|
|
if (cfg.broker() != null && cfg.broker().isConfigured()) {
|
|
replyInbox = AmqpReplyInbox.open(cfg.broker().uri());
|
|
log.info("reply inbox: AMQP broker (durable) at {}", cfg.broker().uri());
|
|
} else {
|
|
replyInbox = new InMemoryReplyInbox();
|
|
log.info("reply inbox: in-memory (soft-state)");
|
|
}
|
|
// CB-307: learn the primary's terminal from orchestration tool calls (or pin from config).
|
|
// The pin also feeds CallerResolver below: a primary running inside a herdr pane would
|
|
// otherwise resolve as a worker and be refused every orchestration tool.
|
|
String pinnedPrimaryTerminal = cfg.primary() != null ? cfg.primary().terminal() : null;
|
|
PrimaryRegistry primaryRegistry = new PrimaryRegistry(pinnedPrimaryTerminal);
|
|
// CB-532: `primary.terminal` is superseded and no longer needed for either of its jobs —
|
|
// identity comes from `leaders:`/`leadScan:`, and reply nudges now follow the delegating
|
|
// lead. Say so once at startup rather than leaving a redundant pin to look load-bearing.
|
|
if (pinnedPrimaryTerminal != null && !pinnedPrimaryTerminal.isBlank()) {
|
|
log.warn("primary.terminal is DEPRECATED (CB-532) and can be deleted: identity now comes "
|
|
+ "from leaders:/leadScan:, and reply nudges follow the lead that delegated. "
|
|
+ "It still works, and is still the fallback nudge destination when a restart "
|
|
+ "has lost the delegation map. Its pushReminders/pushBackoffMs stay valid.");
|
|
}
|
|
// CB-307: active push-to-primary loop — nudge the primary when replies land without an
|
|
// open bridge_send. Uses its own lightweight scheduled executor, separate from the injector.
|
|
int maxReminders = cfg.primary() != null ? cfg.primary().remindersOrDefault() : 5;
|
|
long backoffMs = cfg.primary() != null ? cfg.primary().backoffMsOrDefault() : 15_000L;
|
|
var pushScheduler = Executors.newSingleThreadScheduledExecutor(r ->
|
|
Thread.ofVirtual().name("bridge-push-").unstarted(r));
|
|
// CB-502: the registry is built before the service and the push loop so send/reply outcomes
|
|
// are counted at their single funnel rather than at each of the two caller-facing surfaces.
|
|
// CB-512: the push loop takes it too, so nudge outcomes (delivered|exhausted) are counted.
|
|
Metrics metrics = BridgedMetrics.create(sessions, replyInbox);
|
|
var pushLoop = new ReplyPushLoop(primaryRegistry, agents, replyInbox,
|
|
pushScheduler, maxReminders, backoffMs, metrics);
|
|
MessageService messages = new MessageService(agents, injector, rendezvous, replyInbox,
|
|
pushLoop, metrics);
|
|
|
|
// CB-520: the reply inbox only consumes for agents this gateway owns. own on acquire,
|
|
// release on teardown. Do this before CB-516 so the inbox is owned before any reply can land.
|
|
sessions.onAcquire(replyInbox::own);
|
|
// 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
|
|
});
|
|
|
|
// 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(
|
|
new PaneLocator(herdr), new LsofPeerPidLookup(), new LsofProcessCwdLookup());
|
|
|
|
// CB-501: one resolver behind both entry paths. Worker identity still comes from the
|
|
// connection and is never token-gated, so enabling token mode cannot lock the fleet out.
|
|
final CallerResolver callers;
|
|
if (cfg.auth().tokenMode()) {
|
|
String token = System.getenv(cfg.auth().tokenEnv());
|
|
if (token == null || token.isBlank()) {
|
|
throw new IllegalStateException("auth.mode=token but env var " + cfg.auth().tokenEnv()
|
|
+ " is unset or empty — export it before starting bridged");
|
|
}
|
|
callers = CallerResolver.withLeadsAndArchitects(identity, true, token, leads,
|
|
architects::snapshot);
|
|
log.info("auth: token mode (bearer required for non-worker callers, env {})",
|
|
cfg.auth().tokenEnv());
|
|
} else {
|
|
callers = CallerResolver.withLeadsAndArchitects(identity, false, null, leads,
|
|
architects::snapshot);
|
|
log.info("auth: loopback-trust (any loopback non-worker caller is the primary)");
|
|
}
|
|
|
|
BridgeMcp mcp = new BridgeMcp(messages, workers, sessions, identity, presence,
|
|
primaryRegistry, callers, metrics);
|
|
|
|
// CB-303 part 3: single ordered shutdown hook. Drain sessions first while herdr is still
|
|
// open (so releases reach the daemon), then stop poller/message/mcp/reaper, and close herdr
|
|
// last. This replaces the earlier independent hooks that could race and close herdr early.
|
|
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
|
|
sessions.close(cfg.lifecycle() != null ? cfg.lifecycle().drainTimeoutSeconds() : null);
|
|
poller.stop();
|
|
messages.close();
|
|
pushLoop.close();
|
|
mcp.close();
|
|
if (reaper != null) reaper.stop();
|
|
// Release the broker connection last among message resources (no-op for the in-memory inbox).
|
|
if (replyInbox instanceof AutoCloseable closeable) {
|
|
try {
|
|
closeable.close();
|
|
} catch (Exception e) {
|
|
log.debug("reply inbox close: {}", e.toString());
|
|
}
|
|
}
|
|
herdr.close();
|
|
}));
|
|
|
|
Javalin app = new BridgedApp(herdr, workers, sessions, messages, presence, mcp.servlet(),
|
|
callers, metrics).build();
|
|
app.start(cfg.bind().host(), cfg.bind().port());
|
|
log.info("bridged listening on {}:{}, herdr socket {}",
|
|
cfg.bind().host(), cfg.bind().port(), socket);
|
|
}
|
|
|
|
/**
|
|
* The {@link Injector}'s readiness gate (CB-534): a target is deliverable if it is a worker whose
|
|
* agent has connected the bridge MCP, <em>or</em> a lead.
|
|
*
|
|
* <p>The gate exists for one reason — to hold a delivery out of a <em>spawned</em> worker's boot
|
|
* window, where herdr already reports {@code idle} but the TUI would drop an injected paste. That
|
|
* hazard is a property of spawning. A lead is never spawned: the operator started it and named it
|
|
* (or labelled its tab) only once it was up, so there is no boot window to guard.
|
|
*
|
|
* <p>A lead is also never enrolled in {@link WorkerPresence} — {@code BridgeMcp} marks presence
|
|
* only for a worker, deliberately, since that map doubles as the worker roster's availability
|
|
* signal and a lead counted there would show up as an available worker. So without the second
|
|
* disjunct a lead is permanently un-deliverable: every lead→lead send sat on the gate for
|
|
* {@code READINESS_GRACE_POLLS} (~60s) and then failed having never been typed into the pane.
|
|
*
|
|
* <p>The lead set is read through the supplier on each call rather than snapshotted, so a lead
|
|
* discovered by {@code leadScan} after startup becomes deliverable without a restart.
|
|
*/
|
|
static Predicate<String> deliverableTo(WorkerPresence presence, Supplier<Map<String, String>> leads) {
|
|
return target -> presence.isPresent(target) || leads.get().containsKey(target);
|
|
}
|
|
|
|
/**
|
|
* Poll herdr's {@code ping} until it answers or {@link #HERDR_WAIT_SECONDS} elapses (CB-504).
|
|
*
|
|
* @return true if herdr answered, false if it never did
|
|
*/
|
|
private static boolean awaitHerdr(HerdrClient herdr) {
|
|
long deadline = System.nanoTime() + HERDR_WAIT_SECONDS * 1_000_000_000L;
|
|
boolean waited = false;
|
|
while (true) {
|
|
try {
|
|
herdr.call("ping");
|
|
if (waited) {
|
|
log.info("herdr is up");
|
|
}
|
|
return true;
|
|
} catch (HerdrException e) {
|
|
if (System.nanoTime() >= deadline) {
|
|
return false;
|
|
}
|
|
if (!waited) {
|
|
log.info("waiting up to {}s for the herdr socket…", HERDR_WAIT_SECONDS);
|
|
waited = true;
|
|
}
|
|
try {
|
|
Thread.sleep(HERDR_WAIT_POLL_MILLIS);
|
|
} catch (InterruptedException ie) {
|
|
Thread.currentThread().interrupt();
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private Bridged() {
|
|
}
|
|
}
|