Dai Ha
491 lines
27 KiB
Java
491 lines
27 KiB
Java
package dev.ltms.bridged;
|
|
|
|
import dev.ltms.bridged.config.BridgedConfig;
|
|
import dev.ltms.bridged.config.ConfigRef;
|
|
import dev.ltms.bridged.config.ConfigWatcher;
|
|
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.lead.LeadLauncher;
|
|
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.MemberPresence;
|
|
import dev.ltms.bridged.auth.MemberRegistry;
|
|
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.LeadHeartbeatLoop;
|
|
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.member.ClaudeCodeLauncher;
|
|
import dev.ltms.bridged.member.CompositePeerLauncher;
|
|
import dev.ltms.bridged.member.HerdrPeerLauncher;
|
|
import dev.ltms.bridged.member.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);
|
|
|
|
/** 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);
|
|
// CB-559: `cfg` stays the startup snapshot — every validation and every piece of one-time
|
|
// wiring below reads it, and must, because those decisions cannot be unmade. `config` is the
|
|
// live reference the hot paths read per use. Which keys can actually move is ConfigRef's
|
|
// contract; adding a reader here does not make a key reloadable by itself.
|
|
ConfigRef config = new ConfigRef(configPath, cfg);
|
|
|
|
// 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.validateLeadTabPrefixes();
|
|
// 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();
|
|
cfg.validateCharters();
|
|
// 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.validateMembers();
|
|
|
|
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.Profile> claudeProfiles = new LinkedHashMap<>();
|
|
Map<String, BridgedConfig.Profile> opencodeProfiles = new LinkedHashMap<>();
|
|
cfg.profiles().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.effectiveDefaultProfile(), System::getenv,
|
|
cfg.spawnReadyTimeoutMs(), cfg.spawnReadyPollMs(),
|
|
() -> config.get().fleet()));
|
|
}
|
|
if (!opencodeProfiles.isEmpty()) {
|
|
adapters.add(new OpenCodeLauncher(agents, spaces,
|
|
opencodeProfiles, cfg.effectiveDefaultProfile(), System::getenv,
|
|
cfg.spawnReadyTimeoutMs(), cfg.spawnReadyPollMs(),
|
|
() -> config.get().fleet()));
|
|
}
|
|
AtomicReference<Function<String, Integer>> liveCountRef = new AtomicReference<>(_ -> 0);
|
|
PeerLauncher workers = new CompositePeerLauncher(
|
|
adapters,
|
|
cfg.effectiveDefaultProfile(),
|
|
config,
|
|
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.
|
|
boolean herdrUp = awaitHerdr(herdr);
|
|
if (herdrUp) {
|
|
// 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. CB-557 moved the settings onto the lead they describe, so scanning is on whenever
|
|
// a `fleet.leaders:` entry exists — with no leads configured the supplier is a constant and
|
|
// never touches herdr, exactly as a missing `leadScan:` block used to behave.
|
|
final Supplier<Map<String, String>> leads;
|
|
var leaders = cfg.fleet().leaders();
|
|
if (!leaders.isEmpty()) {
|
|
// One scanner, so one prefix and one interval. Distinct per-lead prefixes would need a
|
|
// scanner each; until a config actually wants that, take the first entry's settings and
|
|
// say so, rather than silently honouring one lead's prefix and dropping another's.
|
|
var scan = leaders.values().iterator().next();
|
|
Set<String> memberSpaces = cfg.profiles().values().stream()
|
|
.map(BridgedConfig.Profile::workspace)
|
|
.filter(Objects::nonNull)
|
|
.collect(Collectors.toSet());
|
|
leads = new LeadTabScanner(herdr, scan.tabPrefix(), memberSpaces, leadTerminals,
|
|
TimeUnit.SECONDS.toNanos(scan.scanIntervalSeconds()), System::nanoTime);
|
|
log.info("lead scan: tabs labelled '{}…' host a lead (rescan every {}s, member spaces {} "
|
|
+ "excluded)",
|
|
scan.tabPrefix(), scan.scanIntervalSeconds(), memberSpaces);
|
|
long distinctPrefixes = leaders.values().stream()
|
|
.map(BridgedConfig.Leader::tabPrefix).distinct().count();
|
|
if (distinctPrefixes > 1) {
|
|
log.warn("fleet.leaders declares {} different tabPrefix values; only '{}' is scanned "
|
|
+ "for. Give every lead the same tabPrefix, or leads under the others "
|
|
+ "will not be discovered.",
|
|
distinctPrefixes, scan.tabPrefix());
|
|
}
|
|
} else {
|
|
leads = () -> leadTerminals;
|
|
}
|
|
|
|
// CB-558: start any declared lead that is not already running. After the scanner is built,
|
|
// because both read the same tab labels and the ordering makes that dependency visible; and
|
|
// only when herdr answered, because the launcher's whole safety property is that it can
|
|
// count live leads first — it must never guess and risk a second orchestrator.
|
|
if (herdrUp && !leaders.isEmpty()) {
|
|
int launched = new LeadLauncher(agents, spaces, cfg).ensureLeads();
|
|
if (launched > 0) {
|
|
log.info("lead auto-launch: {} lead(s) started", launched);
|
|
}
|
|
}
|
|
|
|
// 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.
|
|
MemberRegistry members = new MemberRegistry(cfg.fleet());
|
|
sessions.setMemberLifecycle(members);
|
|
if (!members.slots().isEmpty()) {
|
|
log.info("member slots: {} configured {} — none bound yet (a slot is idle until the "
|
|
+ "spawn lifecycle binds a live terminal to it)",
|
|
members.slots().size(), members.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.
|
|
MemberPresence 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, Injector.POLL_INTERVAL_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);
|
|
// CB-551: idle-lead heartbeat. Opt-in; absent `leadHeartbeat:` this is never constructed, so
|
|
// an upgraded daemon cannot silently start spending subscription on nudging an idle lead.
|
|
// It has its own single-thread scheduler and holds its own scheduler shutdown via close().
|
|
final LeadHeartbeatLoop heartbeat;
|
|
var heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r ->
|
|
Thread.ofVirtual().name("bridge-heartbeat-").unstarted(r));
|
|
if (cfg.leadHeartbeat() != null) {
|
|
var hb = cfg.leadHeartbeat();
|
|
heartbeat = new LeadHeartbeatLoop(primaryRegistry, agents, replyInbox, sessions::roster,
|
|
pushLoop, heartbeatScheduler, System::nanoTime,
|
|
TimeUnit.SECONDS.toNanos(hb.idleAfterSeconds()), hb.backoffMs(), hb.quietNudgeCap(),
|
|
metrics);
|
|
heartbeat.start();
|
|
} else {
|
|
heartbeat = null;
|
|
heartbeatScheduler.shutdownNow();
|
|
}
|
|
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.withLeadsAndMembers(identity, true, token, leads, members);
|
|
log.info("auth: token mode (bearer required for non-worker callers, env {})",
|
|
cfg.auth().tokenEnv());
|
|
} else {
|
|
callers = CallerResolver.withLeadsAndMembers(identity, false, null, leads, members);
|
|
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, new BridgeMcp.CapacitySource(profile -> liveCountRef.get().apply(profile),
|
|
profile -> {
|
|
var configured = config.get().profiles().get(profile);
|
|
return configured == null ? null : configured.maxLoad();
|
|
}, () -> config.get().profiles().keySet(), System::nanoTime));
|
|
|
|
// CB-559: opt-in config reload. With no `configReload:` block nothing is constructed, so an
|
|
// upgraded daemon behaves exactly as before — the file is read once at boot and never again.
|
|
final ConfigWatcher configWatcher;
|
|
if (cfg.configReload() != null && cfg.configReload().isEnabled()) {
|
|
configWatcher = new ConfigWatcher(config, cfg.configReload().intervalSeconds());
|
|
configWatcher.start();
|
|
} else {
|
|
configWatcher = null;
|
|
}
|
|
|
|
// 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();
|
|
if (heartbeat != null) heartbeat.close(); // CB-551: stop the idle-lead heartbeat scheduler
|
|
if (configWatcher != null) configWatcher.stop(); // CB-559: stop polling the config file
|
|
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 spawned
|
|
* member 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> member'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 MemberPresence} — {@code BridgeMcp} marks presence
|
|
* for every spawned member (worker and architect), deliberately, since that map doubles as the
|
|
* member roster's availability signal and a lead counted there would show up as an available
|
|
* member. 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(MemberPresence 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() {
|
|
}
|
|
}
|