Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 31b028e860 | |||
| c935b181dd | |||
| c325054242 | |||
| 4877992a70 | |||
| 0f08b93659 | |||
| 432c1d92d1 | |||
| 60b7e67b42 | |||
| 3fbd43fe3f |
@@ -174,7 +174,13 @@ public final class Fleetd {
|
||||
// genuine cycle. Break it exactly like liveCountRef below: a forwarding sink built now,
|
||||
// pointed at the real one once it exists.
|
||||
AtomicReference<ExhaustionSink> exhaustionSinkRef = new AtomicReference<>(ExhaustionSink.none());
|
||||
ExhaustionSink forwardingExhaustionSink = (target, reason) -> exhaustionSinkRef.get().onExhausted(target, reason);
|
||||
// fleetd #234, round 4: routed through the shared ExhaustionSink.forwardingTo factory
|
||||
// rather than written inline here — not because a lambda at this call site is unsafe
|
||||
// anymore (it is not: the 3-arg overload is now the interface's single abstract method, so
|
||||
// there is no 2-arg overload left for any lambda to silently bind to instead), but so a
|
||||
// test can call the exact same object this line builds, instead of asserting a copy of its
|
||||
// shape (round 3's lesson).
|
||||
ExhaustionSink forwardingExhaustionSink = ExhaustionSink.forwardingTo(exhaustionSinkRef::get);
|
||||
// The claude-code adapter is the always-present default; keep it even with no profiles (so a
|
||||
// bridge configured with no workers, or opencode-only, still has a well-defined base adapter)
|
||||
// unless opencode is the only kind configured.
|
||||
@@ -345,17 +351,41 @@ public final class Fleetd {
|
||||
// model-mismatch check — it needs nothing profile-specific from the caller beyond `target`
|
||||
// (a herdr terminal id) and `reason`, so reusing it here is exactly "the existing
|
||||
// ExhaustionSink path", not a new mechanism.
|
||||
ExhaustionSink exhaustionSink = (target, reason) -> sessions.roster().stream()
|
||||
.filter(session -> target.equals(session.terminalId()))
|
||||
.findFirst()
|
||||
.map(MemberSession::profile)
|
||||
.map(profileName -> config.get().profiles().get(profileName))
|
||||
.ifPresent(profile -> {
|
||||
String credentialId = profile.effectiveCredentialId();
|
||||
quarantine.quarantine(credentialId);
|
||||
log.warn("credential '{}' quarantined for {}s (profile '{}'): {}", credentialId,
|
||||
cfg.quarantineCooldownSeconds(), profile.profile(), reason);
|
||||
});
|
||||
//
|
||||
// fleetd #234: that check fires from SessionAwareHandle.agentSessionId(), which runs during
|
||||
// SessionManager.acquire() BEFORE this session is registered in sessions.roster() — so the
|
||||
// roster-only lookup below used to find nothing, .ifPresent silently no-op'd, and the
|
||||
// ERROR the check had just logged ("quarantining this profile's credential") was a lie:
|
||||
// nothing was quarantined, and nothing said so. Two changes: (1) OpenCodeLauncher now
|
||||
// passes its OWN profile name via ExhaustionSink's 3-arg overload — it already has the
|
||||
// FleetConfig.Profile in hand and does not need the roster at all — used here as a
|
||||
// fallback whenever the roster lookup misses; (2) if a profile still cannot be resolved
|
||||
// (neither the roster nor the hint names a configured one), this logs loudly at ERROR
|
||||
// instead of silently doing nothing — a control that cannot act must say so.
|
||||
// fleetd #234, round 4: the 3-arg overload is now ExhaustionSink's single abstract method,
|
||||
// so this is safely a lambda — there is no separate 2-arg overload left for it to bind to
|
||||
// instead and silently drop profileHint (that was rounds 1-3's whole hazard).
|
||||
ExhaustionSink exhaustionSink = (target, reason, profileHint) -> {
|
||||
String profileName = sessions.roster().stream()
|
||||
.filter(session -> target.equals(session.terminalId()))
|
||||
.findFirst()
|
||||
.map(MemberSession::profile)
|
||||
.orElse(profileHint);
|
||||
FleetConfig.Profile profile = profileName == null ? null : config.get().profiles().get(profileName);
|
||||
if (profile == null) {
|
||||
log.error("quarantine requested for target '{}' ({}) but no profile could be "
|
||||
+ "resolved — the target is not (yet) in the roster, and {} — "
|
||||
+ "credential NOT quarantined (fleetd #234)",
|
||||
target, reason,
|
||||
profileHint == null ? "no profile hint was given"
|
||||
: "the hinted profile '" + profileHint + "' is not configured");
|
||||
return;
|
||||
}
|
||||
String credentialId = profile.effectiveCredentialId();
|
||||
quarantine.quarantine(credentialId);
|
||||
log.warn("credential '{}' quarantined for {}s (profile '{}'): {}", credentialId,
|
||||
cfg.quarantineCooldownSeconds(), profile.profile(), reason);
|
||||
};
|
||||
// fleetd #175: point the forwarding sink handed to OpenCodeLauncher above at the real one,
|
||||
// now that `sessions` exists to resolve target -> session -> profile.
|
||||
exhaustionSinkRef.set(exhaustionSink);
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
package dev.ltms.fleet.inject;
|
||||
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* Notified when {@link CompletionResolver} actually delivers a {@code BACKEND_EXHAUSTED}
|
||||
* classification to a waiting send (CB-578 stage B) — never on a race that lost (see
|
||||
@@ -10,22 +12,81 @@ package dev.ltms.fleet.inject;
|
||||
* profiles or credentials, so mapping {@code target} to whatever should be quarantined is entirely
|
||||
* the sink's job — see {@code Fleetd.main}'s wiring, which resolves target → session → profile →
|
||||
* {@code effectiveCredentialId()} and calls {@code BackendQuarantine.quarantine} on it.
|
||||
*
|
||||
* <p><strong>The 3-arg overload is the single abstract method — fleetd #234, round 4.</strong> Two
|
||||
* earlier rounds each shipped a caller that silently dropped the profile hint (see {@link
|
||||
* #onExhausted(String, String, String)}): a lambda written against this interface can only ever
|
||||
* implement whichever overload is abstract, and while the 2-arg form held that position, EVERY
|
||||
* lambda site — a call-site forwarder in {@code Fleetd.java}, a hand-built test double — bound to
|
||||
* it and silently inherited the profile-dropping default, whether or not its author remembered the
|
||||
* hazard. Making the 3-arg form abstract instead removes the shape entirely: a lambda declared
|
||||
* against this interface today is <em>forced</em> by the compiler to take {@code (target, reason,
|
||||
* profile)}, so there is no overload left for it to bind to that can drop the hint. This is a type
|
||||
* change, not a test — it holds even for a caller that has never heard of fleetd #234.
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface ExhaustionSink {
|
||||
|
||||
/**
|
||||
* @param target the herdr terminal id whose turn was classified {@code BACKEND_EXHAUSTED}
|
||||
* @param reason the matched-line reason carried by the classification
|
||||
* @param target the herdr terminal id whose turn was classified {@code BACKEND_EXHAUSTED}
|
||||
* @param reason the matched-line reason carried by the classification
|
||||
* @param profile the profile the caller already knows should be quarantined, or {@code null}
|
||||
* when the caller has no better answer than {@code target} alone (fleetd #234):
|
||||
* a caller whose {@code target} is not yet resolvable through whatever roster
|
||||
* the sink's implementation consults — {@link
|
||||
* dev.ltms.fleet.member.OpenCodeLauncher}'s model-mismatch check fires from
|
||||
* {@code SessionAwareHandle.agentSessionId()}, which runs during {@code
|
||||
* SessionManager.acquire()} <em>before</em> that session is registered, so a
|
||||
* target -> session -> profile lookup finds nothing at that point. That launcher
|
||||
* already has its own {@code FleetConfig.Profile} in hand and does not need the
|
||||
* roster to know which profile to quarantine, so it supplies this directly
|
||||
* instead of leaving the sink to guess.
|
||||
*/
|
||||
void onExhausted(String target, String reason);
|
||||
void onExhausted(String target, String reason, String profile);
|
||||
|
||||
/**
|
||||
* Convenience for a caller with no profile to offer — every existing call site that predates
|
||||
* the hint (fleetd #234): {@link dev.ltms.fleet.inject.CompletionResolver}'s two call sites
|
||||
* always call with a {@code target} that IS live in the roster at the time of the call, so they
|
||||
* need no hint and keep working exactly as before, unchanged by this default.
|
||||
*
|
||||
* @param target as {@link #onExhausted(String, String, String)}
|
||||
* @param reason as {@link #onExhausted(String, String, String)}
|
||||
*/
|
||||
default void onExhausted(String target, String reason) {
|
||||
onExhausted(target, reason, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Inert sink — nothing happens on exhaustion. The explicit stand-in a caller (or a test not
|
||||
* exercising this feature) passes instead of a defaulting overload, exactly like
|
||||
* {@link ExhaustedPatternLookup#none()}.
|
||||
* {@link ExhaustedPatternLookup#none()}. Safe as a lambda: the 3-arg form is now the interface's
|
||||
* single abstract method, so a lambda here has no other overload to silently bind to instead —
|
||||
* it simply does nothing with all three arguments.
|
||||
*/
|
||||
static ExhaustionSink none() {
|
||||
return (target, reason) -> { };
|
||||
return (target, reason, profile) -> { };
|
||||
}
|
||||
|
||||
/**
|
||||
* A sink that forwards to whatever {@code target} currently supplies (fleetd #234). Exists to
|
||||
* break a genuine construction-order cycle: {@code Fleetd.main} builds its adapters (including
|
||||
* {@link dev.ltms.fleet.member.OpenCodeLauncher}) before {@code sessions} exists, so it cannot
|
||||
* hand them the real sink yet — it hands them a forwarder pointed at an {@code
|
||||
* AtomicReference<ExhaustionSink>} that starts at {@link #none()} and gets {@code .set()} to the
|
||||
* real sink once {@code sessions} is built. {@code target} is evaluated on every call, never
|
||||
* cached, so the forwarder keeps working after the reference is repointed.
|
||||
*
|
||||
* <p>Now safe as a one-line lambda (round 4): forwarding the single 3-arg abstract method
|
||||
* forwards everything a caller can supply — there is no separate 2-arg overload left for a
|
||||
* forwarder to bind to instead and silently lose the hint. Kept as a named factory rather than
|
||||
* written inline at each call site anyway, so a test can call the exact object {@code
|
||||
* Fleetd.java} builds instead of asserting a rebuilt copy of its shape (round 3's lesson).
|
||||
*
|
||||
* @param target supplies the sink to forward to, evaluated fresh on every call
|
||||
* @return a sink whose call delegates to {@code target.get()}
|
||||
*/
|
||||
static ExhaustionSink forwardingTo(Supplier<ExhaustionSink> target) {
|
||||
return (t, r, p) -> target.get().onExhausted(t, r, p);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,11 +3,13 @@ package dev.ltms.fleet.member;
|
||||
import dev.ltms.fleet.config.FleetConfig;
|
||||
import dev.ltms.fleet.herdr.Agent;
|
||||
import dev.ltms.fleet.herdr.AgentControl;
|
||||
import dev.ltms.fleet.herdr.AgentStatus;
|
||||
import dev.ltms.fleet.herdr.HerdrClient;
|
||||
import dev.ltms.fleet.herdr.HerdrException;
|
||||
import dev.ltms.fleet.herdr.Tab;
|
||||
import dev.ltms.fleet.herdr.Workspace;
|
||||
import dev.ltms.fleet.herdr.WorkspaceControl;
|
||||
import dev.ltms.fleet.inject.StatusRefiner;
|
||||
import dev.ltms.fleet.peer.Capability;
|
||||
import dev.ltms.fleet.peer.CharterReceipt;
|
||||
import dev.ltms.fleet.peer.MemberRole;
|
||||
@@ -151,6 +153,14 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
private final LongSupplier nowMillis; // monotonic clock (injectable for tests)
|
||||
private final Runnable sleeper; // sleep/wait hook (injectable for tests; never real-sleep in unit tests)
|
||||
|
||||
/**
|
||||
* fleetd #176 fix 2: the same UNKNOWN-refinement {@code dev.ltms.fleet.inject.StatusPoller}
|
||||
* uses, reused here for the spawn-readiness gate. Constructed once from {@link #agents} — see
|
||||
* {@link #refinedInjectable(String, Agent)} for the corroboration that keeps it from firing on
|
||||
* a dead pane's bare shell prompt.
|
||||
*/
|
||||
private final StatusRefiner statusRefiner;
|
||||
|
||||
// Per-process token mixed into each peer name so a fresh process (nameSeq back at 0) cannot
|
||||
// collide with same-profile peers that outlived a restart. See startUniquelyNamed.
|
||||
private final String nameNonce = String.format("%06x", new SecureRandom().nextInt(1 << 24));
|
||||
@@ -272,6 +282,7 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
this.memberCredentials = memberCredentials;
|
||||
this.hostEnvNames = hostEnvNames != null ? hostEnvNames : () -> System.getenv().keySet();
|
||||
this.config = config;
|
||||
this.statusRefiner = new StatusRefiner(agents);
|
||||
}
|
||||
|
||||
// --- adapter seams -------------------------------------------------------------------------
|
||||
@@ -909,16 +920,35 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
// --- spawn-readiness gate (CB-306) ---------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Poll {@link AgentControl#status} until the pane reports an injectable state or the configured
|
||||
* Poll {@link AgentControl#get} until the pane reports an injectable state or the configured
|
||||
* timeout elapses. On timeout, close the pane (self-reap) and throw.
|
||||
*
|
||||
* <p>fleetd #176 fix 1: {@code agents.get} was previously unguarded here, so a herdr
|
||||
* {@code *_not_found} answer — which is what happens when the backend process EXITED rather
|
||||
* than being slow — propagated as a raw {@link HerdrException} instead of the
|
||||
* {@link PeerUnreachableException} every other failure path of this gate produces, and skipped
|
||||
* teardown ({@link #stop}) entirely, leaking the pane/tab. {@link #failFastOnGoneBackend} closes
|
||||
* that gap: it stops waiting immediately (never burns the rest of the timeout), runs the same
|
||||
* teardown the timeout path below runs, and throws with a message that says the backend exited
|
||||
* rather than that the pane was slow. Any other {@link HerdrException} still propagates
|
||||
* unchanged — this gate does not know how to recover from it.
|
||||
*/
|
||||
private void waitUntilInjectableOrThrow(String paneId) {
|
||||
long deadline = nowMillis.getAsLong() + spawnReadyTimeoutMs;
|
||||
Object lastStatus = null;
|
||||
long start = nowMillis.getAsLong();
|
||||
long deadline = start + spawnReadyTimeoutMs;
|
||||
AgentStatus lastStatus = null;
|
||||
while (nowMillis.getAsLong() < deadline) {
|
||||
var status = agents.status(paneId);
|
||||
lastStatus = status;
|
||||
if (status.injectable()) {
|
||||
Agent sample;
|
||||
try {
|
||||
sample = agents.get(paneId);
|
||||
} catch (HerdrException e) {
|
||||
if (isAlreadyGone(e)) {
|
||||
failFastOnGoneBackend(paneId, e, nowMillis.getAsLong() - start);
|
||||
}
|
||||
throw e; // any other herdr failure is not ours to interpret — let it propagate
|
||||
}
|
||||
lastStatus = sample.status();
|
||||
if (lastStatus.injectable() || refinedInjectable(paneId, sample)) {
|
||||
log.debug("peer pane={} reached injectable state", paneId);
|
||||
return;
|
||||
}
|
||||
@@ -937,6 +967,66 @@ public abstract class HerdrPeerLauncher implements PeerLauncher {
|
||||
+ spawnReadyTimeoutMs + "ms");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #176 fix 1: the backend process exited while the gate was still waiting — herdr
|
||||
* answered {@code *_not_found} instead of ever reporting an injectable status. Fails
|
||||
* immediately (never burns the rest of {@link #spawnReadyTimeoutMs}), runs the exact same
|
||||
* teardown {@link #waitUntilInjectableOrThrow}'s timeout path runs, and throws a
|
||||
* {@link PeerUnreachableException} whose message says the process exited rather than that the
|
||||
* pane was slow.
|
||||
*
|
||||
* @throws PeerUnreachableException always — this method never returns normally
|
||||
*/
|
||||
private void failFastOnGoneBackend(String paneId, HerdrException cause, long elapsedMs) {
|
||||
String tail = readPaneQuietly(paneId); // read before stop() closes the pane
|
||||
log.warn("peer pane={} backend process exited after {}ms while waiting for injectable "
|
||||
+ "state (herdr: {}) — closing. Pane tail:\n{}",
|
||||
paneId, elapsedMs, cause.getMessage(), tail);
|
||||
stop(paneId);
|
||||
throw new PeerUnreachableException(
|
||||
"worker pane " + paneId + " backend process exited after " + elapsedMs
|
||||
+ "ms while waiting to become injectable (herdr reported: " + cause.getMessage()
|
||||
+ "). Pane tail:\n" + tail);
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #176 fix 2: resolve a raw {@link AgentStatus#UNKNOWN} sample into a trustworthy
|
||||
* injectable state via pane content, the same refinement {@code StatusPoller} applies during a
|
||||
* peer's working life — but corroborated, because the trap this gate is exposed to that the
|
||||
* poller is not: a pane whose backend has already exited settles at a plain shell prompt, and
|
||||
* that prompt commonly contains the same {@code ❯} glyph {@link StatusRefiner#classify} treats
|
||||
* as "idle at the Claude Code TUI prompt". Naively wiring the refiner in would turn "the backend
|
||||
* died" into "ready to inject" — strictly worse than today's timeout.
|
||||
*
|
||||
* <p>Two guards, both required:
|
||||
* <ul>
|
||||
* <li>{@link StatusRefiner#classify} is written for the Claude Code TUI only (its own javadoc
|
||||
* says so), so refinement only ever runs for the {@code claude} adapter — never for
|
||||
* {@code opencode} or any future non-Claude backend, whatever its pane looks like.</li>
|
||||
* <li>The refined status is accepted only when the <em>same</em> {@code agents.get} sample
|
||||
* ({@code sample}, taken once by the caller) still reports a non-null
|
||||
* {@link Agent#agentType()} — herdr's own "a supported backend is still detected here"
|
||||
* signal, read from the very sample the raw status came from so the two can never
|
||||
* disagree. A bare shell prompt left by an exited backend reports no {@code agentType}.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>Called only when {@code sample.status() == UNKNOWN}, so a healthy spawn — which never sees
|
||||
* {@code UNKNOWN} — triggers zero extra herdr calls; only a persistently-{@code unknown} pane
|
||||
* pays the one extra {@code agent.read} {@link StatusRefiner#refine} performs.
|
||||
*/
|
||||
private boolean refinedInjectable(String paneId, Agent sample) {
|
||||
if (sample.status() != AgentStatus.UNKNOWN) {
|
||||
return false;
|
||||
}
|
||||
if (!"claude".equals(namePrefix)) {
|
||||
return false; // StatusRefiner.classify reads a Claude Code TUI prompt specifically
|
||||
}
|
||||
if (sample.agentType() == null) {
|
||||
return false; // no corroborating liveness signal — could be a dead pane's bare shell
|
||||
}
|
||||
return statusRefiner.refine(paneId, AgentStatus.UNKNOWN, agents).injectable();
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #220: the pane's recent output, clipped, for the readiness-gate timeout log — or a
|
||||
* short note when it cannot be read. Best-effort by construction: this runs on a path that is
|
||||
|
||||
@@ -22,6 +22,7 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.BooleanSupplier;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.LongSupplier;
|
||||
@@ -706,6 +707,17 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
|
||||
private final ExhaustionSink exhaustionSink;
|
||||
/** CAS'd true the first (and only) time a model mismatch is reported for this handle. */
|
||||
private final AtomicBoolean modelMismatchReported = new AtomicBoolean();
|
||||
/**
|
||||
* The session id, once {@link OpenCodeSessionDiscovery#sessionIdForDirectory} first
|
||||
* resolves a non-null answer for this handle (fleetd #234). Sticky on purpose: {@code
|
||||
* directory} is a shared-cwd heuristic (see {@link OpenCodeSessionDiscovery}'s class
|
||||
* javadoc) that can start returning a DIFFERENT row once another session shares the same
|
||||
* directory and writes a newer one — re-deriving it on every call would let this handle's
|
||||
* identity silently drift to a sibling's session. Once resolved, this IS the answer, and
|
||||
* {@link #checkModelMatch} reads only the row this id names, never "whatever is newest in
|
||||
* the directory right now."
|
||||
*/
|
||||
private final AtomicReference<String> resolvedSessionId = new AtomicReference<>();
|
||||
|
||||
SessionAwareHandle(PeerHandle delegate, OpenCodeSessionDiscovery discovery, String cwd,
|
||||
FleetConfig.Profile cfg,
|
||||
@@ -759,17 +771,27 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
|
||||
}
|
||||
return null;
|
||||
}
|
||||
// fleetd #234: once resolved, stay resolved. Re-deriving from `directory` on every call
|
||||
// would let this handle's identity drift to a sibling session that later shares the
|
||||
// same cwd and writes a newer row — see resolvedSessionId's javadoc.
|
||||
String cached = resolvedSessionId.get();
|
||||
if (cached != null) {
|
||||
return cached;
|
||||
}
|
||||
// Lazy + retried, never a spawn-time blocker: opencode writes the session record only
|
||||
// when the session is first persisted, so null here is the correct interim answer and
|
||||
// the caller re-calls later (each call re-scans, picking up a record that has since
|
||||
// appeared).
|
||||
String id = discovery.sessionIdForDirectory(cwd);
|
||||
if (id != null) {
|
||||
resolvedSessionId.compareAndSet(null, id);
|
||||
}
|
||||
// fleetd #175: check on the SAME tick — while the caller (SessionManager's late-resolve
|
||||
// step) is still re-polling because the id is unknown, the row this id came from (once
|
||||
// it exists) is exactly the row that also carries the actual model. Once id resolves,
|
||||
// the caller stops calling agentSessionId() for this session, so this is naturally a
|
||||
// once-only check that happens right when the row first appears.
|
||||
checkModelMatch();
|
||||
checkModelMatch(id);
|
||||
return id;
|
||||
}
|
||||
|
||||
@@ -777,16 +799,25 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
|
||||
* Verify the live opencode session is running the model {@link #cfg} requested (fleetd
|
||||
* #175) and, on a real mismatch, log an ERROR and quarantine through {@link
|
||||
* #exhaustionSink}. A no-op when there is nothing to compare against — no model configured,
|
||||
* already reported once for this handle, or the actual model is still UNKNOWN (no row yet,
|
||||
* unreadable database, or unparseable evidence). UNKNOWN must never be treated as a
|
||||
* mismatch: that is the single most important safety rule here — a false positive would
|
||||
* already reported once for this handle, {@code sessionId} itself is not resolved yet
|
||||
* (fleetd #234: absent evidence, not a mismatch), or the actual model is still UNKNOWN (no
|
||||
* row yet, unreadable database, or unparseable evidence). UNKNOWN must never be treated as
|
||||
* a mismatch: that is the single most important safety rule here — a false positive would
|
||||
* quarantine a perfectly working profile's credential.
|
||||
*
|
||||
* @param sessionId the id {@link #agentSessionId()} just resolved (or had cached) for THIS
|
||||
* handle — the model is read back for this exact session (fleetd #234's
|
||||
* {@link OpenCodeSessionDiscovery#actualModelForSessionId}), never
|
||||
* re-derived from {@code directory}
|
||||
*/
|
||||
private void checkModelMatch() {
|
||||
private void checkModelMatch(String sessionId) {
|
||||
if (modelMismatchReported.get() || cfg.model() == null || cfg.model().isBlank()) {
|
||||
return;
|
||||
}
|
||||
OpenCodeSessionDiscovery.ActualModel actual = discovery.actualModelForDirectory(cwd);
|
||||
if (sessionId == null || sessionId.isBlank()) {
|
||||
return; // id not resolved yet — UNKNOWN, never a mismatch (fleetd #175's rule)
|
||||
}
|
||||
OpenCodeSessionDiscovery.ActualModel actual = discovery.actualModelForSessionId(sessionId);
|
||||
if (actual == null) {
|
||||
return; // UNKNOWN evidence — never a mismatch
|
||||
}
|
||||
@@ -817,10 +848,16 @@ public final class OpenCodeLauncher extends HerdrPeerLauncher {
|
||||
+ "falls back to a default model, which may be a PAID credential "
|
||||
+ "(fleetd #175); quarantining this profile's credential",
|
||||
cfg.profile(), cfg.model(), actualDisplay);
|
||||
// fleetd #234: pass our OWN profile name too. This check fires from agentSessionId(),
|
||||
// called during SessionManager.acquire() BEFORE this session is registered in
|
||||
// sessions.roster() — a roster-only sink (Fleetd's target -> session -> profile lookup)
|
||||
// finds nothing at this point and silently no-ops (defect 2). We already know exactly
|
||||
// which profile to quarantine without the roster; the sink is passed it explicitly.
|
||||
exhaustionSink.onExhausted(delegate.terminalId(),
|
||||
"opencode model mismatch: profile '" + cfg.profile() + "' requested '"
|
||||
+ cfg.model() + "' but the live session is running '" + actualDisplay
|
||||
+ "' (fleetd #175)");
|
||||
+ "' (fleetd #175)",
|
||||
cfg.profile());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -32,11 +32,19 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
||||
* here, so a layout change, or a switch to the HTTP server, changes exactly one class and nothing
|
||||
* in {@link OpenCodeLauncher}.
|
||||
*
|
||||
* <p>The determinism that makes this useful is structural, not a guess: every fleetd worker runs
|
||||
* in its own unique git worktree, so the row's {@code directory} (its project root) equals the
|
||||
* worker's cwd identifies <em>its</em> session unambiguously. We match on {@code directory} rather
|
||||
* than diffing {@code opencode session list} before/after — that races under concurrent spawns, and
|
||||
* the CLI listing does not even show the directory.
|
||||
* <p><strong>The {@code directory} match is a heuristic, not an identity — fleetd #234.</strong> A
|
||||
* worktree is opt-in: {@code fleet_spawn} only provisions one when the caller passes {@code
|
||||
* worktree:}; the default spawn inherits the lead's own cwd, which every other worker spawned the
|
||||
* same way (and every past session ever run there) shares. {@code directory} therefore does
|
||||
* <em>not</em> identify a session unambiguously in general — only in the special case of a fresh,
|
||||
* unique worktree does "most recently updated row for this directory" reliably mean "this worker's
|
||||
* own row." {@link #sessionIdForDirectory} still has to use this heuristic (the id has to come from
|
||||
* somewhere, and nothing else is available at this layer — see that method's javadoc), but a caller
|
||||
* that already holds a resolved id must never re-derive evidence about that same session via
|
||||
* {@code directory} again; see {@link #actualModelForSessionId}, which looks up by {@code id}
|
||||
* instead for exactly this reason. We match on {@code directory} rather than diffing
|
||||
* {@code opencode session list} before/after — that races under concurrent spawns, and the CLI
|
||||
* listing does not even show the directory.
|
||||
*
|
||||
* <p>All reads are best-effort and never throw: a missing or unreadable database, a query that
|
||||
* fails, or a directory with no row yet all yield {@code null}, and the caller (the session
|
||||
@@ -89,8 +97,16 @@ final class OpenCodeSessionDiscovery {
|
||||
/**
|
||||
* The opencode session id whose row references {@code directory} (the worker's cwd), or
|
||||
* {@code null} when no row matches yet. When several rows share the directory — e.g. repeated
|
||||
* spawns into the same worktree — the row with the highest {@code time_updated} wins: it is
|
||||
* the session the pane most likely corresponds to.
|
||||
* spawns into the same worktree, OR several workers sharing one cwd because none of them was
|
||||
* given a worktree (fleetd #234) — the row with the highest {@code time_updated} wins: it is
|
||||
* the session the pane most likely corresponds to. That "most likely" is a real caveat, not a
|
||||
* formality: when the directory is shared, this can and does pick another session's row (see
|
||||
* the class javadoc). This is the one place fleetd resolves an opencode session id at all —
|
||||
* nothing else is available at this layer to disambiguate further (no {@code opencode session
|
||||
* list} entry names the directory, and diffing before/after races under concurrent spawns) — so
|
||||
* the heuristic stays here unchanged. What must never happen is a SECOND, independent piece of
|
||||
* evidence about the same session being re-derived via {@code directory} once an id has already
|
||||
* come out of this method; see {@link #actualModelForSessionId}.
|
||||
*
|
||||
* <p>Never throws: a missing {@code opencode.db}, a locked/unreadable database, a query
|
||||
* failure, or a directory that has not been persisted yet all resolve to {@code null} rather
|
||||
@@ -134,24 +150,36 @@ final class OpenCodeSessionDiscovery {
|
||||
}
|
||||
|
||||
/**
|
||||
* The model opencode actually ran the {@code directory}'s most-recent session on (fleetd
|
||||
* #175), read from the same row {@link #sessionIdForDirectory} matches — but via its own
|
||||
* query and its own connection, deliberately kept independent so a database whose schema
|
||||
* predates the {@code model} column (or any other read failure on this column alone) can
|
||||
* never take {@link #sessionIdForDirectory}'s id resolution down with it. That would be a
|
||||
* regression of the id-resolution feature #209 shipped; this method degrades on its own.
|
||||
* The model opencode actually ran the session {@code sessionId} on (fleetd #175/#234), read by
|
||||
* primary-key lookup — the ONE row that id names, and no other. Deliberately keyed on
|
||||
* {@code id} rather than {@code directory}: two independent {@code WHERE directory = ? ORDER BY
|
||||
* time_updated DESC LIMIT 1} queries (one for the id, one for the model) can each pick a
|
||||
* DIFFERENT row once more than one session shares a directory (fleetd #234 — the default
|
||||
* no-worktree spawn shares the lead's cwd with every other worker and every past session ever
|
||||
* run there), silently comparing a profile's requested model against a session that is not even
|
||||
* the one whose id was returned. Keying on {@code id} instead makes that impossible: the model
|
||||
* read back is always the SAME session {@link #sessionIdForDirectory} (or a cached copy of its
|
||||
* answer) already resolved.
|
||||
*
|
||||
* <p>Kept as its own query and its own connection, independent from {@link
|
||||
* #sessionIdForDirectory}: a database whose schema predates the {@code model} column (or any
|
||||
* other read failure on this column alone) can never take id resolution down with it. That
|
||||
* would be a regression of the id-resolution feature #209 shipped; this method degrades on its
|
||||
* own.
|
||||
*
|
||||
* <p>Never throws, and every failure mode — no matching row, a missing/unreadable database, a
|
||||
* missing {@code model} column, a null/blank {@code model} value, or JSON that does not parse
|
||||
* into {@code {"id": "...", "providerID": "..."}} with a non-blank {@code id} — resolves to
|
||||
* {@code null}. That is UNKNOWN evidence, not a mismatch signal: the caller must never
|
||||
* quarantine a profile on the strength of a {@code null} here.
|
||||
* quarantine a profile on the strength of a {@code null} here. A blank/null {@code sessionId}
|
||||
* (the id is not resolved yet) is UNKNOWN too, for the same reason — never call this with one.
|
||||
*
|
||||
* @param directory the worker's cwd, as resolved for this spawn
|
||||
* @param sessionId the session id already resolved by {@link #sessionIdForDirectory} for this
|
||||
* spawn — never re-derived from {@code directory} here
|
||||
* @return the actual model, or {@code null} when unknown
|
||||
*/
|
||||
ActualModel actualModelForDirectory(String directory) {
|
||||
if (directory == null || directory.isBlank()) {
|
||||
ActualModel actualModelForSessionId(String sessionId) {
|
||||
if (sessionId == null || sessionId.isBlank()) {
|
||||
return null;
|
||||
}
|
||||
if (!Files.isRegularFile(databasePath)) {
|
||||
@@ -159,10 +187,10 @@ final class OpenCodeSessionDiscovery {
|
||||
// exact condition — do not double-log it here.
|
||||
return null;
|
||||
}
|
||||
String sql = "SELECT model FROM session WHERE directory = ? ORDER BY time_updated DESC LIMIT 1";
|
||||
String sql = "SELECT model FROM session WHERE id = ?";
|
||||
try (Connection connection = openReadOnly();
|
||||
PreparedStatement statement = connection.prepareStatement(sql)) {
|
||||
statement.setString(1, directory);
|
||||
statement.setString(1, sessionId);
|
||||
try (ResultSet rows = statement.executeQuery()) {
|
||||
if (rows.next()) {
|
||||
return parseModel(rows.getString("model"));
|
||||
|
||||
@@ -44,10 +44,13 @@ public final class FakeHerdr implements HerdrClient {
|
||||
private String agentSendErrorCode = null;
|
||||
private boolean noPanes = false;
|
||||
private volatile String agentStatus = "idle"; // steady-state agent.get status
|
||||
private volatile String agentType = "claude"; // detected agent kind on agent.get; null = undetected
|
||||
private volatile String readText = "worker transcript tail"; // canned agent.read output
|
||||
private int pinnedStarts = 0; // how many upcoming agent.start calls report a fixed pane
|
||||
private String pinnedStartTerminal;
|
||||
private String pinnedStartPane;
|
||||
private volatile int agentGetOkCalls = Integer.MAX_VALUE; // how many agent.get calls succeed first
|
||||
private volatile String agentGetFailCode = null; // error code every agent.get call after that reports
|
||||
|
||||
public FakeHerdr healthy(boolean h) {
|
||||
this.healthy = h;
|
||||
@@ -103,6 +106,27 @@ public final class FakeHerdr implements HerdrClient {
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the detected agent kind ({@code "agent"} field) that {@code agent.get} reports —
|
||||
* {@code null} models herdr not (or no longer) detecting a supported backend in the pane, e.g.
|
||||
* a bare shell prompt (fleetd #176 fix 2 corroboration test).
|
||||
*/
|
||||
public FakeHerdr agentType(String type) {
|
||||
this.agentType = type;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Make {@code agent.get} succeed normally for its first {@code okCalls} invocations, then fail
|
||||
* every call after that with {@code code} — fleetd #176 fix 1's "backend exited mid-wait"
|
||||
* fixture. {@code okCalls == 0} fails from the very first call.
|
||||
*/
|
||||
public FakeHerdr agentGetFailsWithAfter(int okCalls, String code) {
|
||||
this.agentGetOkCalls = okCalls;
|
||||
this.agentGetFailCode = code;
|
||||
return this;
|
||||
}
|
||||
|
||||
/** The text {@code agent.read} returns (the CB-106 completion scrape). */
|
||||
public FakeHerdr readText(String text) {
|
||||
this.readText = text;
|
||||
@@ -208,10 +232,21 @@ public final class FakeHerdr implements HerdrClient {
|
||||
}
|
||||
yield mapper.readTree("{\"type\":\"ok\"}");
|
||||
}
|
||||
case "agent.get" -> mapper.readTree(("""
|
||||
{"type":"agent_info","agent":{"terminal_id":"term_a","agent":"claude",
|
||||
case "agent.get" -> {
|
||||
if (agentGetFailCode != null) {
|
||||
long getCalls = calls.stream().filter(c -> c.method().equals("agent.get")).count();
|
||||
if (getCalls > agentGetOkCalls) {
|
||||
throw new HerdrException(
|
||||
"herdr error [" + agentGetFailCode + "]: agent target not found",
|
||||
agentGetFailCode, null);
|
||||
}
|
||||
}
|
||||
String agentField = agentType == null ? "null" : "\"" + agentType + "\"";
|
||||
yield mapper.readTree(("""
|
||||
{"type":"agent_info","agent":{"terminal_id":"term_a","agent":%s,
|
||||
"agent_status":"%s","workspace_id":"w2","tab_id":"w2:t7","pane_id":"w2:p7"}}""")
|
||||
.formatted(agentStatus));
|
||||
.formatted(agentField, agentStatus));
|
||||
}
|
||||
case "agent.read" -> mapper.readTree(mapper.writeValueAsString(
|
||||
java.util.Map.of("type", "agent_read", "read", java.util.Map.of("text", readText))));
|
||||
case "agent.start" -> {
|
||||
|
||||
@@ -499,7 +499,7 @@ class CompletionResolverTest {
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
|
||||
java.util.List<String> notified = new java.util.ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> notified.add(target + ": " + reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
|
||||
|
||||
var waiter = rendezvous.open("term_a");
|
||||
@@ -520,7 +520,7 @@ class CompletionResolverTest {
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
|
||||
java.util.List<String> notified = new java.util.ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> notified.add(target + ": " + reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
|
||||
|
||||
var waiter = rendezvous.open("term_a");
|
||||
@@ -648,7 +648,7 @@ class CompletionResolverTest {
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
|
||||
java.util.List<String> notified = new java.util.ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> notified.add(target + ": " + reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
|
||||
|
||||
var waiter = rendezvous.open("term_a");
|
||||
@@ -701,7 +701,7 @@ class CompletionResolverTest {
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
|
||||
java.util.List<String> notified = new java.util.ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> notified.add(target + ": " + reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
|
||||
|
||||
var waiter = rendezvous.open("term_a");
|
||||
@@ -728,7 +728,7 @@ class CompletionResolverTest {
|
||||
Rendezvous rendezvous = new Rendezvous();
|
||||
ExhaustedPatternLookup patterns = target -> Pattern.compile("usage limit has been reached");
|
||||
java.util.List<String> notified = new java.util.ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> notified.add(target + ": " + reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> notified.add(target + ": " + reason);
|
||||
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, patterns, sink);
|
||||
|
||||
var waiter = rendezvous.open("term_a");
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
package dev.ltms.fleet.inject;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
|
||||
/**
|
||||
* fleetd #234. Round 3: calls the REAL production factory, {@link ExhaustionSink#forwardingTo},
|
||||
* rather than rebuilding its shape locally — a round-2 version of this test built its own copy of
|
||||
* the forwarder inline, so mutating {@code Fleetd.java}'s actual forwarder back into a lambda left
|
||||
* that copy untouched and the test kept passing while production had regressed. Calling the shared
|
||||
* factory here means the same object under test IS the object {@code Fleetd.java} builds via
|
||||
* {@code ExhaustionSink.forwardingTo(exhaustionSinkRef::get)}.
|
||||
*
|
||||
* <p>Round 4: the 3-arg overload is now {@link ExhaustionSink}'s single abstract method, so {@code
|
||||
* forwardingTo} itself is a one-line lambda and every sink below can safely be one too — there is
|
||||
* no 2-arg overload left for any of them to silently bind to instead. This test still catches a
|
||||
* regression inside {@code forwardingTo}'s body (e.g. one that calls the 2-arg default and drops
|
||||
* the hint that way) because it still goes through the shared factory rather than a rebuilt copy.
|
||||
*/
|
||||
class ExhaustionSinkForwardingHazardTest {
|
||||
|
||||
@Test
|
||||
void forwardingToDeliversTheProfileHintToWhateverSinkTheSupplierCurrentlyReturns() {
|
||||
AtomicReference<String> hintSeenByRealSink = new AtomicReference<>("NEVER CALLED");
|
||||
ExhaustionSink real = (target, reason, profileHint) -> hintSeenByRealSink.set(profileHint);
|
||||
|
||||
// Same shape Fleetd.java uses: a reference that starts at none() and is repointed later.
|
||||
AtomicReference<ExhaustionSink> ref = new AtomicReference<>(ExhaustionSink.none());
|
||||
ExhaustionSink forwarding = ExhaustionSink.forwardingTo(ref::get);
|
||||
ref.set(real);
|
||||
|
||||
forwarding.onExhausted("term_x", "model mismatch", "gx");
|
||||
|
||||
assertEquals("gx", hintSeenByRealSink.get(),
|
||||
"ExhaustionSink.forwardingTo must deliver the profile hint to the sink the supplier "
|
||||
+ "currently returns — the exact object Fleetd.java's forwarder is built from");
|
||||
}
|
||||
}
|
||||
@@ -963,6 +963,152 @@ class ClaudeCodeLauncherTest {
|
||||
assertDoesNotThrow(() -> UUID.fromString(handle.id()));
|
||||
}
|
||||
|
||||
// --- fleetd #176 fix 1: fail fast when the backend process exits mid-wait -------------------
|
||||
|
||||
@Test
|
||||
void spawnFailsFastWhenBackendProcessExitsMidWaitInsteadOfBurningTheTimeout() {
|
||||
// First agent.get sees UNKNOWN (one normal tick); the second reports the pane gone, exactly
|
||||
// what herdr answers when the backend process has already exited. The timeout is generous
|
||||
// (60s) so a test that wrongly falls through to the old unguarded call — and therefore
|
||||
// waits out the whole window — is unambiguously distinguishable from one that fails fast.
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
herdr.agentStatus("unknown");
|
||||
herdr.agentGetFailsWithAfter(1, "pane_not_found");
|
||||
long[] clock = {0};
|
||||
|
||||
ClaudeCodeLauncher svc = new ClaudeCodeLauncher(
|
||||
new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
new SubscriptionGuard(Set.of("gx00.gw")),
|
||||
workerConfigMap("ltms-local", null), "ltms-local", _ -> null,
|
||||
60_000, () -> clock[0], () -> clock[0] += 50);
|
||||
|
||||
PeerUnreachableException ex = assertThrows(
|
||||
PeerUnreachableException.class,
|
||||
() -> svc.spawn(new SpawnRequest(null, null, null)));
|
||||
|
||||
assertTrue(ex.getMessage().contains("w9:pRoot_1"),
|
||||
"exception message references the paneId: " + ex.getMessage());
|
||||
assertTrue(ex.getMessage().toLowerCase().contains("exited"),
|
||||
"exception message says the backend exited, not that the pane was slow: "
|
||||
+ ex.getMessage());
|
||||
assertTrue(clock[0] < 60_000,
|
||||
"the gate must not burn the rest of the 60s timeout: clock only reached " + clock[0]);
|
||||
long getCalls = herdr.calls.stream().filter(c -> c.method().equals("agent.get")).count();
|
||||
assertEquals(2, getCalls,
|
||||
"exactly one normal poll then the not_found answer — no further polling after that: "
|
||||
+ getCalls);
|
||||
assertEquals(1, paneCloseCount(herdr, "w9:pRoot_1"),
|
||||
"the pane is torn down (no orphan) even on the fail-fast path");
|
||||
}
|
||||
|
||||
@Test
|
||||
void spawnLetsAnUnrelatedHerdrErrorPropagateUnchanged() {
|
||||
// Fix 1 must only special-case a "*_not_found" answer. Any other herdr failure keeps
|
||||
// propagating as-is — this gate does not know how to recover from it.
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
herdr.agentStatus("unknown");
|
||||
herdr.agentGetFailsWithAfter(0, "internal_error");
|
||||
long[] clock = {0};
|
||||
|
||||
ClaudeCodeLauncher svc = new ClaudeCodeLauncher(
|
||||
new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
new SubscriptionGuard(Set.of("gx00.gw")),
|
||||
workerConfigMap("ltms-local", null), "ltms-local", _ -> null,
|
||||
60_000, () -> clock[0], () -> clock[0] += 50);
|
||||
|
||||
dev.ltms.fleet.herdr.HerdrException ex = assertThrows(
|
||||
dev.ltms.fleet.herdr.HerdrException.class,
|
||||
() -> svc.spawn(new SpawnRequest(null, null, null)));
|
||||
|
||||
assertEquals("internal_error", ex.code());
|
||||
assertEquals(0, paneCloseCount(herdr, "w9:pRoot_1"),
|
||||
"an error this gate does not recognize is not this gate's teardown to run");
|
||||
}
|
||||
|
||||
// --- fleetd #176 fix 2: corroborated UNKNOWN refinement --------------------------------------
|
||||
|
||||
@Test
|
||||
void refinedIdleIsNotAcceptedWhenAgentTypeIsNull() {
|
||||
// The trap fix 2 must close: a pane sitting at a bare shell prompt after its backend exited
|
||||
// still contains the same "❯" glyph StatusRefiner.classify treats as "idle at the Claude
|
||||
// Code TUI prompt". Without the agentType corroboration this would be misread as injectable
|
||||
// and the gate would hand back a peer that never started. herdr reports no agentType for
|
||||
// that bare shell.
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
herdr.agentStatus("unknown"); // always UNKNOWN
|
||||
herdr.agentType(null); // no supported backend detected — could be a bare shell
|
||||
herdr.readText("some-host:~ user$ ❯ "); // looks exactly like an idle Claude Code prompt
|
||||
long[] clock = {0};
|
||||
|
||||
ClaudeCodeLauncher svc = new ClaudeCodeLauncher(
|
||||
new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
new SubscriptionGuard(Set.of("gx00.gw")),
|
||||
workerConfigMap("ltms-local", null), "ltms-local", _ -> null,
|
||||
500, () -> clock[0], () -> clock[0] += 50);
|
||||
|
||||
PeerUnreachableException ex = assertThrows(
|
||||
PeerUnreachableException.class,
|
||||
() -> svc.spawn(new SpawnRequest(null, null, null)));
|
||||
|
||||
assertTrue(clock[0] >= 500,
|
||||
"a null agentType must not let the '❯' prompt refine to injectable — the gate has "
|
||||
+ "to wait out the full timeout: clock only reached " + clock[0]);
|
||||
assertEquals(1, paneCloseCount(herdr, "w9:pRoot_1"),
|
||||
"the pane is torn down on timeout, same as any other never-injectable spawn");
|
||||
}
|
||||
|
||||
@Test
|
||||
void refinedIdleIsAcceptedWhenAgentTypeCorroboratesLiveness() {
|
||||
// The positive case: a genuinely live claude pane that herdr misreports as UNKNOWN (CB-115)
|
||||
// still resolves to injectable once agentType corroborates that a supported backend is
|
||||
// detected in the same sample.
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
herdr.agentStatus("unknown"); // always UNKNOWN at the raw level
|
||||
herdr.agentType("claude"); // herdr still detects a live claude backend
|
||||
herdr.readText("? for shortcuts"); // StatusRefiner.classify's idle footer marker
|
||||
long[] clock = {0};
|
||||
|
||||
ClaudeCodeLauncher svc = new ClaudeCodeLauncher(
|
||||
new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
new SubscriptionGuard(Set.of("gx00.gw")),
|
||||
workerConfigMap("ltms-local", null), "ltms-local", _ -> null,
|
||||
60_000, () -> clock[0], () -> clock[0] += 50);
|
||||
|
||||
PeerHandle handle = svc.spawn(new SpawnRequest(null, null, null));
|
||||
|
||||
assertNotNull(handle, "a corroborated refined-IDLE sample lets the spawn succeed");
|
||||
assertTrue(clock[0] < 60_000,
|
||||
"refinement must resolve well before the timeout: clock reached " + clock[0]);
|
||||
assertEquals(0, paneCloseCount(herdr, "w9:pRoot_1"),
|
||||
"no pane close — the peer is genuinely injectable");
|
||||
}
|
||||
|
||||
@Test
|
||||
void refinementNeverFiresWhenRawStatusIsAlreadyInjectable() {
|
||||
// Acceptance criterion 4: refinement must trigger only on a raw UNKNOWN sample. Seed the
|
||||
// pane content with an active-generation marker that StatusRefiner.classify would read as
|
||||
// WORKING (never injectable) if refine() were wrongly invoked here — proving that a raw
|
||||
// IDLE status short-circuits before refine() (and its extra agent.read call) ever runs.
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
herdr.agentStatus("idle"); // already injectable at the raw level
|
||||
herdr.agentType("claude");
|
||||
herdr.readText("esc to interrupt"); // would classify as WORKING if refine() ran anyway
|
||||
|
||||
long[] clock = {0};
|
||||
ClaudeCodeLauncher gated = new ClaudeCodeLauncher(
|
||||
new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
new SubscriptionGuard(Set.of("gx00.gw")),
|
||||
workerConfigMap("ltms-local", null), "ltms-local", _ -> null,
|
||||
5000, () -> clock[0], () -> clock[0] += 300);
|
||||
|
||||
PeerHandle handle = gated.spawn(new SpawnRequest(null, null, null));
|
||||
|
||||
assertNotNull(handle, "an already-injectable raw status succeeds without ever refining");
|
||||
assertFalse(herdr.called("agent.read"),
|
||||
"refine() must never run (and so never call agent.read) when the raw status is "
|
||||
+ "already injectable");
|
||||
}
|
||||
|
||||
// --- CB-511: worker environment seeding -----------------------------------------------------
|
||||
|
||||
@Test
|
||||
|
||||
@@ -17,6 +17,7 @@ import dev.ltms.fleet.peer.MemberRole;
|
||||
import dev.ltms.fleet.peer.PeerHandle;
|
||||
import dev.ltms.fleet.peer.PeerUnreachableException;
|
||||
import dev.ltms.fleet.peer.SpawnRequest;
|
||||
import dev.ltms.fleet.placement.BackendQuarantine;
|
||||
import dev.ltms.fleet.session.MemberSession;
|
||||
import dev.ltms.fleet.session.SessionManager;
|
||||
import org.junit.jupiter.api.Test;
|
||||
@@ -34,6 +35,7 @@ import java.util.Optional;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
@@ -52,6 +54,19 @@ class OpenCodeLauncherTest {
|
||||
null, null, gitTokenEnv, null, FleetConfig.Profile.KIND_OPENCODE);
|
||||
}
|
||||
|
||||
/**
|
||||
* A profile carrying an explicit {@code credentialId} (fleetd #234 defect 2) — distinct from
|
||||
* the profile's own name, so a test can assert on the credential precisely rather than relying
|
||||
* on {@code effectiveCredentialId()}'s profile-name fallback.
|
||||
*/
|
||||
private static FleetConfig.Profile opencodeCfgWithCredential(String profileName, String model,
|
||||
String credentialId) {
|
||||
return new FleetConfig.Profile(profileName, null, model, null, "FLEETD_WORKER_TOKEN",
|
||||
List.of("opencode"), "tab", "fleetd-workers", "opencode: {model} #{n}", null,
|
||||
null, List.of(), null, null, FleetConfig.Profile.KIND_OPENCODE, Map.of(), 1.0f,
|
||||
null, false, null, credentialId, null);
|
||||
}
|
||||
|
||||
/** Gate-disabled launcher whose per-spawn config dirs land under an inspectable temp root. */
|
||||
private static OpenCodeLauncher service(FakeHerdr herdr, Path configRoot, FleetConfig.Profile cfg) {
|
||||
return new OpenCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
@@ -396,6 +411,46 @@ class OpenCodeLauncherTest {
|
||||
assertNotNull(ex.getMessage());
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #176 fix 2's per-adapter guard: {@code StatusRefiner.classify} is written for the
|
||||
* Claude Code TUI only, and the spawn-readiness gate must never run it against an opencode
|
||||
* pane. {@code agentType("opencode")} deliberately satisfies the OTHER guard (the corroborating
|
||||
* liveness check) so it cannot be what makes this test pass — only the {@code namePrefix}
|
||||
* check can be. If that check were ever removed, this pane's {@code ❯} content would refine
|
||||
* straight to IDLE and the gate would report ready before the backend actually was.
|
||||
*/
|
||||
@Test
|
||||
void opencodePaneIsNeverRefinedEvenWhenItsContentLooksLikeAnIdleClaudePrompt(@TempDir Path root) {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
herdr.agentStatus("unknown"); // always UNKNOWN
|
||||
herdr.agentType("opencode"); // non-null — satisfies the liveness guard on its own
|
||||
herdr.readText("some-host:~ user$ ❯ "); // content StatusRefiner.classify reads as IDLE
|
||||
long[] clock = {0};
|
||||
OpenCodeLauncher svc = new OpenCodeLauncher(new AgentControl(herdr), new WorkspaceControl(herdr),
|
||||
Map.of("gemini", opencodeCfg(null, null, null)), "gemini", _ -> null,
|
||||
1000, () -> clock[0], () -> clock[0] += 50, root, root);
|
||||
|
||||
assertThrows(PeerUnreachableException.class,
|
||||
() -> svc.spawn(new SpawnRequest(null, null, null)));
|
||||
|
||||
assertTrue(clock[0] >= 1000,
|
||||
"an opencode pane must never refine to injectable, however its content reads — "
|
||||
+ "the gate has to wait out the full timeout: clock only reached " + clock[0]);
|
||||
// A blanket "agent.read is never called" does not hold here: the timeout path itself reads
|
||||
// the pane tail (source=recent) for its own log message, on every timeout, regardless of
|
||||
// adapter — see HerdrPeerLauncher.readPaneQuietly. So assert on the refiner's OWN probe
|
||||
// source (StatusRefiner.PROBE_SOURCE = "detection") instead — that call happens only inside
|
||||
// StatusRefiner.refine, so its absence proves the refiner itself was never reached for this
|
||||
// opencode pane, not merely that its answer was discarded.
|
||||
long detectionReads = herdr.calls.stream()
|
||||
.filter(c -> c.method().equals("agent.read"))
|
||||
.filter(c -> "detection".equals(((Map<?, ?>) c.params()).get("source")))
|
||||
.count();
|
||||
assertEquals(0, detectionReads,
|
||||
"the refiner's own pane probe (source=detection) must never run against an "
|
||||
+ "opencode pane — the namePrefix guard has to stop it before that call");
|
||||
}
|
||||
|
||||
@Test
|
||||
void spawnReturnsHandleWhenGateDisabled(@TempDir Path root) {
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
@@ -870,7 +925,7 @@ class OpenCodeLauncherTest {
|
||||
// credentialId — the profile that actually escaped the fleet's accounting.
|
||||
FleetConfig.Profile cfg = opencodeCfg("opencode/nemotron-3-ultra-free", null, null);
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(target + "|" + reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(target + "|" + reason);
|
||||
OpenCodeLauncher launcher = serviceWithSink(herdr, configRoot, discRoot, cfg, sink);
|
||||
|
||||
SessionManager sessions = new SessionManager(launcher);
|
||||
@@ -905,7 +960,7 @@ class OpenCodeLauncherTest {
|
||||
void aProviderPrefixedModelMatchingBothIdAndProviderIsNotAMismatch(@TempDir Path configRoot,
|
||||
@TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg("openai/gpt-5.6-terra", null, null);
|
||||
PeerHandle handle = serviceWithSink(new FakeHerdr(), configRoot, discRoot, cfg, sink)
|
||||
.spawn(new SpawnRequest(null, "/work/dir", null));
|
||||
@@ -921,7 +976,7 @@ class OpenCodeLauncherTest {
|
||||
void aGxProviderPrefixedModelMatchingBothIdAndProviderIsNotAMismatch(@TempDir Path configRoot,
|
||||
@TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg("gx/deepseek-v4-flash", null, null);
|
||||
PeerHandle handle = serviceWithSink(new FakeHerdr(), configRoot, discRoot, cfg, sink)
|
||||
.spawn(new SpawnRequest(null, "/work/dir", null));
|
||||
@@ -943,7 +998,7 @@ class OpenCodeLauncherTest {
|
||||
void aMissingProviderIdInTheEvidenceIsUnknownNotAMismatchWhenTheIdMatches(
|
||||
@TempDir Path configRoot, @TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg("openai/gpt-5.6-terra", null, null);
|
||||
PeerHandle handle = serviceWithSink(new FakeHerdr(), configRoot, discRoot, cfg, sink)
|
||||
.spawn(new SpawnRequest(null, "/work/dir", null));
|
||||
@@ -964,7 +1019,7 @@ class OpenCodeLauncherTest {
|
||||
void aMissingProviderIdInTheEvidenceStillCatchesARealIdMismatch(
|
||||
@TempDir Path configRoot, @TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg("openai/gpt-5.6-terra", null, null);
|
||||
PeerHandle handle = serviceWithSink(new FakeHerdr(), configRoot, discRoot, cfg, sink)
|
||||
.spawn(new SpawnRequest(null, "/work/dir", null));
|
||||
@@ -988,7 +1043,7 @@ class OpenCodeLauncherTest {
|
||||
void aBareModelWithNoProviderPrefixMatchesOnIdAloneAndIsNotAMismatch(@TempDir Path configRoot,
|
||||
@TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg("deepseek-v4-flash", null, null);
|
||||
PeerHandle handle = serviceWithSink(new FakeHerdr(), configRoot, discRoot, cfg, sink)
|
||||
.spawn(new SpawnRequest(null, "/work/dir", null));
|
||||
@@ -1010,7 +1065,7 @@ class OpenCodeLauncherTest {
|
||||
void aRealIdMismatchLogsAnErrorNamingBothModelsAndQuarantinesThroughTheSink(
|
||||
@TempDir Path configRoot, @TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(target + "|" + reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(target + "|" + reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg("opencode/nemotron-3-ultra-free", null, null);
|
||||
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(OpenCodeLauncher.class);
|
||||
@@ -1057,7 +1112,7 @@ class OpenCodeLauncherTest {
|
||||
void unknownOrUnparseableModelEvidenceNeverQuarantines(@TempDir Path configRoot,
|
||||
@TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg("openai/gpt-5.6-terra", null, null);
|
||||
PeerHandle handle = serviceWithSink(new FakeHerdr(), configRoot, discRoot, cfg, sink)
|
||||
.spawn(new SpawnRequest(null, "/work/dir", null));
|
||||
@@ -1083,7 +1138,7 @@ class OpenCodeLauncherTest {
|
||||
void aProfileWithNoConfiguredModelIsNeverCheckedForAMismatch(@TempDir Path configRoot,
|
||||
@TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason) -> exhausted.add(reason);
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg(null, null, null);
|
||||
PeerHandle handle = serviceWithSink(new FakeHerdr(), configRoot, discRoot, cfg, sink)
|
||||
.spawn(new SpawnRequest(null, "/work/dir", null));
|
||||
@@ -1093,4 +1148,200 @@ class OpenCodeLauncherTest {
|
||||
assertEquals("ses_x", handle.agentSessionId());
|
||||
assertTrue(exhausted.isEmpty(), "no model configured → nothing to compare: " + exhausted);
|
||||
}
|
||||
|
||||
// --- fleetd #234 defect 1: key the model check on the RESOLVED id, not the shared directory --
|
||||
|
||||
/**
|
||||
* The exact shape fleetd #234 reported: {@code fleet_spawn} with no {@code worktree:} shares
|
||||
* the lead's cwd across every worker, so more than one session row can exist for the SAME
|
||||
* {@code directory}. Once THIS handle's own session id is resolved, a sibling member spawned
|
||||
* later into the same shared directory — writing a NEWER, unrelated row — must never make the
|
||||
* already-resolved session look mismatched. Today's code re-derives "the newest row in this
|
||||
* directory" on every call (both for the id AND, independently, for the model), so it would
|
||||
* pick up the sibling's row on the second call and flag a false mismatch AND flip the returned
|
||||
* id. The fix (fleetd #234) makes the id sticky once resolved and reads the model back for
|
||||
* exactly that id (see {@link OpenCodeSessionDiscovery#actualModelForSessionId}) — never
|
||||
* "whatever is newest in the directory right now."
|
||||
*/
|
||||
@Test
|
||||
void modelCheckReadsTheResolvedSessionsOwnRowNotWhateverIsNewestInTheSharedDirectory(
|
||||
@TempDir Path configRoot, @TempDir Path discRoot) throws Exception {
|
||||
List<String> exhausted = new ArrayList<>();
|
||||
ExhaustionSink sink = (target, reason, profile) -> exhausted.add(reason);
|
||||
FleetConfig.Profile cfg = opencodeCfg("openai/gpt-5.6-terra", null, null);
|
||||
PeerHandle handle = serviceWithSink(new FakeHerdr(), configRoot, discRoot, cfg, sink)
|
||||
.spawn(new SpawnRequest(null, "/work/dir", null));
|
||||
|
||||
// Our own session's row, correctly matching the profile's requested model.
|
||||
OpenCodeSessionDiscoveryTest.writeRecord(discRoot, "ses_ours", "/work/dir", 1000L,
|
||||
"{\"id\":\"gpt-5.6-terra\",\"providerID\":\"openai\"}");
|
||||
assertEquals("ses_ours", handle.agentSessionId(), "resolves to our own session");
|
||||
assertTrue(exhausted.isEmpty(), "matching model → no mismatch on first resolve: " + exhausted);
|
||||
|
||||
// A sibling member, spawned later into the SAME shared directory (no worktree, fleetd
|
||||
// #234's default), writes a newer row running a totally different model.
|
||||
OpenCodeSessionDiscoveryTest.writeRecord(discRoot, "ses_sibling", "/work/dir", 9000L,
|
||||
"{\"id\":\"deepseek-v4-flash\",\"providerID\":\"gx\"}");
|
||||
|
||||
assertEquals("ses_ours", handle.agentSessionId(),
|
||||
"the session id, once resolved, must not flip to a sibling sharing the directory");
|
||||
assertTrue(exhausted.isEmpty(),
|
||||
"a sibling's later, unrelated row in the same shared directory must never be read "
|
||||
+ "as OUR session's model: " + exhausted);
|
||||
}
|
||||
|
||||
// --- fleetd #234 defect 2: the quarantine the ERROR announces must actually happen -----------
|
||||
|
||||
/**
|
||||
* fleetd #234: {@link OpenCodeLauncher.SessionAwareHandle#checkModelMatch} fires from {@code
|
||||
* agentSessionId()}, which {@code SessionManager.acquire()} calls to build the very first
|
||||
* {@code MemberSession} record — BEFORE that session is put into the registry {@code
|
||||
* sessions.roster()} reads. A sink that resolves {@code target -> profile} ONLY through the
|
||||
* roster (today's {@code Fleetd.java} code, before this fix) therefore finds nothing at this
|
||||
* exact moment and silently does not quarantine, even though it just logged an ERROR saying it
|
||||
* would. This test drives the REAL path — {@code SessionManager.acquire()} — not the sink
|
||||
* directly, because the bug is entirely about this ordering; a direct-sink test cannot see it
|
||||
* (and is exactly why #175's own test suite, which only ever called the sink directly or after
|
||||
* registration, never caught this).
|
||||
*
|
||||
* <p>The sink under test mirrors {@code Fleetd.main()}'s real wiring after the fix: resolve
|
||||
* via the roster first (unchanged for {@code CompletionResolver}'s two call sites), falling
|
||||
* back to the profile hint {@link OpenCodeLauncher} now supplies via {@link
|
||||
* ExhaustionSink#onExhausted(String, String, String)} when the roster lookup misses.
|
||||
*/
|
||||
@Test
|
||||
void aSpawnTimeModelMismatchActuallyQuarantinesTheCredentialThroughTheRealAcquirePath(
|
||||
@TempDir Path configRoot, @TempDir Path discRoot) throws Exception {
|
||||
FleetConfig.Profile cfg = opencodeCfgWithCredential(
|
||||
"terra", "opencode/nemotron-3-ultra-free", "openai-shared");
|
||||
Map<String, FleetConfig.Profile> profiles = Map.of(cfg.profile(), cfg);
|
||||
BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.SECONDS.toNanos(1800));
|
||||
|
||||
// fleetd #234, round 4: safely a lambda now — the 3-arg overload is the interface's single
|
||||
// abstract method, so there is no 2-arg overload left to bind to instead.
|
||||
ExhaustionSink sink = (target, reason, profileHint) -> {
|
||||
FleetConfig.Profile profile = profileHint == null ? null : profiles.get(profileHint);
|
||||
if (profile != null) {
|
||||
quarantine.quarantine(profile.effectiveCredentialId());
|
||||
}
|
||||
};
|
||||
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
OpenCodeLauncher launcher = serviceWithSink(herdr, configRoot, discRoot, cfg, sink);
|
||||
SessionManager sessions = new SessionManager(launcher);
|
||||
|
||||
// The mismatching row exists BEFORE the spawn — reproducing fleetd #234's exact timing:
|
||||
// opencode's session table already carries evidence by the moment acquire() first asks.
|
||||
OpenCodeSessionDiscoveryTest.writeRecord(discRoot, "ses_x", "/work/dir", 1000L,
|
||||
"{\"id\":\"gpt-5.6-sol\",\"providerID\":\"openai\"}");
|
||||
|
||||
assertFalse(quarantine.isQuarantined("openai-shared"), "nothing quarantined before the spawn");
|
||||
|
||||
// The real production entrypoint: acquire() builds the MemberSession by calling
|
||||
// handle.agentSessionId() BEFORE registry.put() runs.
|
||||
MemberSession acquired = sessions.acquire(cfg.profile(), "/work/dir", null, null);
|
||||
|
||||
assertEquals("ses_x", acquired.agentSessionId(), "the id itself still resolves correctly");
|
||||
assertTrue(quarantine.isQuarantined("openai-shared"),
|
||||
"the mismatch fires DURING acquire(), before roster registration, and must still "
|
||||
+ "reach the quarantine via the profile hint — not silently no-op");
|
||||
}
|
||||
|
||||
/**
|
||||
* The other half of the same proof: a sink that resolves {@code target -> profile} ONLY
|
||||
* through the roster (i.e. ignores the profile hint entirely, ~today's pre-fix {@code
|
||||
* Fleetd.java}) drops the SAME spawn-time mismatch silently — the credential is never
|
||||
* quarantined even though {@link OpenCodeLauncher} logged the mismatch ERROR. This is the
|
||||
* failure fleetd #234 reported, reproduced through the real {@code SessionManager.acquire()}
|
||||
* path rather than asserted by inspecting the fix.
|
||||
*/
|
||||
@Test
|
||||
void aRosterOnlySinkSilentlyDropsTheSpawnTimeQuarantine(@TempDir Path configRoot,
|
||||
@TempDir Path discRoot) throws Exception {
|
||||
FleetConfig.Profile cfg = opencodeCfgWithCredential(
|
||||
"terra", "opencode/nemotron-3-ultra-free", "openai-shared");
|
||||
BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.SECONDS.toNanos(1800));
|
||||
|
||||
// Deliberately ignores the profile hint — the pre-fix shape: only a roster lookup (modelled
|
||||
// here as always empty, since acquire() has not registered the session yet either way).
|
||||
ExhaustionSink rosterOnlySink = (target, reason, profile) -> { };
|
||||
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
OpenCodeLauncher launcher = serviceWithSink(herdr, configRoot, discRoot, cfg, rosterOnlySink);
|
||||
SessionManager sessions = new SessionManager(launcher);
|
||||
|
||||
OpenCodeSessionDiscoveryTest.writeRecord(discRoot, "ses_x", "/work/dir", 1000L,
|
||||
"{\"id\":\"gpt-5.6-sol\",\"providerID\":\"openai\"}");
|
||||
|
||||
MemberSession acquired = sessions.acquire(cfg.profile(), "/work/dir", null, null);
|
||||
|
||||
assertEquals("ses_x", acquired.agentSessionId(), "the id itself still resolves correctly");
|
||||
assertFalse(quarantine.isQuarantined("openai-shared"),
|
||||
"a roster-only sink cannot see this target yet — the quarantine silently never "
|
||||
+ "happens, which is exactly fleetd #234 defect 2");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #234, round 2: the two tests above inject a sink DIRECTLY into {@link OpenCodeLauncher},
|
||||
* which is not what {@code Fleetd.main} actually does. Production has one more hop: {@code
|
||||
* Fleetd.java} builds the adapters (including {@code OpenCodeLauncher}) before {@code sessions}
|
||||
* exists — a genuine construction-order cycle — so it hands the launcher a <em>forwarding</em>
|
||||
* sink pointed at an {@code AtomicReference<ExhaustionSink>}, and only later {@code .set(...)}s
|
||||
* that reference to the real sink once {@code sessions} is built.
|
||||
*
|
||||
* <p><strong>Round 3:</strong> a round-2 version of this test built its own copy of that
|
||||
* forwarder's shape as an anonymous class. Mutating {@code Fleetd.java}'s ACTUAL forwarder back
|
||||
* into a broken lambda left this test's own copy untouched, so it kept passing while production
|
||||
* had regressed to the exact bug being fixed. This version instead calls {@link
|
||||
* ExhaustionSink#forwardingTo}, the same factory {@code Fleetd.java} calls — the identical
|
||||
* object, not a rebuilt copy of its shape — so a regression at either the {@code Fleetd.java}
|
||||
* call site or inside {@code forwardingTo} itself has nowhere left to hide. See {@code
|
||||
* ExhaustionSinkForwardingHazardTest} for the same factory exercised in isolation.
|
||||
*/
|
||||
@Test
|
||||
void theSpawnTimeQuarantineSurvivesTheFleetdStyleForwardingHop(@TempDir Path configRoot,
|
||||
@TempDir Path discRoot) throws Exception {
|
||||
FleetConfig.Profile cfg = opencodeCfgWithCredential(
|
||||
"terra", "opencode/nemotron-3-ultra-free", "openai-shared");
|
||||
Map<String, FleetConfig.Profile> profiles = Map.of(cfg.profile(), cfg);
|
||||
BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.SECONDS.toNanos(1800));
|
||||
|
||||
// Fleetd.java:176 — the forwarding sink is built BEFORE the real one can exist, and the
|
||||
// launcher below is constructed against this forwarder, exactly like Fleetd.main. Calling
|
||||
// the SAME factory Fleetd.java calls — ExhaustionSink.forwardingTo — rather than rebuilding
|
||||
// the forwarder's shape here is the whole point (fleetd #234, round 3): a round-2 version of
|
||||
// this test built its own copy, so mutating Fleetd.java's real forwarder back into a lambda
|
||||
// left this test untouched. Routed through the shared factory, a regression at either the
|
||||
// Fleetd.java call site (reverting to a hand-written lambda) or inside the factory body
|
||||
// itself has nowhere left to hide from this test.
|
||||
java.util.concurrent.atomic.AtomicReference<ExhaustionSink> exhaustionSinkRef =
|
||||
new java.util.concurrent.atomic.AtomicReference<>(ExhaustionSink.none());
|
||||
ExhaustionSink forwardingExhaustionSink = ExhaustionSink.forwardingTo(exhaustionSinkRef::get);
|
||||
|
||||
FakeHerdr herdr = new FakeHerdr();
|
||||
OpenCodeLauncher launcher = serviceWithSink(herdr, configRoot, discRoot, cfg, forwardingExhaustionSink);
|
||||
SessionManager sessions = new SessionManager(launcher);
|
||||
|
||||
// Fleetd.java:359-390 — the real sink is only built and pointed to AFTER `sessions` exists,
|
||||
// same order as production. Safely a lambda (round 4): see the note on the forwarder above.
|
||||
ExhaustionSink realSink = (target, reason, profileHint) -> {
|
||||
FleetConfig.Profile profile = profileHint == null ? null : profiles.get(profileHint);
|
||||
if (profile != null) {
|
||||
quarantine.quarantine(profile.effectiveCredentialId());
|
||||
}
|
||||
};
|
||||
exhaustionSinkRef.set(realSink);
|
||||
|
||||
OpenCodeSessionDiscoveryTest.writeRecord(discRoot, "ses_x", "/work/dir", 1000L,
|
||||
"{\"id\":\"gpt-5.6-sol\",\"providerID\":\"openai\"}");
|
||||
|
||||
assertFalse(quarantine.isQuarantined("openai-shared"), "nothing quarantined before the spawn");
|
||||
|
||||
MemberSession acquired = sessions.acquire(cfg.profile(), "/work/dir", null, null);
|
||||
|
||||
assertEquals("ses_x", acquired.agentSessionId(), "the id itself still resolves correctly");
|
||||
assertTrue(quarantine.isQuarantined("openai-shared"),
|
||||
"the profile hint must survive the Fleetd-style forwarding hop and reach the real "
|
||||
+ "sink — a lambda forwarder drops it and this must go red");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -146,14 +146,14 @@ class OpenCodeSessionDiscoveryTest {
|
||||
"an unreadable database resolves to null, not an exception");
|
||||
}
|
||||
|
||||
// --- fleetd #175: actualModelForDirectory / the model JSON column ---------------------------
|
||||
// --- fleetd #175/#234: actualModelForSessionId / the model JSON column -----------------------
|
||||
|
||||
@Test
|
||||
void parsesTheModelJsonIntoProviderAndId(@TempDir Path root) throws Exception {
|
||||
writeRecord(root, "ses_aaa", "/w/a", 1000L, "{\"id\":\"gpt-5.6-terra\",\"providerID\":\"openai\"}");
|
||||
|
||||
OpenCodeSessionDiscovery.ActualModel actual =
|
||||
new OpenCodeSessionDiscovery(root).actualModelForDirectory("/w/a");
|
||||
new OpenCodeSessionDiscovery(root).actualModelForSessionId("ses_aaa");
|
||||
|
||||
assertNotNull(actual, "a well-formed model JSON parses");
|
||||
assertEquals("gpt-5.6-terra", actual.id());
|
||||
@@ -161,28 +161,39 @@ class OpenCodeSessionDiscoveryTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
void prefersTheModelOfTheMostRecentlyUpdatedRow(@TempDir Path root) throws Exception {
|
||||
writeRecord(root, "ses_old", "/w/a", 1000L, "{\"id\":\"old-model\",\"providerID\":\"openai\"}");
|
||||
writeRecord(root, "ses_new", "/w/a", 5000L, "{\"id\":\"new-model\",\"providerID\":\"openai\"}");
|
||||
void looksUpByIdEvenWhenAnotherRowInTheSameDirectoryIsNewer(@TempDir Path root) throws Exception {
|
||||
writeRecord(root, "ses_ours", "/work/dir", 1000L, "{\"id\":\"gpt-5.6-terra\",\"providerID\":\"openai\"}");
|
||||
writeRecord(root, "ses_sibling", "/work/dir", 9000L, "{\"id\":\"deepseek-v4-flash\",\"providerID\":\"gx\"}");
|
||||
|
||||
assertEquals("new-model",
|
||||
new OpenCodeSessionDiscovery(root).actualModelForDirectory("/w/a").id(),
|
||||
"the model of the row with the highest time_updated wins, same as the id");
|
||||
OpenCodeSessionDiscovery discovery = new OpenCodeSessionDiscovery(root);
|
||||
assertEquals("gpt-5.6-terra", discovery.actualModelForSessionId("ses_ours").id(),
|
||||
"querying by id reads OUR row, not the directory's newest row");
|
||||
assertEquals("deepseek-v4-flash", discovery.actualModelForSessionId("ses_sibling").id(),
|
||||
"each id resolves to its own row independently of time_updated ordering");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aNonMatchingDirectoryYieldsUnknownModelRatherThanAMismatch(@TempDir Path root) throws Exception {
|
||||
void anUnknownSessionIdYieldsUnknownModelRatherThanAMismatch(@TempDir Path root) throws Exception {
|
||||
writeRecord(root, "ses_aaa", "/w/a", 1000L, "{\"id\":\"x\",\"providerID\":\"y\"}");
|
||||
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForDirectory("/w/other"),
|
||||
"no row for this cwd yet → unknown, not a wrong model");
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForSessionId("ses_no_such_row"),
|
||||
"no row for this id yet → unknown, not a wrong model");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aBlankOrNullSessionIdYieldsUnknownModel(@TempDir Path root) throws Exception {
|
||||
writeRecord(root, "ses_aaa", "/w/a", 1000L, "{\"id\":\"x\",\"providerID\":\"y\"}");
|
||||
|
||||
OpenCodeSessionDiscovery discovery = new OpenCodeSessionDiscovery(root);
|
||||
assertNull(discovery.actualModelForSessionId(null));
|
||||
assertNull(discovery.actualModelForSessionId(" "));
|
||||
}
|
||||
|
||||
@Test
|
||||
void aNullModelColumnYieldsUnknownWithoutThrowing(@TempDir Path root) throws Exception {
|
||||
writeRecord(root, "ses_aaa", "/w/a", 1000L, null);
|
||||
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForDirectory("/w/a"),
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForSessionId("ses_aaa"),
|
||||
"a row with no model value yet is unknown, not a mismatch");
|
||||
}
|
||||
|
||||
@@ -190,7 +201,7 @@ class OpenCodeSessionDiscoveryTest {
|
||||
void unparseableModelJsonYieldsUnknownWithoutThrowing(@TempDir Path root) throws Exception {
|
||||
writeRecord(root, "ses_aaa", "/w/a", 1000L, "this is not json");
|
||||
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForDirectory("/w/a"),
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForSessionId("ses_aaa"),
|
||||
"JSON that fails to parse resolves to unknown, never an exception");
|
||||
}
|
||||
|
||||
@@ -198,13 +209,13 @@ class OpenCodeSessionDiscoveryTest {
|
||||
void modelJsonMissingIdYieldsUnknown(@TempDir Path root) throws Exception {
|
||||
writeRecord(root, "ses_aaa", "/w/a", 1000L, "{\"providerID\":\"openai\"}");
|
||||
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForDirectory("/w/a"),
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForSessionId("ses_aaa"),
|
||||
"no id in the JSON → unknown, since id is what a caller actually compares");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aMissingDatabaseYieldsUnknownModelWithoutThrowing(@TempDir Path root) {
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForDirectory("/w/a"));
|
||||
assertNull(new OpenCodeSessionDiscovery(root).actualModelForSessionId("ses_aaa"));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -212,7 +223,7 @@ class OpenCodeSessionDiscoveryTest {
|
||||
* shape a real opencode upgrade/downgrade could produce. This must degrade to UNKNOWN for the
|
||||
* model, and — the property that actually matters — must NOT take id resolution down with it.
|
||||
* A combined single query for both columns would fail this test; that is why
|
||||
* {@link OpenCodeSessionDiscovery#actualModelForDirectory} runs its own independent query.
|
||||
* {@link OpenCodeSessionDiscovery#actualModelForSessionId} runs its own independent query.
|
||||
*/
|
||||
@Test
|
||||
void aMissingModelColumnYieldsUnknownButIdResolutionStillWorks(@TempDir Path root) throws Exception {
|
||||
@@ -234,7 +245,7 @@ class OpenCodeSessionDiscoveryTest {
|
||||
OpenCodeSessionDiscovery discovery = new OpenCodeSessionDiscovery(root);
|
||||
assertEquals("ses_aaa", discovery.sessionIdForDirectory("/w/a"),
|
||||
"id resolution must survive a database with no model column at all");
|
||||
assertNull(discovery.actualModelForDirectory("/w/a"),
|
||||
assertNull(discovery.actualModelForSessionId("ses_aaa"),
|
||||
"no model column → unknown, not a throw and not a mismatch");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user