Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 38f13f4b3c |
@@ -47,18 +47,15 @@ bind:
|
||||
# (say a Claude lead and an opencode lead) work as peers: the second is silently demoted and refused
|
||||
# every orchestration call. List each lead's pane here and all of them resolve as leads.
|
||||
#
|
||||
# tab → the ONLY field identity depends on (CB-579); the exact label of the tab hosting the lead.
|
||||
# Label the tab yourself, or let bridged label one it launches — see `fleet.leaders:` below.
|
||||
# terminal → the ONLY field identity depends on; get it from that session's bridge_whoami
|
||||
# kind/model → descriptive; they document what runs in the pane and are echoed by bridge_whoami
|
||||
#
|
||||
# A lead's tab must already carry its label (or be launched by bridged, which labels it) — there is
|
||||
# no terminal id to paste in and nothing to re-pin when the session restarts: the tab survives, so
|
||||
# the same label resolves the same lead again on the next scan.
|
||||
# A lead is never spawned — it pre-exists, which is exactly why it must be named rather than created.
|
||||
# `bridge_whoami` reports `{"role":"primary","leader":"<name>"}`; role stays "primary" because a lead
|
||||
# IS a primary for authorization, so nothing that keys on the role breaks.
|
||||
#
|
||||
# KEEP `primary:` when adding leads: it still addresses the CB-307 push loop, which needs a single
|
||||
# destination for its nudges, and is a separate mechanism from lead identity — see `fleet.leaders:`.
|
||||
# destination for its nudges. If both name the same terminal, the `fleet.leaders:` entry wins.
|
||||
#
|
||||
# Leads are configured under `fleet.leaders:` — see THE FLEET further down.
|
||||
#
|
||||
@@ -312,18 +309,15 @@ fleet:
|
||||
|
||||
# Panes that orchestrate rather than are orchestrated. A lead may now be CREATED as well as
|
||||
# recognised: give it a `profile:` and the daemon launches the shortfall when fewer than
|
||||
# `instances` are live. Omit `profile:` and it is recognise-only, as before.
|
||||
# `instances` are live. Give it only a `terminal:` and it is recognise-only, as before.
|
||||
#
|
||||
# `tab:` (CB-579) is REQUIRED and is the only field identity depends on — the exact label of the
|
||||
# tab hosting the lead, matched case-insensitively. Label the tab yourself and put that same
|
||||
# string here, and the pane is recognised on the next rescan. Reopen the tab later, or the session
|
||||
# inside it restarts — the terminal id changes; the tab, and its label, do not, so no config edit
|
||||
# follows a restart.
|
||||
# `tabPrefix` is the naming convention that finds a lead without pasting a terminal id: label the
|
||||
# tab `lead: <name>` when you open it and the pane is recognised on the next rescan. Reopen the
|
||||
# tab later and the id changes; the label does not.
|
||||
#
|
||||
# A lead the daemon launches is labelled BY the daemon with this same `tab:` value, so it is found
|
||||
# A lead the daemon launches is labelled BY the daemon, using the same convention, so it is found
|
||||
# by the same scan. A lead counts as live only when herdr also reports a running agent in that
|
||||
# tab — a label left behind by a session that died does not block the relaunch, and a tab that is
|
||||
# gone entirely drops out of the next scan rather than being remembered forever.
|
||||
# tab — a label left behind by a session that died does not block the relaunch.
|
||||
#
|
||||
# An auto-launched lead is NOT a member: it gets no worker reply charter, is never registered with
|
||||
# the session lifecycle (the idle reaper would kill your orchestrator), and stays on the
|
||||
@@ -332,9 +326,10 @@ fleet:
|
||||
# opus-5.0:
|
||||
# profile: opus # omit to never create this lead, only recognise it
|
||||
# instances: 1 # desired live count; only the shortfall is launched. 0 = off
|
||||
# tab: "lead: opus-5.0" # REQUIRED — the exact tab label this lead lives in
|
||||
# tabPrefix: "lead:" # only used to guard against a worker tabLabel colliding with
|
||||
# # this convention at startup; plays no part in matching a lead
|
||||
# terminal: term_0123456789abcd # optional hand-pin; usually found by tabPrefix instead.
|
||||
# # A running agent on this terminal also counts as live, so a
|
||||
# # lead you opened by hand is not relaunched under you.
|
||||
# tabPrefix: "lead:" # `lead: opus-5.0` ⇒ a lead named opus-5.0 (case-insensitive)
|
||||
# scanIntervalSeconds: 10 # rescan cadence, and the worst case before a new tab is seen
|
||||
# workspace: leads # where a launched lead's tab is created (default "leads").
|
||||
# # MUST NOT be a member workspace — those are excluded from the
|
||||
@@ -342,7 +337,7 @@ fleet:
|
||||
# cwd: /path/to/repo # the launched lead's working directory (default: bridged's own)
|
||||
# kind: claude # descriptive; reported by bridge_whoami
|
||||
# gpt-sol-5.6:
|
||||
# tab: "lead: gpt-sol-5.6"
|
||||
# terminal: term_fedcba9876543
|
||||
# kind: opencode
|
||||
# model: openai/gpt-5.6-terra
|
||||
|
||||
|
||||
@@ -193,34 +193,34 @@ public final class Bridged {
|
||||
if (leadTerminals.size() > 1) {
|
||||
log.info("leads: {} panes recognised {}", leadTerminals.size(), leadTerminals.values());
|
||||
}
|
||||
// CB-531: on top of the legacy primary.terminal pin, 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.
|
||||
// CB-579: each lead now names its own exact `tab:` label, so one scanner discovers every
|
||||
// configured lead regardless of how differently their tabs are labelled — the old
|
||||
// single-shared-tabPrefix limitation (and its warning) is gone.
|
||||
// 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());
|
||||
Map<String, String> tabToName = new LinkedHashMap<>();
|
||||
leaders.forEach((name, leader) -> {
|
||||
if (leader != null && leader.tab() != null && !leader.tab().isBlank()) {
|
||||
tabToName.put(leader.tab(), name);
|
||||
}
|
||||
});
|
||||
// One shared rescan cadence: still taken from the first entry, as before — it is an
|
||||
// operational cadence, not identity, so there is no correctness reason to give every
|
||||
// lead its own scanner.
|
||||
int scanIntervalSeconds = leaders.values().iterator().next().scanIntervalSeconds();
|
||||
leads = new LeadTabScanner(herdr, tabToName, memberSpaces,
|
||||
TimeUnit.SECONDS.toNanos(scanIntervalSeconds), System::nanoTime);
|
||||
log.info("lead scan: tabs {} host a lead (rescan every {}s, member spaces {} excluded)",
|
||||
tabToName.keySet(), scanIntervalSeconds, memberSpaces);
|
||||
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;
|
||||
}
|
||||
|
||||
@@ -458,32 +458,24 @@ public record BridgedConfig(
|
||||
* pre-existed, which is why it had to be recognised by configuration rather than created. With
|
||||
* {@code profile} and {@code instances} the daemon may stand one up when none is live, so the
|
||||
* pane no longer has to exist before the daemon does. Recognition still comes first: a lead
|
||||
* already running in its configured {@code tab} is adopted, and only the shortfall is launched.
|
||||
*
|
||||
* <p><b>{@code tab} replaced {@code terminal} (CB-579).</b> A herdr {@code terminal_id} changes
|
||||
* every time the lead's session restarts, so pinning one cost a config edit and a daemon restart
|
||||
* per restart. A tab is stable: a human opens it once, it holds exactly one pane, and its label
|
||||
* survives restarts of the agent inside it — so identity is now the tab label alone.
|
||||
* already running under {@code tabPrefix} is adopted, and only the shortfall is launched.
|
||||
*
|
||||
* @param profile the {@code profiles:} entry to launch this lead on when one must
|
||||
* be created; {@code null} ⇒ recognise-only, never create
|
||||
* @param tab the exact tab label hosting this lead, matched case-insensitively;
|
||||
* the only field identity depends on. Required — a lead with no
|
||||
* {@code tab} can never be discovered, launched or not
|
||||
* @param terminal the lead's herdr {@code terminal_id} when pinned by hand; the only
|
||||
* field identity depends on. {@code null} ⇒ found by {@code tabPrefix}
|
||||
* @param instances how many of this lead should be live (default 1). The daemon
|
||||
* launches only the shortfall, so a restart adopts rather than doubles
|
||||
* @param tabPrefix no longer used to find a lead's tab — {@code tab} is matched
|
||||
* exactly. Its only remaining job is the startup collision guard
|
||||
* ({@link #validateLeadTabPrefixes()}), which still uses it to refuse
|
||||
* a worker {@code tabLabel} template that could be misread as a lead.
|
||||
* Default {@code "lead:"}
|
||||
* @param tabPrefix label prefix marking this lead's tab, matched case-insensitively;
|
||||
* the remainder is the lead's name ({@code "lead: opus"} →
|
||||
* {@code opus}). Default {@code "lead:"}
|
||||
* @param scanIntervalSeconds how long a tab scan is cached before herdr is asked again; also the
|
||||
* worst case before a newly-labelled tab is recognised. Default 10
|
||||
* @param kind which agent runs there ({@code claude}, {@code opencode}, …)
|
||||
* @param model the model or selector it runs, for operators reading the roster
|
||||
*/
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
public record Leader(String profile, String tab, Integer instances, String tabPrefix,
|
||||
public record Leader(String profile, String terminal, Integer instances, String tabPrefix,
|
||||
Integer scanIntervalSeconds, String kind, String model,
|
||||
String workspace, String cwd) {
|
||||
|
||||
@@ -501,13 +493,12 @@ public record BridgedConfig(
|
||||
(scanIntervalSeconds == null || scanIntervalSeconds <= 0) ? 10 : scanIntervalSeconds;
|
||||
workspace = (workspace == null || workspace.isBlank())
|
||||
? DEFAULT_WORKSPACE : workspace.strip();
|
||||
tab = (tab == null || tab.isBlank()) ? null : tab.strip();
|
||||
}
|
||||
|
||||
/** Back-compat 7-arg form — no workspace or cwd, so both take their defaults. */
|
||||
public Leader(String profile, String tab, Integer instances, String tabPrefix,
|
||||
public Leader(String profile, String terminal, Integer instances, String tabPrefix,
|
||||
Integer scanIntervalSeconds, String kind, String model) {
|
||||
this(profile, tab, instances, tabPrefix, scanIntervalSeconds, kind, model, null, null);
|
||||
this(profile, terminal, instances, tabPrefix, scanIntervalSeconds, kind, model, null, null);
|
||||
}
|
||||
|
||||
/** True when this lead may be launched by the daemon rather than only recognised. */
|
||||
@@ -515,9 +506,9 @@ public record BridgedConfig(
|
||||
return profile != null && !profile.isBlank() && instances > 0;
|
||||
}
|
||||
|
||||
/** The tab label an auto-launched instance of this lead gets — its configured {@code tab}. */
|
||||
public String tabLabel() {
|
||||
return tab;
|
||||
/** The tab label an auto-launched instance of this lead gets — what the scanner reads back. */
|
||||
public String tabLabel(String name) {
|
||||
return tabPrefix + " " + name;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -716,20 +707,28 @@ public record BridgedConfig(
|
||||
}
|
||||
|
||||
/**
|
||||
* The terminal → lead-name map seeded from the legacy singular {@code primary:} pin (CB-530).
|
||||
* The terminal → lead-name map that {@link dev.ltms.bridged.auth.CallerResolver} resolves
|
||||
* against, merging the {@code leaders:} registry with the legacy singular {@code primary:} pin.
|
||||
*
|
||||
* <p>{@code fleet.leaders} no longer carries a per-entry terminal pin (CB-579): a lead's identity
|
||||
* comes from its {@code tab} alone, resolved live by {@code LeadTabScanner}. This method now
|
||||
* exists only for the {@code primary.terminal} fallback — a config that never migrated off it
|
||||
* still resolves that one pane as a lead named {@code "primary"}, exactly as before CB-530.
|
||||
* <p>Precedence: an explicit {@code leaders:} entry wins over the {@code primary:} pin for the
|
||||
* same terminal. The pin is the older, less expressive spelling of the same fact, so when both
|
||||
* name a pane the named entry is the one an operator meant. The pin is still honoured on its
|
||||
* own — a config carrying only {@code primary:} behaves exactly as it did before CB-530.
|
||||
*
|
||||
* @return an unmodifiable map, empty when {@code primary.terminal} is not configured (nothing is
|
||||
* pinned, and every pane therefore resolves as a worker — the pre-CB-307 behaviour)
|
||||
* @return an unmodifiable map, empty when neither block is configured (nothing is pinned, and
|
||||
* every pane therefore resolves as a worker — the pre-CB-307 behaviour)
|
||||
*/
|
||||
public Map<String, String> leaderTerminals() {
|
||||
Map<String, String> byTerminal = new LinkedHashMap<>();
|
||||
if (fleet != null) {
|
||||
fleet.leaders().forEach((name, leader) -> {
|
||||
if (leader != null && leader.terminal() != null && !leader.terminal().isBlank()) {
|
||||
byTerminal.put(leader.terminal(), name);
|
||||
}
|
||||
});
|
||||
}
|
||||
if (primary != null && primary.terminal() != null && !primary.terminal().isBlank()) {
|
||||
byTerminal.put(primary.terminal(), "primary");
|
||||
byTerminal.putIfAbsent(primary.terminal(), "primary");
|
||||
}
|
||||
return Collections.unmodifiableMap(byTerminal);
|
||||
}
|
||||
@@ -841,7 +840,6 @@ public record BridgedConfig(
|
||||
try {
|
||||
String yaml = Files.readString(path);
|
||||
rejectRenamedTopLevelKeys(yaml);
|
||||
rejectLeaderTerminalKey(yaml);
|
||||
warnUnknownTopLevelKeys(yaml, path);
|
||||
rejectDuplicateMemberSlots(yaml);
|
||||
BridgedConfig cfg = YAML.readValue(yaml, BridgedConfig.class);
|
||||
@@ -1066,43 +1064,6 @@ public record BridgedConfig(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Reject a config whose {@code fleet.leaders.<name>} still carries the retired {@code terminal:}
|
||||
* pin (CB-579), naming {@code tab:} as its replacement.
|
||||
*
|
||||
* <p>{@code Leader} is {@code @JsonIgnoreProperties(ignoreUnknown = true)}, so simply dropping
|
||||
* the record component would make a leftover {@code terminal:} key silently no-op — the daemon
|
||||
* would start, the pin would never take effect, and nothing would say why. Fatal and specific
|
||||
* instead, exactly like {@link #rejectRenamedTopLevelKeys}, which this mirrors for a key one
|
||||
* level deeper than the ones that method covers.
|
||||
*
|
||||
* @param yaml the raw config text
|
||||
* @throws IllegalStateException when any {@code fleet.leaders.<name>.terminal} key is present
|
||||
*/
|
||||
static void rejectLeaderTerminalKey(String yaml) {
|
||||
Map<?, ?> raw;
|
||||
try {
|
||||
raw = YAML.readValue(yaml, Map.class);
|
||||
} catch (IOException | IllegalArgumentException e) {
|
||||
return; // a malformed file is reported by the real parse, not here
|
||||
}
|
||||
if (raw == null || !(raw.get("fleet") instanceof Map<?, ?> fleet)
|
||||
|| !(fleet.get("leaders") instanceof Map<?, ?> leaders)) {
|
||||
return;
|
||||
}
|
||||
List<String> bad = leaders.entrySet().stream()
|
||||
.filter(e -> e.getValue() instanceof Map<?, ?> leader && leader.containsKey("terminal"))
|
||||
.map(e -> String.valueOf(e.getKey()))
|
||||
.sorted()
|
||||
.toList();
|
||||
if (!bad.isEmpty()) {
|
||||
throw new IllegalStateException("refusing to start: fleet.leaders entries ["
|
||||
+ String.join(", ", bad) + "] still use the retired 'terminal:' key — replace it "
|
||||
+ "with 'tab:', the exact tab label hosting the lead. A terminal_id changes on "
|
||||
+ "every restart of the lead's session; a tab label does not.");
|
||||
}
|
||||
}
|
||||
|
||||
static List<String> unknownTopLevelKeys(String yaml) {
|
||||
Map<?, ?> raw;
|
||||
try {
|
||||
@@ -1352,9 +1313,10 @@ public record BridgedConfig(
|
||||
+ "', which is not a configured profiles: entry (have: " + profiles.keySet()
|
||||
+ ").");
|
||||
}
|
||||
if (leader.tab() == null || leader.tab().isBlank()) {
|
||||
bad.add("fleet.leaders." + name + " has no tab: — a lead is now found (and, if "
|
||||
+ "auto-launched, labelled) purely by its tab, so every entry must name one.");
|
||||
if (!leader.isCreatable() && (leader.terminal() == null || leader.terminal().isBlank())) {
|
||||
bad.add("fleet.leaders." + name + " can neither be found nor created — it pins no "
|
||||
+ "terminal: and names no profile: to launch one on. Give it one or the "
|
||||
+ "other, or drop the entry.");
|
||||
}
|
||||
});
|
||||
if (!bad.isEmpty()) {
|
||||
|
||||
@@ -30,11 +30,7 @@ import java.util.function.Supplier;
|
||||
* {@code fleet:} (every role pool, {@code charters}, and {@code tabLabel}),
|
||||
* {@code placement:}, and an existing profile's {@code weight} / {@code maxLoad}. Those
|
||||
* three are read through a supplier on {@code CompositePeerLauncher}, which is what makes
|
||||
* them hot — not the fact that they are config. <strong>This does NOT include
|
||||
* {@code fleet.leaders}</strong>: {@code Bridged.main} reads {@code cfg.fleet().leaders()}
|
||||
* once at startup to build the {@code LeadTabScanner} and the {@code LeadLauncher}, and
|
||||
* neither is reconstructed on reload — so a lead added, removed, or re-{@code tab}'d under
|
||||
* {@code fleet.leaders} needs a restart, the same as any deferred key below.</li>
|
||||
* them hot — not the fact that they are config.</li>
|
||||
* <li><strong>Deferred</strong> — accepted into the new snapshot, but the wiring built at startup
|
||||
* keeps the old value until a restart: {@code lifecycle:}, {@code leadHeartbeat:},
|
||||
* {@code spawnReadyTimeoutMs} / {@code spawnReadyPollMs}, {@code guard:},
|
||||
|
||||
@@ -6,7 +6,6 @@ import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.function.LongSupplier;
|
||||
@@ -24,15 +23,6 @@ import java.util.function.Supplier;
|
||||
* by first starting the session and asking it. Scanning closes that loop: label the tab, and the
|
||||
* pane is recognised on the next resolve.
|
||||
*
|
||||
* <p><strong>CB-579 — matched by name, not prefix.</strong> This used to strip one shared
|
||||
* {@code tabPrefix} off a label to derive the lead's name, and merged a config-supplied
|
||||
* {@code terminal_id} pin over every scan result so the pin could never expire. Both are gone: each
|
||||
* lead now configures its own exact {@code tab} label ({@code fleet.leaders.<name>.tab}), so this
|
||||
* class is handed a {@code tab → name} map up front and matches labels against it exactly
|
||||
* (case-insensitively). There is no merge step — a scan result is the whole answer. That is the
|
||||
* fix for the bug this replaces: a {@code terminal_id} pin surviving in config after the pane it
|
||||
* named was gone, so the daemon kept treating a dead session as a live lead forever.
|
||||
*
|
||||
* <p><strong>Direction of trust.</strong> The label names the lead; it never <em>grants</em>
|
||||
* anything a pane could take for itself. Three properties keep that honest:
|
||||
* <ol>
|
||||
@@ -58,7 +48,7 @@ import java.util.function.Supplier;
|
||||
* ever make a decision that <em>removes</em> something based on this map, add the same check.
|
||||
* The remaining hazard is an <em>operator</em> one — a worker {@code tabLabel} template that
|
||||
* happens to start with the same prefix would promote the whole fleet — and that is refused at
|
||||
* startup by {@code BridgedConfig.validateLeadTabPrefixes} rather than documented here.
|
||||
* startup by {@code BridgedConfig.validateLeadScan} rather than documented here.
|
||||
*
|
||||
* <p><strong>Caching.</strong> {@link #get()} is on the request path (every resolve), so the scan
|
||||
* is TTL-cached and a stale-but-valid map is preferred to a herdr round-trip. A failed scan keeps
|
||||
@@ -70,46 +60,38 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
|
||||
private static final Logger log = LoggerFactory.getLogger(LeadTabScanner.class);
|
||||
|
||||
private final HerdrClient herdr;
|
||||
private final Map<String, String> tabToName;
|
||||
private final String tabPrefix;
|
||||
private final Set<String> excludedWorkspaceLabels;
|
||||
private final Map<String, String> configuredLeads;
|
||||
private final long ttlNanos;
|
||||
private final LongSupplier clock;
|
||||
|
||||
private Map<String, String> cached = Map.of();
|
||||
private Map<String, String> cached;
|
||||
private long scannedAtNanos;
|
||||
private boolean everScanned;
|
||||
|
||||
/**
|
||||
* @param herdr the herdr client to query ({@code workspace.list},
|
||||
* {@code tab.list}, {@code pane.list} — all read-only)
|
||||
* @param tabToName every configured lead's exact tab label → its name
|
||||
* ({@code fleet.leaders.<name>.tab}), matched case-insensitively
|
||||
* @param tabPrefix a tab whose label starts with this (case-insensitively) hosts a
|
||||
* lead; the rest of the label, trimmed, is the lead's name
|
||||
* @param excludedWorkspaceLabels workspaces never scanned — the configured worker spaces
|
||||
* @param configuredLeads the static {@code leaders:}/{@code primary:} registry, merged
|
||||
* over every scan result. Explicit config outranks the
|
||||
* convention, and survives a scan that cannot run at all
|
||||
* @param ttlNanos how long a scan result is reused before the next one
|
||||
* @param clock nanosecond time source ({@code System::nanoTime} in production)
|
||||
*/
|
||||
public LeadTabScanner(HerdrClient herdr, Map<String, String> tabToName,
|
||||
Set<String> excludedWorkspaceLabels, long ttlNanos, LongSupplier clock) {
|
||||
public LeadTabScanner(HerdrClient herdr, String tabPrefix, Set<String> excludedWorkspaceLabels,
|
||||
Map<String, String> configuredLeads, long ttlNanos, LongSupplier clock) {
|
||||
this.herdr = herdr;
|
||||
this.tabToName = normalize(tabToName);
|
||||
this.tabPrefix = tabPrefix == null || tabPrefix.isBlank() ? "lead:" : tabPrefix.strip();
|
||||
this.excludedWorkspaceLabels = excludedWorkspaceLabels == null
|
||||
? Set.of() : Set.copyOf(excludedWorkspaceLabels);
|
||||
this.configuredLeads = configuredLeads == null ? Map.of() : Map.copyOf(configuredLeads);
|
||||
this.ttlNanos = ttlNanos;
|
||||
this.clock = clock;
|
||||
}
|
||||
|
||||
/** Keys stripped and lower-cased once, so every lookup is a plain map hit. */
|
||||
private static Map<String, String> normalize(Map<String, String> tabToName) {
|
||||
if (tabToName == null || tabToName.isEmpty()) {
|
||||
return Map.of();
|
||||
}
|
||||
Map<String, String> out = new LinkedHashMap<>();
|
||||
tabToName.forEach((tab, name) -> {
|
||||
if (tab != null && !tab.isBlank() && name != null && !name.isBlank()) {
|
||||
out.put(tab.strip().toLowerCase(Locale.ROOT), name);
|
||||
}
|
||||
});
|
||||
return Collections.unmodifiableMap(out);
|
||||
this.cached = this.configuredLeads;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -169,20 +151,25 @@ public final class LeadTabScanner implements Supplier<Map<String, String>> {
|
||||
}
|
||||
}
|
||||
}
|
||||
byTerminal.putAll(configuredLeads); // an explicit pin outranks a label
|
||||
return Collections.unmodifiableMap(byTerminal);
|
||||
}
|
||||
|
||||
/**
|
||||
* The lead name a tab label declares, or {@code null} if it names none of the configured leads.
|
||||
* The lead name a tab label declares, or {@code null} if it declares none.
|
||||
*
|
||||
* <p>Exact match (case-insensitive, ends stripped) against {@link #tabToName} — no prefix
|
||||
* stripping, so an operator's {@code "lead: something-else"} tab is never mistaken for a
|
||||
* configured lead just because it shares a prefix.
|
||||
* <p>{@code "lead: opus-5.0"} → {@code "opus-5.0"}. A bare {@code "lead:"} names nobody and is
|
||||
* rejected: an unnamed lead would resolve as {@code PRIMARY} with nothing to attribute it to.
|
||||
*/
|
||||
private String leadNameOf(String label) {
|
||||
if (label == null) {
|
||||
return null;
|
||||
}
|
||||
return tabToName.get(label.strip().toLowerCase(Locale.ROOT));
|
||||
String l = label.strip();
|
||||
if (!l.regionMatches(true, 0, tabPrefix, 0, tabPrefix.length())) {
|
||||
return null;
|
||||
}
|
||||
String name = l.substring(tabPrefix.length()).strip();
|
||||
return name.isEmpty() ? null : name;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -59,7 +59,7 @@ public final class LeadLauncher {
|
||||
/**
|
||||
* @param agents herdr agent control (start, list)
|
||||
* @param spaces workspace / tab control (ensure, create, label, list)
|
||||
* @param cfg the loaded config — {@code fleet.leaders}, {@code profiles} and each lead's tab
|
||||
* @param cfg the loaded config — {@code fleet.leaders}, {@code profiles} and the lead pins
|
||||
*/
|
||||
public LeadLauncher(AgentControl agents, WorkspaceControl spaces, BridgedConfig cfg) {
|
||||
this.agents = agents;
|
||||
@@ -102,8 +102,8 @@ public final class LeadLauncher {
|
||||
continue;
|
||||
}
|
||||
if (!lead.isCreatable()) {
|
||||
// A lead with a `tab:` but no `profile:` is recognise-only by design: the operator
|
||||
// opens it by hand. Say so once rather than looking like a silent failure.
|
||||
// A lead with a `terminal:` pin and no `profile:` is recognise-only by design: the
|
||||
// operator opens it by hand. Say so once rather than looking like a silent failure.
|
||||
log.info("lead '{}' is not live, and names no profile — it can be recognised but not "
|
||||
+ "launched. Add `profile:` under fleet.leaders.{} to have bridged start it.",
|
||||
name, name);
|
||||
@@ -127,16 +127,18 @@ public final class LeadLauncher {
|
||||
}
|
||||
|
||||
/**
|
||||
* How many live leads exist per configured name: a running agent in a tab labelled with that
|
||||
* lead's exact {@code tab} (CB-579). Member workspaces are excluded, exactly as the scanner
|
||||
* excludes them: a member must not be counted as a lead because it happens to sit in a matching
|
||||
* tab.
|
||||
* How many live leads exist per configured name.
|
||||
*
|
||||
* <p>There used to be a second path here — a running agent on the terminal a
|
||||
* {@code fleet.leaders.<name>.terminal} pin named, for a lead opened and pinned by hand. That
|
||||
* pin is retired: {@code tab} is now the only field identity depends on, and {@link Agent}
|
||||
* already carries {@link Agent#tabId()} directly, so a hand-opened lead is found the same way an
|
||||
* auto-launched one is — by labelling its tab to match.
|
||||
* <p>Two independent pieces of evidence, because either alone double-spawns:
|
||||
* <ul>
|
||||
* <li>a running agent in a tab labelled {@code "<tabPrefix> <name>"} — how an auto-launched
|
||||
* lead, or an operator following the labelling convention, is found;</li>
|
||||
* <li>a running agent on a terminal the config pins in {@code fleet.leaders.<name>.terminal} —
|
||||
* how a lead the operator opened and pinned by hand is found. Without this, a pinned lead
|
||||
* whose tab carries no matching label would be relaunched on every boot.</li>
|
||||
* </ul>
|
||||
* Member workspaces are excluded, exactly as the scanner excludes them: a member must not be
|
||||
* counted as a lead because it happens to sit in a matching tab.
|
||||
*/
|
||||
private Map<String, Integer> liveLeads(Map<String, BridgedConfig.Leader> leaders) {
|
||||
Set<String> memberSpaces = cfg.profiles().values().stream()
|
||||
@@ -158,9 +160,20 @@ public final class LeadLauncher {
|
||||
}
|
||||
}
|
||||
|
||||
// terminalId → the lead name the config pins it to.
|
||||
Map<String, String> nameByPinnedTerminal = new LinkedHashMap<>();
|
||||
leaders.forEach((name, lead) -> {
|
||||
if (lead.terminal() != null && !lead.terminal().isBlank()) {
|
||||
nameByPinnedTerminal.put(lead.terminal().strip(), name);
|
||||
}
|
||||
});
|
||||
|
||||
Map<String, Integer> counts = new LinkedHashMap<>();
|
||||
for (Agent a : agents.list()) {
|
||||
String name = nameByTab.get(a.tabId());
|
||||
if (name == null) {
|
||||
name = nameByPinnedTerminal.get(a.terminalId());
|
||||
}
|
||||
if (name != null) {
|
||||
counts.merge(name, 1, Integer::sum);
|
||||
}
|
||||
@@ -171,7 +184,7 @@ public final class LeadLauncher {
|
||||
/**
|
||||
* The configured lead a tab label names, or {@code null} for a label that names none.
|
||||
*
|
||||
* <p>Matched exactly (case-insensitively) against each lead's configured {@code tab}, so an
|
||||
* <p>Matched against the declared lead names rather than by splitting on the prefix, so an
|
||||
* operator's {@code "lead: something-else"} tab is not mistaken for a configured lead.
|
||||
*/
|
||||
private String leadNameOf(String label, Map<String, BridgedConfig.Leader> leaders) {
|
||||
@@ -180,8 +193,7 @@ public final class LeadLauncher {
|
||||
}
|
||||
String l = label.strip();
|
||||
for (Map.Entry<String, BridgedConfig.Leader> e : leaders.entrySet()) {
|
||||
String tab = e.getValue().tabLabel();
|
||||
if (tab != null && l.equalsIgnoreCase(tab.strip())) {
|
||||
if (l.equalsIgnoreCase(e.getValue().tabLabel(e.getKey()).strip())) {
|
||||
return e.getKey();
|
||||
}
|
||||
}
|
||||
@@ -190,7 +202,7 @@ public final class LeadLauncher {
|
||||
|
||||
/** Start one lead. Returns false (having logged) rather than throwing on any failure. */
|
||||
private boolean launch(String name, BridgedConfig.Leader lead, BridgedConfig.Profile profile) {
|
||||
String label = lead.tabLabel();
|
||||
String label = lead.tabLabel(name);
|
||||
String cwd = (lead.cwd() == null || lead.cwd().isBlank())
|
||||
? System.getProperty("user.dir") : lead.cwd();
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionException;
|
||||
@@ -170,9 +171,8 @@ public final class MessageService {
|
||||
private final Metrics metrics; // CB-502: nullable — no registry in unit tests
|
||||
private final ConcurrentHashMap<String, ReentrantLock> sessionLocks = new ConcurrentHashMap<>();
|
||||
private final ConcurrentHashMap<String, Task> tasks = new ConcurrentHashMap<>();
|
||||
/** Async task that owns each exact forward rendezvous waiter. */
|
||||
private final ConcurrentHashMap<CompletableFuture<Rendezvous.Resolution>, Task> asyncTasksByWaiter =
|
||||
new ConcurrentHashMap<>();
|
||||
/** Async tasks that have accepted delivery for a target. */
|
||||
private final ConcurrentHashMap<String, Set<Task>> asyncTasksByTarget = new ConcurrentHashMap<>();
|
||||
/** Async tickets paused on a specific {@code bridge_ask} turn. */
|
||||
private final ConcurrentHashMap<String, Task> asyncTasksByTurn = new ConcurrentHashMap<>();
|
||||
private final AtomicLong ticketSeq = new AtomicLong();
|
||||
@@ -385,9 +385,6 @@ public final class MessageService {
|
||||
// failed send leaves no stale waiter behind.
|
||||
CompletableFuture<Rendezvous.Resolution> reply = rendezvous.open(target);
|
||||
try {
|
||||
if (task != null) {
|
||||
asyncTasksByWaiter.put(reply, task);
|
||||
}
|
||||
TurnToken token = new TurnToken(target, reply);
|
||||
// The send has won the lock; the accepted-delivery hook records delegator ownership
|
||||
// here (CB-548). It runs BEFORE enqueue so a throwing hook — onAccepted is now a
|
||||
@@ -412,7 +409,6 @@ public final class MessageService {
|
||||
throw new IllegalStateException("interrupted awaiting reply from " + target, e);
|
||||
}
|
||||
} finally {
|
||||
asyncTasksByWaiter.remove(reply);
|
||||
rendezvous.close(target, reply);
|
||||
}
|
||||
} finally {
|
||||
@@ -438,15 +434,11 @@ public final class MessageService {
|
||||
if (ticket.fresh()) {
|
||||
// Register the reverse waiter first, then surface the question — so the answer, which can
|
||||
// arrive the instant the primary reacts, always finds an open waiter to resolve.
|
||||
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(workerSession);
|
||||
Task task = markAsyncQuestion(waiter, question, ticket.turnId());
|
||||
if (!rendezvous.resolveQuestion(workerSession, question, ticket.turnId())) {
|
||||
if (task != null) {
|
||||
clearAsyncQuestion(ticket.turnId(), true);
|
||||
}
|
||||
rendezvous.closeAsk(ticket.turnId());
|
||||
return new AskResult(AskOutcome.NO_WAITER, null); // no primary is blocked on this worker
|
||||
}
|
||||
markAsyncQuestion(workerSession, question, ticket.turnId());
|
||||
}
|
||||
try {
|
||||
String answer = ticket.answer().get(timeoutMillis, TimeUnit.MILLISECONDS);
|
||||
@@ -545,7 +537,13 @@ public final class MessageService {
|
||||
tasks.put(ticket, task);
|
||||
asyncExecutor.submit(() -> {
|
||||
try {
|
||||
Reply result = send(target, content, ASYNC_TIMEOUT_MS, onAccepted, task);
|
||||
Runnable trackingAccepted = () -> {
|
||||
if (onAccepted != null) {
|
||||
onAccepted.run();
|
||||
}
|
||||
asyncTasksByTarget.computeIfAbsent(target, _ -> ConcurrentHashMap.newKeySet()).add(task);
|
||||
};
|
||||
Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted, task);
|
||||
if (result.outcome() == Outcome.QUESTION) {
|
||||
// Keep the accepted owner until answer() finishes it. markAsyncQuestion may run
|
||||
// just after resolveQuestion wakes this thread.
|
||||
@@ -554,6 +552,7 @@ public final class MessageService {
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
task.future.completeExceptionally(t);
|
||||
untrackAsyncTarget(task);
|
||||
}
|
||||
});
|
||||
pruneTerminalTickets();
|
||||
@@ -615,14 +614,17 @@ public final class MessageService {
|
||||
}
|
||||
|
||||
/** Record the active question for an async ticket; blocking sends have no entry and stay unchanged. */
|
||||
private Task markAsyncQuestion(CompletableFuture<Rendezvous.Resolution> waiter, String text, String turnId) {
|
||||
Task task = waiter == null ? null : asyncTasksByWaiter.get(waiter);
|
||||
private void markAsyncQuestion(String target, String text, String turnId) {
|
||||
Set<Task> targetTasks = asyncTasksByTarget.get(target);
|
||||
Task task = targetTasks == null ? null : targetTasks.stream()
|
||||
.filter(candidate -> !candidate.future.isDone())
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
if (task != null) {
|
||||
task.question = new Reply(Outcome.QUESTION, text, turnId);
|
||||
task.turnId = turnId;
|
||||
asyncTasksByTurn.put(turnId, task);
|
||||
}
|
||||
return task;
|
||||
}
|
||||
|
||||
/** Clear an answered or lapsed question, but only when it matches the ticket's current turn. */
|
||||
@@ -640,6 +642,7 @@ public final class MessageService {
|
||||
/** Complete and detach an async ticket after its worker's actual terminal reply. */
|
||||
private void finishAsyncTask(Task task, Reply result) {
|
||||
task.future.complete(result);
|
||||
untrackAsyncTarget(task);
|
||||
if (task.turnId != null) {
|
||||
asyncTasksByTurn.remove(task.turnId, task);
|
||||
}
|
||||
@@ -658,6 +661,17 @@ public final class MessageService {
|
||||
return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));
|
||||
}
|
||||
|
||||
/** Stop tracking a task once it no longer owns an accepted target turn. */
|
||||
private void untrackAsyncTarget(Task task) {
|
||||
Set<Task> targetTasks = asyncTasksByTarget.get(task.target);
|
||||
if (targetTasks != null) {
|
||||
targetTasks.remove(task);
|
||||
if (targetTasks.isEmpty()) {
|
||||
asyncTasksByTarget.remove(task.target, targetTasks);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Release the async executor. */
|
||||
public void close() {
|
||||
asyncExecutor.shutdown();
|
||||
|
||||
@@ -194,7 +194,7 @@ class BridgedConfigTest {
|
||||
fleet:
|
||||
leaders:
|
||||
opus:
|
||||
tab: "lead: opus"
|
||||
terminal: term_opus
|
||||
""");
|
||||
|
||||
BridgedConfig.Leader lead = BridgedConfig.load(f).fleet().leaders().get("opus");
|
||||
@@ -212,7 +212,7 @@ class BridgedConfigTest {
|
||||
fleet:
|
||||
leaders:
|
||||
opus:
|
||||
tab: "drive: opus"
|
||||
terminal: term_opus
|
||||
tabPrefix: "drive:"
|
||||
scanIntervalSeconds: 30
|
||||
""");
|
||||
@@ -224,9 +224,7 @@ class BridgedConfigTest {
|
||||
|
||||
/**
|
||||
* The pane no longer has to exist before the daemon does (CB-557): a lead naming a profile may
|
||||
* be launched, while one that names no profile is recognised and never created. Either way it
|
||||
* still needs its own {@code tab:} (CB-579) — that part is unconditional, see
|
||||
* {@link #aLeadWithNoTabRefusesToStart}.
|
||||
* be launched, while one that names only a terminal is recognised and never created.
|
||||
*/
|
||||
@Test
|
||||
void aLeadIsCreatableOnlyWhenItNamesAProfile(@TempDir Path dir) throws Exception {
|
||||
@@ -241,9 +239,8 @@ class BridgedConfigTest {
|
||||
leaders:
|
||||
launched:
|
||||
profile: opus
|
||||
tab: "lead: launched"
|
||||
pinned:
|
||||
tab: "lead: pinned"
|
||||
terminal: term_opus
|
||||
""");
|
||||
|
||||
var leaders = BridgedConfig.load(f).fleet().leaders();
|
||||
@@ -252,12 +249,8 @@ class BridgedConfigTest {
|
||||
"no profile to launch on ⇒ recognise-only, the pre-CB-557 behaviour");
|
||||
}
|
||||
|
||||
/**
|
||||
* CB-579: {@code tab} is the only field a lead's identity depends on now, so it is required
|
||||
* whether the entry is creatable or recognise-only — without it the entry can never be found.
|
||||
*/
|
||||
@Test
|
||||
void aLeadWithNoTabRefusesToStart(@TempDir Path dir) throws Exception {
|
||||
void aLeadThatCanBeNeitherFoundNorCreatedRefusesToStart(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("useless-lead.yaml");
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
@@ -271,80 +264,6 @@ class BridgedConfigTest {
|
||||
|
||||
IllegalStateException e = assertThrows(IllegalStateException.class, cfg::validateMembers);
|
||||
assertTrue(e.getMessage().contains("ghost"), "the message must name the useless entry");
|
||||
assertTrue(e.getMessage().contains("tab:"), "the message must say what is missing");
|
||||
}
|
||||
|
||||
/**
|
||||
* CB-579 acceptance (2): a config still spelling {@code fleet.leaders.<name>.terminal} must fail
|
||||
* loudly at load, not be silently dropped by {@code Leader}'s {@code @JsonIgnoreProperties}.
|
||||
*/
|
||||
@Test
|
||||
void aLeaderTerminalKeyFailsLoadAndNamesTabAsTheReplacement(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("stale-terminal.yaml");
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
port: 8080
|
||||
fleet:
|
||||
leaders:
|
||||
opus:
|
||||
terminal: term_opus
|
||||
""");
|
||||
|
||||
IllegalStateException e =
|
||||
assertThrows(IllegalStateException.class, () -> BridgedConfig.load(f));
|
||||
assertTrue(e.getMessage().contains("opus"), "the message must name the offending entry");
|
||||
assertTrue(e.getMessage().contains("tab:"), "the message must name the replacement key");
|
||||
assertTrue(e.getMessage().contains("terminal"), "the message must name the retired key");
|
||||
}
|
||||
|
||||
/** The same refusal, and it must name every offending entry, not just the first. */
|
||||
@Test
|
||||
void everyLeaderStillUsingTerminalIsReportedAtOnce(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("stale-terminals.yaml");
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
port: 8080
|
||||
fleet:
|
||||
leaders:
|
||||
opus:
|
||||
terminal: term_opus
|
||||
sol:
|
||||
terminal: term_sol
|
||||
""");
|
||||
|
||||
IllegalStateException e =
|
||||
assertThrows(IllegalStateException.class, () -> BridgedConfig.load(f));
|
||||
assertTrue(e.getMessage().contains("opus"));
|
||||
assertTrue(e.getMessage().contains("sol"));
|
||||
}
|
||||
|
||||
/** A {@code terminal:} anywhere else in the document (not under a leader entry) is unaffected. */
|
||||
@Test
|
||||
void aTerminalKeyOutsideFleetLeadersIsNotRejected(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("primary-terminal-ok.yaml");
|
||||
Files.writeString(f, "bind:\n port: 8080\nprimary:\n terminal: term_fixed\n");
|
||||
|
||||
assertDoesNotThrow(() -> BridgedConfig.load(f));
|
||||
}
|
||||
|
||||
/** CB-579 acceptance (3): distinct `tab:` labels need no shared prefix — one scanner finds both. */
|
||||
@Test
|
||||
void twoLeadersWithDifferentTabsAreBothConfigured(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("two-tabs.yaml");
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
port: 8080
|
||||
fleet:
|
||||
leaders:
|
||||
opus:
|
||||
tab: "lead: opus"
|
||||
sol:
|
||||
tab: "captain: sol"
|
||||
""");
|
||||
|
||||
var leaders = BridgedConfig.load(f).fleet().leaders();
|
||||
assertEquals("lead: opus", leaders.get("opus").tab());
|
||||
assertEquals("captain: sol", leaders.get("sol").tab());
|
||||
}
|
||||
|
||||
// ── CB-551: the idle-lead heartbeat ─────────────────────────────────────────────────────────
|
||||
@@ -409,7 +328,7 @@ class BridgedConfigTest {
|
||||
fleet:
|
||||
leaders:
|
||||
opus:
|
||||
tab: "lead: opus"
|
||||
terminal: term_opus
|
||||
tabPrefix: "lead:"
|
||||
""");
|
||||
BridgedConfig cfg = BridgedConfig.load(f);
|
||||
@@ -430,7 +349,7 @@ class BridgedConfigTest {
|
||||
tabLabel: "lead: {role} {profile}"
|
||||
leaders:
|
||||
opus:
|
||||
tab: "lead: opus"
|
||||
terminal: term_opus
|
||||
""");
|
||||
BridgedConfig cfg = BridgedConfig.load(f);
|
||||
|
||||
@@ -455,7 +374,7 @@ class BridgedConfigTest {
|
||||
fleet:
|
||||
leaders:
|
||||
opus:
|
||||
tab: "lead: opus"
|
||||
terminal: term_opus
|
||||
""");
|
||||
|
||||
assertDoesNotThrow(() -> BridgedConfig.load(f).validateLeadTabPrefixes());
|
||||
@@ -481,7 +400,7 @@ class BridgedConfigTest {
|
||||
"a label that collides with a convention nobody reads is not a problem");
|
||||
}
|
||||
|
||||
// ── CB-530/CB-579: the leaders registry ─────────────────────────────────────────────────────
|
||||
// ── CB-530: the leaders registry ────────────────────────────────────────────────────────────
|
||||
|
||||
@Test
|
||||
void leadersBlockRegistersEveryPaneByName(@TempDir Path dir) throws Exception {
|
||||
@@ -492,10 +411,10 @@ class BridgedConfigTest {
|
||||
fleet:
|
||||
leaders:
|
||||
opus-5.0:
|
||||
tab: "lead: opus-5.0"
|
||||
terminal: term_opus
|
||||
kind: claude
|
||||
gpt-sol-5.6:
|
||||
tab: "lead: gpt-sol-5.6"
|
||||
terminal: term_sol
|
||||
kind: opencode
|
||||
model: openai/gpt-5.6-terra
|
||||
""");
|
||||
@@ -506,9 +425,9 @@ class BridgedConfigTest {
|
||||
assertEquals(Set.of("opus-5.0", "gpt-sol-5.6"), leaders.keySet());
|
||||
assertEquals("opencode", leaders.get("gpt-sol-5.6").kind());
|
||||
assertEquals("openai/gpt-5.6-terra", leaders.get("gpt-sol-5.6").model());
|
||||
// Identity is the tab now (CB-579) — both entries carry their own, distinct label.
|
||||
assertEquals("lead: opus-5.0", leaders.get("opus-5.0").tab());
|
||||
assertEquals("lead: gpt-sol-5.6", leaders.get("gpt-sol-5.6").tab());
|
||||
// The whole point: BOTH panes resolve as leads, so neither is demoted to worker.
|
||||
assertEquals(Map.of("term_opus", "opus-5.0", "term_sol", "gpt-sol-5.6"),
|
||||
cfg.leaderTerminals());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -520,6 +439,42 @@ class BridgedConfigTest {
|
||||
"configs that never migrate must behave exactly as they did before CB-530");
|
||||
}
|
||||
|
||||
@Test
|
||||
void anExplicitLeadersEntryWinsOverThePinForTheSameTerminal(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("both.yaml");
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
port: 8080
|
||||
primary:
|
||||
terminal: term_shared
|
||||
fleet:
|
||||
leaders:
|
||||
opus-5.0:
|
||||
terminal: term_shared
|
||||
""");
|
||||
|
||||
assertEquals(Map.of("term_shared", "opus-5.0"), BridgedConfig.load(f).leaderTerminals(),
|
||||
"the pin is the older spelling of the same fact; the named entry is what was meant");
|
||||
}
|
||||
|
||||
@Test
|
||||
void bothBlocksTogetherRegisterTheUnionOfTheirTerminals(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("union.yaml");
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
port: 8080
|
||||
primary:
|
||||
terminal: term_pinned
|
||||
fleet:
|
||||
leaders:
|
||||
gpt-sol-5.6:
|
||||
terminal: term_sol
|
||||
""");
|
||||
|
||||
assertEquals(Map.of("term_pinned", "primary", "term_sol", "gpt-sol-5.6"),
|
||||
BridgedConfig.load(f).leaderTerminals());
|
||||
}
|
||||
|
||||
@Test
|
||||
void neitherBlockLeavesNothingRegistered(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("none.yaml");
|
||||
@@ -528,25 +483,22 @@ class BridgedConfigTest {
|
||||
assertTrue(BridgedConfig.load(f).leaderTerminals().isEmpty());
|
||||
}
|
||||
|
||||
/**
|
||||
* CB-579: {@code fleet.leaders} no longer feeds {@code leaderTerminals()} at all — a lead's
|
||||
* identity comes from the live tab scan, not a config-held terminal map. This method now exists
|
||||
* only for the {@code primary.terminal} fallback.
|
||||
*/
|
||||
/** A lead entry with no terminal identifies nothing — it must not register a null key. */
|
||||
@Test
|
||||
void fleetLeadersNeverContributesToLeaderTerminals(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("leaders-only.yaml");
|
||||
void aLeadWithoutATerminalIsNotRegistered(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("no-terminal.yaml");
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
port: 8080
|
||||
fleet:
|
||||
leaders:
|
||||
opus-5.0:
|
||||
tab: "lead: opus-5.0"
|
||||
sketch:
|
||||
kind: opencode
|
||||
real:
|
||||
terminal: term_real
|
||||
""");
|
||||
|
||||
assertTrue(BridgedConfig.load(f).leaderTerminals().isEmpty(),
|
||||
"no primary.terminal pin ⇒ nothing registered, even with fleet.leaders configured");
|
||||
assertEquals(Map.of("term_real", "real"), BridgedConfig.load(f).leaderTerminals());
|
||||
}
|
||||
|
||||
// ── CB-548: the architects registry ────────────────────────────────────────────────────────
|
||||
|
||||
@@ -15,9 +15,8 @@ import java.util.concurrent.atomic.AtomicLong;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
/**
|
||||
* CB-531/CB-579. A lead is never spawned, so the daemon has to <em>find</em> it: these assert that
|
||||
* an operator-labelled tab matching a configured {@code tab:} is what makes a pane a lead, and —
|
||||
* just as importantly — what does not, and that a stale entry does not linger forever.
|
||||
* CB-531. A lead is never spawned, so the daemon has to <em>find</em> it: these assert that an
|
||||
* operator-labelled tab is what makes a pane a lead, and — just as importantly — what does not.
|
||||
*/
|
||||
class LeadTabScannerTest {
|
||||
|
||||
@@ -121,48 +120,23 @@ class LeadTabScannerTest {
|
||||
.pane("w9:p1", "w9:t1", "term_worker");
|
||||
}
|
||||
|
||||
/** The {@code tab:} → name map {@code twoLeads()}'s two lead tabs are configured under. */
|
||||
private static Map<String, String> twoLeadsConfigured() {
|
||||
return Map.of("lead: opus-5.0", "opus-5.0", "lead: gpt-sol-5.6", "gpt-sol-5.6");
|
||||
}
|
||||
|
||||
private LeadTabScanner scanner(TopologyHerdr herdr, Map<String, String> tabToName,
|
||||
private LeadTabScanner scanner(TopologyHerdr herdr, Map<String, String> configured,
|
||||
AtomicLong clock) {
|
||||
return new LeadTabScanner(herdr, tabToName, Set.of("bridged-workers"), TTL, clock::get);
|
||||
return new LeadTabScanner(herdr, "lead:", Set.of("bridged-workers"), configured, TTL,
|
||||
clock::get);
|
||||
}
|
||||
|
||||
@Test
|
||||
void everyConfiguredTabBecomesALeadNamedByItsEntry() {
|
||||
Map<String, String> leads = scanner(twoLeads(), twoLeadsConfigured(), new AtomicLong()).get();
|
||||
void everyLabelledTabBecomesALeadNamedByItsLabel() {
|
||||
Map<String, String> leads = scanner(twoLeads(), Map.of(), new AtomicLong()).get();
|
||||
|
||||
assertEquals(Map.of("term_opus", "opus-5.0", "term_gpt", "gpt-sol-5.6"), leads,
|
||||
"two leads discovered by their configured tab — no terminal_id was ever configured");
|
||||
"two leads discovered from labels alone — no terminal_id was ever configured");
|
||||
}
|
||||
|
||||
@Test
|
||||
void anUnconfiguredTabContributesNothing() {
|
||||
assertFalse(scanner(twoLeads(), twoLeadsConfigured(), new AtomicLong())
|
||||
.get().containsKey("term_notes"));
|
||||
}
|
||||
|
||||
/**
|
||||
* CB-579: matching is exact against the configured map now, not a shared prefix — two leads with
|
||||
* completely different labels are both discovered by one scanner, no convention required.
|
||||
*/
|
||||
@Test
|
||||
void twoLeadsWithCompletelyDifferentLabelsAreBothDiscovered() {
|
||||
TopologyHerdr herdr = new TopologyHerdr()
|
||||
.workspace("w1", "main")
|
||||
.tab("w1:t1", "w1", "orchestrator: opus")
|
||||
.tab("w1:t2", "w1", "captain: sol")
|
||||
.pane("w1:p1", "w1:t1", "term_opus")
|
||||
.pane("w1:p2", "w1:t2", "term_sol");
|
||||
Map<String, String> tabToName = Map.of("orchestrator: opus", "opus", "captain: sol", "sol");
|
||||
|
||||
Map<String, String> leads = scanner(herdr, tabToName, new AtomicLong()).get();
|
||||
|
||||
assertEquals(Map.of("term_opus", "opus", "term_sol", "sol"), leads,
|
||||
"no shared prefix needed — each lead is matched by its own configured tab");
|
||||
void anUnlabelledTabContributesNothing() {
|
||||
assertFalse(scanner(twoLeads(), Map.of(), new AtomicLong()).get().containsKey("term_notes"));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -174,28 +148,25 @@ class LeadTabScannerTest {
|
||||
void aTabInAWorkerSpaceIsNeverALeadEvenWhenItsLabelMatches() {
|
||||
TopologyHerdr herdr = twoLeads().tab("w9:t2", "w9", "lead: impostor")
|
||||
.pane("w9:p2", "w9:t2", "term_impostor");
|
||||
Map<String, String> tabToName = new LinkedHashMap<>(twoLeadsConfigured());
|
||||
tabToName.put("lead: impostor", "impostor");
|
||||
|
||||
assertFalse(scanner(herdr, tabToName, new AtomicLong()).get().containsKey("term_impostor"));
|
||||
assertFalse(scanner(herdr, Map.of(), new AtomicLong()).get().containsKey("term_impostor"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void aLabelWithNoConfiguredEntryIsIgnored() {
|
||||
void aBarePrefixNamesNobodyAndIsRejected() {
|
||||
TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
|
||||
.tab("w1:t1", "w1", "lead: nobody-configured").pane("w1:p1", "w1:t1", "term_a");
|
||||
.tab("w1:t1", "w1", "lead:").pane("w1:p1", "w1:t1", "term_a");
|
||||
|
||||
assertEquals(Map.of(), scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get(),
|
||||
"a label that names no configured lead resolves nobody");
|
||||
assertEquals(Map.of(), scanner(herdr, Map.of(), new AtomicLong()).get(),
|
||||
"a lead with no name would resolve as PRIMARY with nothing to attribute it to");
|
||||
}
|
||||
|
||||
@Test
|
||||
void matchingIsCaseInsensitiveAndToleratesSurroundingWhitespace() {
|
||||
void thePrefixMatchesCaseInsensitivelyAndTheNameIsTrimmed() {
|
||||
TopologyHerdr herdr = new TopologyHerdr().workspace("w1", "main")
|
||||
.tab("w1:t1", "w1", " LEAD: Opus-5.0 ").pane("w1:p1", "w1:t1", "term_a");
|
||||
.tab("w1:t1", "w1", " LEAD: opus-5.0 ").pane("w1:p1", "w1:t1", "term_a");
|
||||
|
||||
assertEquals(Map.of("term_a", "opus-5.0"),
|
||||
scanner(herdr, Map.of("lead: Opus-5.0", "opus-5.0"), new AtomicLong()).get());
|
||||
assertEquals(Map.of("term_a", "opus-5.0"), scanner(herdr, Map.of(), new AtomicLong()).get());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -204,51 +175,18 @@ class LeadTabScannerTest {
|
||||
// nothing bridged placed can land here (see the worker-space test above).
|
||||
TopologyHerdr herdr = twoLeads().pane("w1:p1b", "w1:t1", "term_opus_split");
|
||||
|
||||
assertEquals("opus-5.0",
|
||||
scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get().get("term_opus_split"));
|
||||
assertEquals("opus-5.0", scanner(herdr, Map.of(), new AtomicLong()).get().get("term_opus_split"));
|
||||
}
|
||||
|
||||
/**
|
||||
* CB-579 acceptance (6): this is the bug the ticket closes. A stale pin used to be merged back
|
||||
* over every scan and never expire; now a scan is the whole answer, so a lead whose tab is gone
|
||||
* drops out on the very next scan.
|
||||
*/
|
||||
@Test
|
||||
void aTabNoLongerPresentDropsTheLeadOnTheNextScan() {
|
||||
TopologyHerdr herdr = twoLeads();
|
||||
AtomicLong clock = new AtomicLong();
|
||||
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
|
||||
assertTrue(s.get().containsKey("term_opus"));
|
||||
void anExplicitlyConfiguredLeadIsMergedInAndOutranksALabel() {
|
||||
Map<String, String> configured = Map.of("term_opus", "pinned-name", "term_extra", "from-config");
|
||||
|
||||
// The session behind term_opus restarted — herdr no longer reports that tab or pane at all.
|
||||
herdr.tabs.remove("w1:t1");
|
||||
herdr.panes.remove("w1:p1");
|
||||
clock.addAndGet(TTL);
|
||||
Map<String, String> leads = scanner(twoLeads(), configured, new AtomicLong()).get();
|
||||
|
||||
assertFalse(s.get().containsKey("term_opus"),
|
||||
"a stale entry must expire once the tab it named is gone, not be merged back forever");
|
||||
}
|
||||
|
||||
/**
|
||||
* CB-579 acceptance (5): the whole point of matching by tab instead of {@code terminal_id} — a
|
||||
* restart changes the terminal, not the tab, so the lead resolves under the same name with no
|
||||
* config edit.
|
||||
*/
|
||||
@Test
|
||||
void aLeadRestartingInTheSameTabResolvesUnderTheSameName() {
|
||||
TopologyHerdr herdr = twoLeads();
|
||||
AtomicLong clock = new AtomicLong();
|
||||
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
|
||||
assertEquals("opus-5.0", s.get().get("term_opus"));
|
||||
|
||||
// The session restarts: herdr assigns the pane a new terminal_id, same tab (w1:t1).
|
||||
herdr.panes.remove("w1:p1");
|
||||
herdr.pane("w1:p1", "w1:t1", "term_opus_v2");
|
||||
clock.addAndGet(TTL);
|
||||
|
||||
Map<String, String> leads = s.get();
|
||||
assertEquals("opus-5.0", leads.get("term_opus_v2"), "the new terminal resolves immediately");
|
||||
assertFalse(leads.containsKey("term_opus"), "the old terminal_id is simply gone, not carried");
|
||||
assertEquals("pinned-name", leads.get("term_opus"), "an explicit pin is the operator's last word");
|
||||
assertEquals("from-config", leads.get("term_extra"), "a configured lead needs no tab at all");
|
||||
assertEquals("gpt-sol-5.6", leads.get("term_gpt"));
|
||||
}
|
||||
|
||||
// ── caching ─────────────────────────────────────────────────────────────────────────────────
|
||||
@@ -257,7 +195,7 @@ class LeadTabScannerTest {
|
||||
void aSecondLookupWithinTheTtlDoesNotTouchHerdr() {
|
||||
TopologyHerdr herdr = twoLeads();
|
||||
AtomicLong clock = new AtomicLong();
|
||||
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
|
||||
LeadTabScanner s = scanner(herdr, Map.of(), clock);
|
||||
|
||||
s.get();
|
||||
int afterFirst = herdr.calls;
|
||||
@@ -272,23 +210,21 @@ class LeadTabScannerTest {
|
||||
void aTabLabelledAfterStartupIsPickedUpOnceTheTtlExpires() {
|
||||
TopologyHerdr herdr = twoLeads();
|
||||
AtomicLong clock = new AtomicLong();
|
||||
Map<String, String> tabToName = new LinkedHashMap<>(twoLeadsConfigured());
|
||||
tabToName.put("lead: late-arrival", "late-arrival");
|
||||
LeadTabScanner s = scanner(herdr, tabToName, clock);
|
||||
LeadTabScanner s = scanner(herdr, Map.of(), clock);
|
||||
assertFalse(s.get().containsKey("term_notes"));
|
||||
|
||||
herdr.tab("w1:t3", "w1", "lead: late-arrival"); // the operator renames their tab
|
||||
clock.addAndGet(TTL);
|
||||
|
||||
assertEquals("late-arrival", s.get().get("term_notes"),
|
||||
"the whole point over a config-held terminal_id: no config edit, no restart");
|
||||
"the whole point over `leaders:`: no config edit, no restart");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aFailedScanKeepsTheLeadsAlreadyKnownRatherThanDemotingThem() {
|
||||
TopologyHerdr herdr = twoLeads();
|
||||
AtomicLong clock = new AtomicLong();
|
||||
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
|
||||
LeadTabScanner s = scanner(herdr, Map.of(), clock);
|
||||
Map<String, String> before = s.get();
|
||||
|
||||
herdr.failing = true;
|
||||
@@ -299,14 +235,14 @@ class LeadTabScannerTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
void aFailedFirstScanReturnsEmptyRatherThanThrowing() {
|
||||
void aFailedFirstScanStillHonoursTheConfiguredLeads() {
|
||||
TopologyHerdr herdr = twoLeads();
|
||||
herdr.failing = true;
|
||||
|
||||
Map<String, String> leads = scanner(herdr, twoLeadsConfigured(), new AtomicLong()).get();
|
||||
Map<String, String> leads = scanner(herdr, Map.of("term_x", "opus-5.0"), new AtomicLong()).get();
|
||||
|
||||
assertEquals(Map.of(), leads,
|
||||
"with nothing scanned yet and no override to fall back on, the map is simply empty");
|
||||
assertEquals(Map.of("term_x", "opus-5.0"), leads,
|
||||
"config-named leads must not depend on herdr answering at all");
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -314,7 +250,7 @@ class LeadTabScannerTest {
|
||||
TopologyHerdr herdr = twoLeads();
|
||||
herdr.failing = true;
|
||||
AtomicLong clock = new AtomicLong();
|
||||
LeadTabScanner s = scanner(herdr, twoLeadsConfigured(), clock);
|
||||
LeadTabScanner s = scanner(herdr, Map.of(), clock);
|
||||
|
||||
s.get();
|
||||
int afterFirst = herdr.calls;
|
||||
|
||||
@@ -7,8 +7,6 @@ import ch.qos.logback.core.read.ListAppender;
|
||||
import dev.ltms.bridged.herdr.AgentControl;
|
||||
import dev.ltms.bridged.herdr.FakeHerdr;
|
||||
import dev.ltms.bridged.msg.Rendezvous;
|
||||
import dev.ltms.bridged.msg.TestTurnTokens;
|
||||
import dev.ltms.bridged.msg.TurnToken;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
@@ -49,7 +47,7 @@ class CompletionResolverTest {
|
||||
Rendezvous rendezvous = new Rendezvous(); // no waiter opened
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
|
||||
|
||||
resolver.captureBaseline("term_a", TestTurnTokens.inert("term_a")); // no send to attribute a later completion to
|
||||
resolver.captureBaseline("term_a"); // no send to attribute a later completion to
|
||||
|
||||
assertFalse(herdr.called("agent.read"),
|
||||
"with no waiting send there is no turn to baseline — skip the scrape");
|
||||
@@ -196,7 +194,7 @@ class CompletionResolverTest {
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
|
||||
var waiter = rendezvous.open("term_a");
|
||||
resolver.captureBaseline("term_a", new TurnToken("term_a", waiter));
|
||||
resolver.captureBaseline("term_a");
|
||||
herdr.readText("⏺ answer that /clear would erase\n❯ ");
|
||||
|
||||
resolver.resolveBeforePostAction("term_a");
|
||||
@@ -219,7 +217,7 @@ class CompletionResolverTest {
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous);
|
||||
|
||||
var waiter = rendezvous.open("term_a"); // a send is blocked on this turn
|
||||
resolver.captureBaseline("term_a", new TurnToken("term_a", waiter)); // baseline is the clipped >cap block
|
||||
resolver.captureBaseline("term_a"); // baseline is the clipped >cap block
|
||||
var turn = resolver.inFlight("term_a");
|
||||
assertEquals(CompletionResolver.MAX_SCRAPE_CHARS, turn.baseline().length(),
|
||||
"the delivery baseline is clipped to the same cap resolve() applies to the tail");
|
||||
|
||||
@@ -8,7 +8,6 @@ import dev.ltms.bridged.herdr.AgentControl;
|
||||
import dev.ltms.bridged.herdr.AgentStatus;
|
||||
import dev.ltms.bridged.herdr.FakeHerdr;
|
||||
import dev.ltms.bridged.herdr.HerdrException;
|
||||
import dev.ltms.bridged.msg.TestTurnTokens;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
@@ -48,7 +47,7 @@ class InjectorTest {
|
||||
|
||||
@Test
|
||||
void deliversWhenIdle() {
|
||||
CompletableFuture<Void> f = injector.enqueue(T, "hello", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> f = injector.enqueue(T, "hello");
|
||||
assertFalse(f.isDone(), "not delivered until an injectable status arrives");
|
||||
injector.onStatus(T, AgentStatus.IDLE);
|
||||
assertTrue(f.isDone());
|
||||
@@ -60,7 +59,7 @@ class InjectorTest {
|
||||
// CB-113: idle alone is not enough — hold until the worker's MCP is connected (ready).
|
||||
java.util.Set<String> ready = new java.util.HashSet<>();
|
||||
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, ready::contains);
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE); // idle but not yet available → held out of the boot window
|
||||
assertEquals(List.of(), sent(), "must not deliver into a not-yet-available worker");
|
||||
@@ -80,7 +79,7 @@ class InjectorTest {
|
||||
void resubmitsEnterWhenADeliveredMessageIsNotPickedUp() {
|
||||
// CB-113: the Enter at delivery can race the paste; while the worker stays idle (not picked
|
||||
// up), the injector re-nudges Enter so the pending paste submits.
|
||||
injector.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "task");
|
||||
injector.onStatus(T, AgentStatus.IDLE); // deliver: paste + one Enter
|
||||
long afterDeliver = enterKeystrokes();
|
||||
|
||||
@@ -96,7 +95,7 @@ class InjectorTest {
|
||||
|
||||
@Test
|
||||
void holdsWhileWorkingThenDeliversOnIdle() {
|
||||
injector.enqueue(T, "later", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "later");
|
||||
injector.onStatus(T, AgentStatus.WORKING);
|
||||
assertEquals(List.of(), sent(), "must not inject mid-turn");
|
||||
injector.onStatus(T, AgentStatus.IDLE);
|
||||
@@ -105,7 +104,7 @@ class InjectorTest {
|
||||
|
||||
@Test
|
||||
void blockedIsInjectableButUnknownIsNot() {
|
||||
injector.enqueue(T, "answer", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "answer");
|
||||
injector.onStatus(T, AgentStatus.UNKNOWN);
|
||||
assertEquals(List.of(), sent(), "unknown status is not safe to inject");
|
||||
injector.onStatus(T, AgentStatus.BLOCKED);
|
||||
@@ -114,8 +113,8 @@ class InjectorTest {
|
||||
|
||||
@Test
|
||||
void twoRapidDeliveriesNeverInterleave() {
|
||||
injector.enqueue(T, "m1", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "m2", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "m1");
|
||||
injector.enqueue(T, "m2");
|
||||
|
||||
// First idle window delivers only m1, even if idle is observed twice before pickup.
|
||||
injector.onStatus(T, AgentStatus.IDLE);
|
||||
@@ -130,8 +129,8 @@ class InjectorTest {
|
||||
|
||||
@Test
|
||||
void transientUnknownDoesNotReleaseThePickupLatch() {
|
||||
injector.enqueue(T, "m1", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "m2", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "m1");
|
||||
injector.enqueue(T, "m2");
|
||||
injector.onStatus(T, AgentStatus.IDLE); // m1 sent, awaiting pickup
|
||||
assertEquals(List.of("m1"), sent());
|
||||
|
||||
@@ -146,8 +145,8 @@ class InjectorTest {
|
||||
|
||||
@Test
|
||||
void missedPickupEdgeIsReleasedByGraceSoTheQueueNeverWedges() {
|
||||
injector.enqueue(T, "m1", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "m2", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "m1");
|
||||
injector.enqueue(T, "m2");
|
||||
injector.onStatus(T, AgentStatus.IDLE); // m1 sent
|
||||
assertEquals(List.of("m1"), sent());
|
||||
|
||||
@@ -159,9 +158,9 @@ class InjectorTest {
|
||||
|
||||
@Test
|
||||
void fifoOrderAcrossManyTurns() {
|
||||
injector.enqueue(T, "a", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "b", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "c", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "a");
|
||||
injector.enqueue(T, "b");
|
||||
injector.enqueue(T, "c");
|
||||
for (int i = 0; i < 3; i++) {
|
||||
injector.onStatus(T, AgentStatus.IDLE); // deliver one
|
||||
injector.onStatus(T, AgentStatus.WORKING); // pickup
|
||||
@@ -173,7 +172,7 @@ class InjectorTest {
|
||||
@Test
|
||||
void activeWhileQueuedOrInFlightThenQuietAfterTurnCompletes() {
|
||||
assertTrue(injector.activeTargets().isEmpty());
|
||||
injector.enqueue(T, "x", TestTurnTokens.inert(T));
|
||||
injector.enqueue(T, "x");
|
||||
assertEquals(Set.of(T), injector.activeTargets(), "active while a message is queued");
|
||||
|
||||
injector.onStatus(T, AgentStatus.IDLE); // delivers; awaiting pickup
|
||||
@@ -192,7 +191,7 @@ class InjectorTest {
|
||||
void firesTurnCompleteOnAConfirmedWorkingThenIdle() {
|
||||
List<String> completed = new ArrayList<>();
|
||||
Injector inj = new Injector(new AgentControl(herdr), completed::add);
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE); // deliver
|
||||
inj.onStatus(T, AgentStatus.WORKING); // pickup + turn running
|
||||
@@ -227,8 +226,8 @@ class InjectorTest {
|
||||
}
|
||||
ResetListener listener = new ResetListener();
|
||||
Injector inj = new Injector(agents, listener);
|
||||
inj.enqueue(T, "first", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "second", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "first");
|
||||
inj.enqueue(T, "second");
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE); // first delegation
|
||||
inj.onStatus(T, AgentStatus.WORKING);
|
||||
@@ -247,7 +246,7 @@ class InjectorTest {
|
||||
void doesNotSynthesizeCompletionFromAnUnconfirmedTurn() {
|
||||
List<String> completed = new ArrayList<>();
|
||||
Injector inj = new Injector(new AgentControl(herdr), completed::add);
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
|
||||
// Deliver, then only ever idle — a `working` sample is never seen. The pickup grace unwedges
|
||||
// the queue but must NOT invent a completion: without a sampled turn there is no trustworthy
|
||||
@@ -286,7 +285,7 @@ class InjectorTest {
|
||||
void failsAnOutstandingDelegationWhoseWorkerWedgesInUnknown() {
|
||||
Captor cap = new Captor();
|
||||
Injector inj = new Injector(new AgentControl(herdr), cap);
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE); // deliver
|
||||
inj.onStatus(T, AgentStatus.WORKING); // worker starts the turn
|
||||
@@ -301,7 +300,7 @@ class InjectorTest {
|
||||
void aTransientUnknownGlitchNeitherFailsNorBlocksCompletion() {
|
||||
Captor cap = new Captor();
|
||||
Injector inj = new Injector(new AgentControl(herdr), cap);
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE); // deliver
|
||||
inj.onStatus(T, AgentStatus.WORKING); // confirmed turn
|
||||
@@ -316,7 +315,7 @@ class InjectorTest {
|
||||
void sendFailureDropsMessageAndFailsItsFuture() {
|
||||
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
|
||||
Injector inj = new Injector(new AgentControl(failing));
|
||||
CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> f = inj.enqueue(T, "boom");
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE);
|
||||
assertTrue(f.isCompletedExceptionally());
|
||||
@@ -325,7 +324,7 @@ class InjectorTest {
|
||||
|
||||
@Test
|
||||
void dropFailsPendingWaiters() {
|
||||
CompletableFuture<Void> f = injector.enqueue(T, "orphan", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> f = injector.enqueue(T, "orphan");
|
||||
injector.drop(T, new HerdrException("worker gone", "pane_not_found", null));
|
||||
assertTrue(f.isCompletedExceptionally(), "queued waiters unblock when the worker vanishes");
|
||||
}
|
||||
@@ -334,8 +333,8 @@ class InjectorTest {
|
||||
void dropPassesTheRealCauseForQueuedAndDeliveredWork() {
|
||||
Captor cap = new Captor();
|
||||
Injector inj = new Injector(new AgentControl(herdr), cap);
|
||||
CompletableFuture<Void> delivered = inj.enqueue(T, "delivered", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> queued = inj.enqueue(T, "queued", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> delivered = inj.enqueue(T, "delivered");
|
||||
CompletableFuture<Void> queued = inj.enqueue(T, "queued");
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE); // deliver the first message
|
||||
inj.onStatus(T, AgentStatus.WORKING); // its turn is now in flight; one remains queued
|
||||
@@ -353,7 +352,7 @@ class InjectorTest {
|
||||
// leave its send hanging. A vanished worker must fail that in-flight turn too.
|
||||
Captor cap = new Captor();
|
||||
Injector inj = new Injector(new AgentControl(herdr), cap);
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
inj.onStatus(T, AgentStatus.IDLE); // deliver
|
||||
inj.onStatus(T, AgentStatus.WORKING); // turn running
|
||||
|
||||
@@ -368,7 +367,7 @@ class InjectorTest {
|
||||
// by an off-sub worker's review of CB-110, delegated through the bridge.)
|
||||
Captor cap = new Captor();
|
||||
Injector inj = new Injector(new AgentControl(herdr), cap);
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
inj.onStatus(T, AgentStatus.IDLE); // deliver; pickup never confirmed
|
||||
|
||||
inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
|
||||
@@ -387,7 +386,7 @@ class InjectorTest {
|
||||
Captor cap = new Captor();
|
||||
List<String> forgotten = new ArrayList<>();
|
||||
Injector inj = new Injector(new AgentControl(herdr), cap, _ -> false, forgotten::add);
|
||||
CompletableFuture<Void> f = inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> f = inj.enqueue(T, "task");
|
||||
|
||||
for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);
|
||||
|
||||
@@ -415,7 +414,7 @@ class InjectorTest {
|
||||
try {
|
||||
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> false, _ -> {
|
||||
});
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
|
||||
for (int i = 0; i < READINESS_SAMPLES; i++) inj.onStatus(T, AgentStatus.IDLE);
|
||||
|
||||
@@ -440,7 +439,7 @@ class InjectorTest {
|
||||
Set<String> ready = new java.util.HashSet<>();
|
||||
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, ready::contains, _ -> {
|
||||
});
|
||||
inj.enqueue(T, "task", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "task");
|
||||
|
||||
for (int i = 0; i < 100; i++) inj.onStatus(T, AgentStatus.IDLE); // still booting, well under grace
|
||||
assertEquals(List.of(), sent());
|
||||
@@ -456,7 +455,7 @@ class InjectorTest {
|
||||
// linger past the worker's life (MemberPresence.forget had no caller before this).
|
||||
List<String> forgotten = new ArrayList<>();
|
||||
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> true, forgotten::add);
|
||||
inj.enqueue(T, "orphan", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "orphan");
|
||||
inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
|
||||
assertEquals(List.of(T), forgotten, "drop clears the gone worker's presence");
|
||||
}
|
||||
@@ -477,7 +476,7 @@ class InjectorTest {
|
||||
try {
|
||||
Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> true, _ -> {
|
||||
});
|
||||
inj.enqueue(T, "orphan", TestTurnTokens.inert(T));
|
||||
inj.enqueue(T, "orphan");
|
||||
inj.drop(T, new HerdrException("worker gone", "pane_not_found", null));
|
||||
|
||||
String warn = appender.list.stream()
|
||||
@@ -501,7 +500,7 @@ class InjectorTest {
|
||||
StatusPoller poller = new StatusPoller(new AgentControl(idle), inj, 10);
|
||||
poller.start();
|
||||
try {
|
||||
CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> delivered = inj.enqueue(T, "via-poller");
|
||||
delivered.get(2, TimeUnit.SECONDS); // completes when the poller drives the send
|
||||
} finally {
|
||||
poller.stop();
|
||||
@@ -520,7 +519,7 @@ class InjectorTest {
|
||||
void deliveredFutureCarriesSendFailure() {
|
||||
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
|
||||
Injector inj = new Injector(new AgentControl(failing));
|
||||
CompletableFuture<Void> f = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> f = inj.enqueue(T, "boom");
|
||||
inj.onStatus(T, AgentStatus.IDLE);
|
||||
ExecutionException ex = assertThrows(ExecutionException.class, f::get);
|
||||
assertInstanceOf(HerdrException.class, ex.getCause());
|
||||
|
||||
@@ -41,8 +41,8 @@ class LeadLauncherTest {
|
||||
null, null, fleet, null, "fixed", null).withDefaults();
|
||||
}
|
||||
|
||||
private static BridgedConfig.Leader lead(String profile, String tab, int instances) {
|
||||
return new BridgedConfig.Leader(profile, tab, instances, "lead:", 10, null, null,
|
||||
private static BridgedConfig.Leader lead(String profile, String terminal, int instances) {
|
||||
return new BridgedConfig.Leader(profile, terminal, instances, "lead:", 10, null, null,
|
||||
"leads", "/repo");
|
||||
}
|
||||
|
||||
@@ -70,16 +70,16 @@ class LeadLauncherTest {
|
||||
void startsTheDeclaredLeadWhenNoneIsRunning() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
|
||||
assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
|
||||
assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
|
||||
assertTrue(herdr.called("agent.start"), "a lead must actually be started");
|
||||
assertEquals("lead-opus", startedName(herdr));
|
||||
}
|
||||
|
||||
/** The tab is labelled with the configured `tab:` so the scanner finds the lead on the next resolve. */
|
||||
/** The tab is labelled so the scanner finds the lead on the next resolve. */
|
||||
@Test
|
||||
void labelsTheTabWithTheConfiguredTabValue() {
|
||||
void labelsTheTabWithThePrefixTheScannerReadsBack() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
|
||||
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
|
||||
|
||||
assertEquals("lead: opus",
|
||||
((Map<?, ?>) herdr.lastCall("tab.rename").params()).get("label"));
|
||||
@@ -90,7 +90,7 @@ class LeadLauncherTest {
|
||||
void startsAsManyInstancesAsAreDeclared() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
|
||||
assertEquals(2, launcher(herdr, configWith(lead("opus", "lead: opus", 2))).ensureLeads());
|
||||
assertEquals(2, launcher(herdr, configWith(lead("opus", null, 2))).ensureLeads());
|
||||
assertEquals(2, herdr.calls.stream().filter(c -> c.method().equals("agent.start")).count());
|
||||
}
|
||||
|
||||
@@ -104,7 +104,7 @@ class LeadLauncherTest {
|
||||
.withTab("wL", "wL:t1", "lead: opus")
|
||||
.withAgent("lead-opus", "term_lead", "wL:p1", "wL:t1");
|
||||
|
||||
assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
|
||||
assertEquals(0, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
|
||||
assertFalse(herdr.called("agent.start"), "the live lead must not be duplicated");
|
||||
}
|
||||
|
||||
@@ -118,23 +118,20 @@ class LeadLauncherTest {
|
||||
.withWorkspace("wL", "leads")
|
||||
.withTab("wL", "wL:t1", "lead: opus"); // label only — nothing running in it
|
||||
|
||||
assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads(),
|
||||
assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads(),
|
||||
"a stale label is not a lead; the lead must be relaunched");
|
||||
}
|
||||
|
||||
/**
|
||||
* A lead the operator opened by hand is live once its tab carries the configured `tab:` label —
|
||||
* CB-579 retired the `terminal:` pin, so a hand-opened lead is found the same way an
|
||||
* auto-launched one is, by its tab, not by a terminal id nobody wrote down in advance.
|
||||
* A lead the operator opened by hand and pinned with `terminal:` is live even though its tab
|
||||
* carries no matching label. Counting labels alone would relaunch it on every boot.
|
||||
*/
|
||||
@Test
|
||||
void aHandOpenedLeadWithTheConfiguredTabLabelCountsAsLive() {
|
||||
void aPinnedTerminalWithARunningAgentCountsAsLive() {
|
||||
FakeHerdr herdr = new FakeHerdr()
|
||||
.withWorkspace("wX", "main")
|
||||
.withTab("wX", "wX:t1", "lead: opus")
|
||||
.withAgent("hand-opened", "term_hand", "wX:p1", "wX:t1");
|
||||
.withAgent("hand-opened", "term_pinned", "wX:p1", "wX:t1");
|
||||
|
||||
assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
|
||||
assertEquals(0, launcher(herdr, configWith(lead("opus", "term_pinned", 1))).ensureLeads());
|
||||
assertFalse(herdr.called("agent.start"));
|
||||
}
|
||||
|
||||
@@ -146,7 +143,7 @@ class LeadLauncherTest {
|
||||
.withTab("wM", "wM:t1", "lead: opus") // a member tab that looks like a lead
|
||||
.withAgent("claude-opus-x", "term_m", "wM:p1", "wM:t1");
|
||||
|
||||
assertEquals(1, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads(),
|
||||
assertEquals(1, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads(),
|
||||
"a member in a lead-labelled tab is not a lead, so the real lead is still missing");
|
||||
}
|
||||
|
||||
@@ -155,7 +152,7 @@ class LeadLauncherTest {
|
||||
void anUncountableHerdrStartsNothing() {
|
||||
FakeHerdr herdr = new FakeHerdr().healthy(false);
|
||||
|
||||
assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads());
|
||||
assertEquals(0, launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads());
|
||||
assertFalse(herdr.called("agent.start"));
|
||||
}
|
||||
|
||||
@@ -169,7 +166,7 @@ class LeadLauncherTest {
|
||||
@Test
|
||||
void theLeadNeverReceivesTheWorkerReplyCharter() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
|
||||
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
|
||||
|
||||
List<String> args = startedArgs(herdr);
|
||||
assertFalse(args.contains("--append-system-prompt"),
|
||||
@@ -181,7 +178,7 @@ class LeadLauncherTest {
|
||||
@Test
|
||||
void theLeadMountsTheBridgeMcpAndPinsItsModel() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
|
||||
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
|
||||
|
||||
List<String> args = startedArgs(herdr);
|
||||
assertTrue(args.contains("--mcp-config"));
|
||||
@@ -195,7 +192,7 @@ class LeadLauncherTest {
|
||||
@Test
|
||||
void theLeadEnvCarriesNoAnthropicBinding() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
|
||||
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
|
||||
|
||||
Map<String, String> env = tabEnv(herdr);
|
||||
assertNull(env.get("ANTHROPIC_BASE_URL"));
|
||||
@@ -208,7 +205,7 @@ class LeadLauncherTest {
|
||||
@Test
|
||||
void theLeadTabIsCreatedOutsideEveryMemberWorkspace() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
launcher(herdr, configWith(lead("opus", "lead: opus", 1))).ensureLeads();
|
||||
launcher(herdr, configWith(lead("opus", null, 1))).ensureLeads();
|
||||
|
||||
String label = (String) ((Map<?, ?>) herdr.lastCall("workspace.create").params()).get("label");
|
||||
assertEquals("leads", label);
|
||||
@@ -217,12 +214,12 @@ class LeadLauncherTest {
|
||||
|
||||
// ── recognise-only and misconfiguration ───────────────────────────────────────────────────
|
||||
|
||||
/** A lead with a tab but no profile is recognise-only by design — not an error, not a launch. */
|
||||
/** A lead with a pin but no profile is recognise-only by design — not an error, not a launch. */
|
||||
@Test
|
||||
void aLeadThatNamesNoProfileIsRecognisedButNeverLaunched() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
|
||||
assertEquals(0, launcher(herdr, configWith(lead(null, "lead: dead", 1))).ensureLeads());
|
||||
assertEquals(0, launcher(herdr, configWith(lead(null, "term_dead", 1))).ensureLeads());
|
||||
assertFalse(herdr.called("agent.start"));
|
||||
}
|
||||
|
||||
@@ -231,7 +228,7 @@ class LeadLauncherTest {
|
||||
void zeroInstancesLaunchesNothing() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
|
||||
assertEquals(0, launcher(herdr, configWith(lead("opus", "lead: opus", 0))).ensureLeads());
|
||||
assertEquals(0, launcher(herdr, configWith(lead("opus", null, 0))).ensureLeads());
|
||||
assertFalse(herdr.called("agent.start"));
|
||||
}
|
||||
|
||||
@@ -240,7 +237,7 @@ class LeadLauncherTest {
|
||||
void anUnknownProfileIsSkippedRatherThanThrown() {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
|
||||
assertEquals(0, launcher(herdr, configWith(lead("nope", "lead: opus", 1))).ensureLeads());
|
||||
assertEquals(0, launcher(herdr, configWith(lead("nope", null, 1))).ensureLeads());
|
||||
assertFalse(herdr.called("agent.start"));
|
||||
}
|
||||
|
||||
|
||||
@@ -127,7 +127,7 @@ class MessageServiceTest {
|
||||
injector.onStatus(T, AgentStatus.IDLE); // first delivery
|
||||
injector.onStatus(T, AgentStatus.WORKING); // first turn in flight
|
||||
|
||||
CompletableFuture<Void> queued = injector.enqueue(T, "second task", TestTurnTokens.inert(T));
|
||||
CompletableFuture<Void> queued = injector.enqueue(T, "second task");
|
||||
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(T);
|
||||
injector.drop(T, new HerdrException("agent target sol not found", "agent_not_found", null));
|
||||
|
||||
@@ -676,41 +676,6 @@ class MessageServiceTest {
|
||||
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
|
||||
}
|
||||
|
||||
@Test
|
||||
void unansweredAsyncQuestionReturnsTheTicketToPendingAndReleasesItsTarget() throws Exception {
|
||||
String ticket = messages.sendAsync(T, "task that asks");
|
||||
awaitWaiting();
|
||||
injectDelivery();
|
||||
|
||||
assertEquals(MessageService.AskOutcome.TIMED_OUT,
|
||||
messages.ask(T, "which config?", 200).outcome());
|
||||
assertEquals(MessageService.Phase.PENDING, messages.poll(ticket).phase(),
|
||||
"only the question wait ended; the delegated turn may still finish");
|
||||
|
||||
String next = messages.sendAsync(T, "next task");
|
||||
awaitWaiting();
|
||||
assertTrue(rendezvous.resolve(T, "done"));
|
||||
assertEquals(MessageService.Phase.DONE, awaitTicketPhase(next, MessageService.Phase.DONE).phase());
|
||||
}
|
||||
|
||||
@Test
|
||||
void asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter() throws Exception {
|
||||
String first = messages.sendAsync(T, "first task");
|
||||
awaitWaiting();
|
||||
|
||||
CompletableFuture<MessageService.AskResult> ask =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
|
||||
assertEquals(MessageService.Phase.ASKING, awaitTicketPhase(first, MessageService.Phase.ASKING).phase());
|
||||
|
||||
MessageService.TaskView asking = messages.poll(first);
|
||||
CompletableFuture<MessageService.Reply> answer = CompletableFuture.supplyAsync(
|
||||
() -> messages.answer(asking.turnId(), "config.yaml", 5000));
|
||||
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
|
||||
awaitWaiting();
|
||||
assertTrue(rendezvous.resolve(T, "done"));
|
||||
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome());
|
||||
}
|
||||
|
||||
private void assertFailedTicket(String ticket, String reason) throws Exception {
|
||||
MessageService.TaskView view = awaitTicketPhase(ticket, MessageService.Phase.FAILED);
|
||||
assertEquals(reason, view.detail());
|
||||
|
||||
@@ -1,20 +0,0 @@
|
||||
package dev.ltms.bridged.msg;
|
||||
|
||||
/**
|
||||
* Explicit unbound tokens for tests that exercise delivery without an accepted send.
|
||||
*
|
||||
* <p>The waiter is {@code null} on purpose. "No accepted send" is an <em>absence</em>, and a helper
|
||||
* that handed back a fresh {@code CompletableFuture} would invent one — which is how the first
|
||||
* version of this class turned {@code captureBaselineSkipsTheReadWhenNoSendIsWaiting} red: the
|
||||
* resolver saw a non-null waiter, decided a turn was in flight, and scraped a pane that no send was
|
||||
* blocked on. An inert value must omit the fact, never fabricate it.
|
||||
*/
|
||||
public final class TestTurnTokens {
|
||||
private TestTurnTokens() {
|
||||
}
|
||||
|
||||
/** A token for a delivery that no send is waiting on: it authorises nothing. */
|
||||
public static TurnToken inert(String target) {
|
||||
return new TurnToken(target, null);
|
||||
}
|
||||
}
|
||||
@@ -10,7 +10,6 @@ import dev.ltms.bridged.herdr.AgentControl;
|
||||
import dev.ltms.bridged.herdr.FakeHerdr;
|
||||
import dev.ltms.bridged.herdr.WorkspaceControl;
|
||||
import dev.ltms.bridged.member.ClaudeCodeLauncher;
|
||||
import dev.ltms.bridged.msg.TestTurnTokens;
|
||||
import dev.ltms.bridged.peer.PeerUnreachableException;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.slf4j.LoggerFactory;
|
||||
@@ -102,7 +101,7 @@ class SessionManagerTest {
|
||||
|
||||
assertDoesNotThrow(() -> sessions.asPresence().markPresent(null),
|
||||
"the primary's null terminal must not blow up an unrelated tool call");
|
||||
assertDoesNotThrow(() -> sessions.onDelivered(null, TestTurnTokens.inert(null)));
|
||||
assertDoesNotThrow(() -> sessions.onDelivered(null));
|
||||
assertDoesNotThrow(() -> sessions.onTurnComplete(null));
|
||||
assertDoesNotThrow(() -> sessions.onTurnFailed(null));
|
||||
|
||||
@@ -122,7 +121,7 @@ class SessionManagerTest {
|
||||
"MCP presence moves SPAWNING → READY");
|
||||
assertTrue(sessions.asPresence().isPresent(terminal), "presence is also recorded");
|
||||
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
assertEquals(MemberSession.State.BUSY, sessions.get(session.paneId()).orElseThrow().state(),
|
||||
"delivery moves READY → BUSY");
|
||||
|
||||
@@ -154,7 +153,7 @@ class SessionManagerTest {
|
||||
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
|
||||
String terminal = session.terminalId();
|
||||
sessions.asPresence().markPresent(terminal);
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
|
||||
sessions.onTurnFailed(terminal);
|
||||
|
||||
@@ -182,7 +181,7 @@ class SessionManagerTest {
|
||||
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
|
||||
String terminal = session.terminalId();
|
||||
sessions.asPresence().markPresent(terminal);
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
|
||||
sessions.onTurnFailed(terminal);
|
||||
|
||||
@@ -265,7 +264,7 @@ class SessionManagerTest {
|
||||
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
|
||||
String terminal = session.terminalId();
|
||||
sessions.asPresence().markPresent(terminal);
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
|
||||
clock[0] = 100;
|
||||
assertEquals(0, sessions.reapIdle(10), "BUSY session past TTL is never reaped");
|
||||
@@ -282,7 +281,7 @@ class SessionManagerTest {
|
||||
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
|
||||
String terminal = session.terminalId();
|
||||
sessions.asPresence().markPresent(terminal);
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
sessions.onTurnComplete(terminal);
|
||||
|
||||
clock[0] = 21;
|
||||
@@ -300,7 +299,7 @@ class SessionManagerTest {
|
||||
MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "owner2");
|
||||
sessions.asPresence().markPresent(ready.terminalId());
|
||||
sessions.asPresence().markPresent(busy.terminalId());
|
||||
sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));
|
||||
sessions.onDelivered(busy.terminalId());
|
||||
|
||||
clock[0] = 50;
|
||||
assertEquals(1, sessions.reapIdle(30), "only READY past TTL is reaped");
|
||||
@@ -318,9 +317,9 @@ class SessionManagerTest {
|
||||
String terminal = session.terminalId();
|
||||
sessions.asPresence().markPresent(terminal);
|
||||
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
sessions.onTurnComplete(terminal);
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
sessions.onTurnComplete(terminal);
|
||||
|
||||
MemberSession updated = sessions.get(session.paneId()).orElseThrow();
|
||||
@@ -338,13 +337,13 @@ class SessionManagerTest {
|
||||
String terminal = session.terminalId();
|
||||
sessions.asPresence().markPresent(terminal);
|
||||
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
sessions.onTurnComplete(terminal);
|
||||
assertEquals(MemberSession.State.DONE,
|
||||
sessions.get(session.paneId()).orElseThrow().state(),
|
||||
"first turn completes without release");
|
||||
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal));
|
||||
sessions.onDelivered(terminal);
|
||||
sessions.onTurnComplete(terminal);
|
||||
|
||||
assertTrue(sessions.get(session.paneId()).isEmpty(), "session released after cap reached");
|
||||
@@ -360,7 +359,7 @@ class SessionManagerTest {
|
||||
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
|
||||
sessions.asPresence().markPresent(session.terminalId());
|
||||
|
||||
sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
|
||||
sessions.onDelivered(session.terminalId());
|
||||
assertTrue(sessions.onTurnCompleteWithPostAction(session.terminalId()));
|
||||
|
||||
MemberSession updated = sessions.get(session.paneId()).orElseThrow();
|
||||
@@ -374,7 +373,7 @@ class SessionManagerTest {
|
||||
SessionManager sessions = sessionManager(herdr, () -> 0L, 1, true);
|
||||
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
|
||||
sessions.asPresence().markPresent(session.terminalId());
|
||||
sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
|
||||
sessions.onDelivered(session.terminalId());
|
||||
|
||||
assertFalse(sessions.hasPostTurnAction(session.terminalId()),
|
||||
"a session at its cap will be released, not reset for reuse");
|
||||
@@ -389,7 +388,7 @@ class SessionManagerTest {
|
||||
SessionManager sessions = sessionManager(herdr, () -> 0L, 0, false);
|
||||
MemberSession session = sessions.acquire("ltms-local", null, "/caller", "term_primary");
|
||||
sessions.asPresence().markPresent(session.terminalId());
|
||||
sessions.onDelivered(session.terminalId(), TestTurnTokens.inert(session.terminalId()));
|
||||
sessions.onDelivered(session.terminalId());
|
||||
|
||||
sessions.onTurnComplete(session.terminalId());
|
||||
|
||||
@@ -407,7 +406,7 @@ class SessionManagerTest {
|
||||
MemberSession busy = sessions.acquire("ltms-local", "/busy", "/caller", "ownerB");
|
||||
sessions.asPresence().markPresent(ready.terminalId());
|
||||
sessions.asPresence().markPresent(busy.terminalId());
|
||||
sessions.onDelivered(busy.terminalId(), TestTurnTokens.inert(busy.terminalId()));
|
||||
sessions.onDelivered(busy.terminalId());
|
||||
|
||||
sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
|
||||
|
||||
|
||||
@@ -7,7 +7,6 @@ import dev.ltms.bridged.herdr.AgentControl;
|
||||
import dev.ltms.bridged.herdr.FakeHerdr;
|
||||
import dev.ltms.bridged.herdr.WorkspaceControl;
|
||||
import dev.ltms.bridged.member.ClaudeCodeLauncher;
|
||||
import dev.ltms.bridged.msg.TestTurnTokens;
|
||||
import dev.ltms.bridged.peer.MemberRole;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
@@ -205,7 +204,7 @@ class WorktreeSessionManagerTest {
|
||||
new WorktreeRequest("cb-544", null));
|
||||
String terminal = s.terminalId();
|
||||
sessions.asPresence().markPresent(terminal);
|
||||
sessions.onDelivered(terminal, TestTurnTokens.inert(terminal)); // BUSY, never completes → still BUSY when the timeout hits
|
||||
sessions.onDelivered(terminal); // BUSY, never completes → still BUSY when the timeout hits
|
||||
|
||||
sessions.drainAll(TimeUnit.MILLISECONDS.toNanos(100));
|
||||
|
||||
|
||||
+2
-22
@@ -744,28 +744,8 @@ worktree discovery.
|
||||
|
||||
Acceptance criteria:
|
||||
|
||||
1. Every accepted send receives a stable `TurnToken` tied to target, exact waiter, and delivery
|
||||
baseline.
|
||||
|
||||
**Corrected during implementation (2026-08-15).** This criterion first also required the session
|
||||
turn number and the task outcome. That is not implementable at this layer, and the implementer
|
||||
refused it three times rather than fabricate a value — correctly. The reason is an ordering fact
|
||||
that is invisible from any single class: `MessageService` owns acceptance and holds the waiter and
|
||||
the async `Task`, but it learns nothing about delivery, because the delivery event goes to
|
||||
`CompletionResolver` through `TurnListener.onDelivered`. And `CompletionResolver.onDelivered` runs
|
||||
*before* `SessionManager.onDelivered`, so the session turn number does not exist yet at the only
|
||||
point where the token could capture it.
|
||||
|
||||
Two ways out were rejected. A shared registry keyed by target reintroduces exactly the "whichever
|
||||
send happens to be waiting" ambiguity the token exists to remove — the same weak claim
|
||||
`Rendezvous.currentWaiter` warns about. Injecting a turn counter into `MessageService` adds a
|
||||
required cross-layer dependency to populate a field that nothing in this slice reads, which is
|
||||
speculative coupling across a boundary already shown to be fragile.
|
||||
|
||||
So the token identifies the **accepted send**, and `SessionManager` keeps verifying its own
|
||||
delivery separately. Repair (criterion 2) does need the session turn; binding it means resolving
|
||||
that acceptance-versus-delivery ordering first, and that work belongs to the repair unit, not
|
||||
here. The token record carries a comment saying the field is deliberately absent.
|
||||
1. Every accepted send receives a stable `TurnToken` tied to target, exact waiter, session turn,
|
||||
delivery baseline, and task outcome.
|
||||
2. Repair requires the same `BUSY` token, two raw `IDLE` or `DONE` snapshots, no conflicting
|
||||
observation, exact open waiter, successful baseline, and new recognised assistant output.
|
||||
3. Missing, failed, late, or post-restart baseline never authorises repair.
|
||||
|
||||
Reference in New Issue
Block a user