diff --git a/fleetd/src/main/java/dev/ltms/fleet/Fleetd.java b/fleetd/src/main/java/dev/ltms/fleet/Fleetd.java index 74f4c0f..4bd2bfa 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/Fleetd.java +++ b/fleetd/src/main/java/dev/ltms/fleet/Fleetd.java @@ -528,8 +528,11 @@ public final class Fleetd { } }; Predicate deliverable = deliverableTo(presence, leads); + // fleetd #556: registration is wired directly to `completion`, not folded into the + // `turnListener` fan-out above — so it survives `sessions.onDelivered` (or any future + // listener) throwing, regardless of call order. See TurnRegistrar's javadoc. Injector injector = new Injector(router, turnListener, deliverable, - presence::forget); + presence::forget, completion::register); StatusPoller poller = new StatusPoller(router, injector, Injector.POLL_INTERVAL_MILLIS); poller.start(); diff --git a/fleetd/src/main/java/dev/ltms/fleet/inject/CompletionResolver.java b/fleetd/src/main/java/dev/ltms/fleet/inject/CompletionResolver.java index d1faa05..baa07c5 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/inject/CompletionResolver.java +++ b/fleetd/src/main/java/dev/ltms/fleet/inject/CompletionResolver.java @@ -50,7 +50,7 @@ import java.util.regex.Pattern; * so the scrape's herdr round-trip never stalls the status poller. The captured waiter is read on the * poller thread (before any next-turn delivery can overwrite it) and passed into the virtual thread. */ -public final class CompletionResolver implements TurnListener { +public final class CompletionResolver implements TurnListener, TurnRegistrar { private static final Logger log = LoggerFactory.getLogger(CompletionResolver.class); @@ -235,6 +235,26 @@ public final class CompletionResolver implements TurnListener { this.worktreeBranches = Objects.requireNonNull(worktreeBranches, "worktreeBranches"); } + /** + * fleetd #556: {@link TurnRegistrar}'s structural half of what {@link #captureBaseline} used to + * do alone — record the waiter, no I/O. Called directly and unconditionally by the {@link + * Injector} as part of delivering a turn, before {@link #onDelivered} ever runs, so this + * registration cannot be skipped by a {@link TurnListener} throwing (from {@code onDelivered} or + * any other callback). {@link #captureBaseline} still performs this exact check-and-put itself + * as well — harmless and idempotent when it runs right after this — so a caller that only wires + * the {@link TurnListener} path (e.g. an existing test that never mentions {@link TurnRegistrar}) + * keeps working unchanged. + */ + @Override + public void register(String target, TurnToken token) { + CompletableFuture waiter = token.waiter(); + if (waiter == null) { + inFlight.remove(target); // no send is waiting on this delivery — nothing to resolve later + return; + } + inFlight.put(target, new InFlight(waiter, null, nowNanos.getAsLong(), token.injectedText())); + } + @Override public void onDelivered(String target, TurnToken token) { // Capture the exact waiter this turn belongs to (CB-116) and snapshot the pane's pre-turn @@ -244,7 +264,14 @@ public final class CompletionResolver implements TurnListener { captureBaseline(target, token); } - /** Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link #onDelivered}). */ + /** + * Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link + * #onDelivered}). fleetd #556: this still does its own full waiter check and {@code inFlight.put} + * — the same registration {@link #register} performs — so it keeps working standalone (as every + * test calling it directly already does) even though the {@link Injector} now also calls {@link + * #register} on its own, earlier and unconditionally. The two writes are idempotent with each + * other; only the second (this one) carries a real baseline, since only this one pays for a scrape. + */ void captureBaseline(String target, TurnToken token) { CompletableFuture waiter = token.waiter(); if (waiter == null) { diff --git a/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java b/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java index f26c82a..bf79d8a 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java +++ b/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java @@ -11,6 +11,7 @@ import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Deque; import java.util.List; +import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -93,6 +94,11 @@ public final class Injector { private final AgentControl agents; private final HerdrRouter router; private final TurnListener turnListener; + /** + * fleetd #556: the Injector's own registration invariant, called directly and unconditionally — + * never folded into {@link #turnListener}'s fan-out. See {@link TurnRegistrar}. + */ + private final TurnRegistrar registrar; private final Predicate ready; // CB-113: a target is deliverable only when available private final Consumer forget; // CB-114: clear a gone worker's readiness/presence /** @@ -134,15 +140,43 @@ public final class Injector { */ public Injector(AgentControl agents, TurnListener turnListener, Predicate ready, Consumer forget) { - this(agents, turnListener, ready, forget, System::currentTimeMillis); + // fleetd #556: no explicit registrar named here, so fall back to turnListener itself when it + // happens to also implement TurnRegistrar (true for CompletionResolver, the only production + // TurnListener that needs CB-106 registration) — every existing caller of this overload keeps + // working unchanged. A turnListener that does NOT implement TurnRegistrar (a fan-out object, + // or a bare test lambda) gets TurnRegistrar.NOOP here, same as before this ticket. + this(agents, turnListener, ready, forget, + turnListener instanceof TurnRegistrar r ? r : TurnRegistrar.NOOP); } - /** Full constructor — for tests: an injectable wall-clock supplier (fleetd #501). */ + /** + * fleetd #556: explicit-registrar overload. Use this whenever {@code turnListener} does not + * itself implement {@link TurnRegistrar} — e.g. a fan-out object that also notifies unrelated + * observers — so registration is wired directly to the one component that owns it, rather than + * relying on {@code turnListener} happening to implement both interfaces. + */ + public Injector(AgentControl agents, TurnListener turnListener, Predicate ready, + Consumer forget, TurnRegistrar registrar) { + this(agents, turnListener, ready, forget, registrar, System::currentTimeMillis); + } + + /** + * Package-private constructor — for tests: an injectable wall-clock supplier (fleetd #501), + * with no explicit registrar (fleetd #556 fallback, same as the public 4-arg overload above). + */ Injector(AgentControl agents, TurnListener turnListener, Predicate ready, Consumer forget, LongSupplier nowMillis) { + this(agents, turnListener, ready, forget, + turnListener instanceof TurnRegistrar r ? r : TurnRegistrar.NOOP, nowMillis); + } + + /** Full constructor — for tests: an explicit registrar and an injectable wall-clock supplier. */ + Injector(AgentControl agents, TurnListener turnListener, Predicate ready, + Consumer forget, TurnRegistrar registrar, LongSupplier nowMillis) { this.agents = agents; this.router = null; this.turnListener = turnListener; + this.registrar = Objects.requireNonNull(registrar, "registrar"); this.ready = ready; this.forget = forget; this.nowMillis = nowMillis; @@ -150,15 +184,28 @@ public final class Injector { public Injector(HerdrRouter router, TurnListener turnListener, Predicate ready, Consumer forget) { - this(router, turnListener, ready, forget, System::currentTimeMillis); + // fleetd #556: see the AgentControl overload above for the fallback rationale. + this(router, turnListener, ready, forget, + turnListener instanceof TurnRegistrar r ? r : TurnRegistrar.NOOP); + } + + /** + * fleetd #556: explicit-registrar overload — production wiring ({@code Fleetd}) uses this, since + * its {@code turnListener} is a fan-out object that also notifies {@code SessionManager} and does + * not itself implement {@link TurnRegistrar}. + */ + public Injector(HerdrRouter router, TurnListener turnListener, Predicate ready, + Consumer forget, TurnRegistrar registrar) { + this(router, turnListener, ready, forget, registrar, System::currentTimeMillis); } /** Full constructor — for tests: an injectable wall-clock supplier (fleetd #501). */ Injector(HerdrRouter router, TurnListener turnListener, Predicate ready, - Consumer forget, LongSupplier nowMillis) { + Consumer forget, TurnRegistrar registrar, LongSupplier nowMillis) { this.agents = null; this.router = router; this.turnListener = turnListener; + this.registrar = Objects.requireNonNull(registrar, "registrar"); this.ready = ready; this.forget = forget; this.nowMillis = nowMillis; @@ -506,20 +553,25 @@ public final class Injector { // // There are TWO futures at stake here, not one: `sent.delivered()` (the delivery future) // and `sent.token().waiter()` (the rendezvous waiter a blocking fleet_send actually waits - // on for the worker's ANSWER). The waiter is registered only by turnListener.onDelivered(), - // called from inside the `if (sent != null)` block below. Completing `sent.delivered()` - // without also calling onDelivered leaves the waiter unregistered forever — a hang with a - // success receipt, which is worse than the plain hang this ticket is about. So the - // `finally` below does not just complete the future: on the path where the `if (sent != - // null)` block never ran, it does that block's whole job — onDelivered, then complete. + // on for the worker's ANSWER). fleetd #556: the waiter is registered by `registrar.register` + // now — called directly and unconditionally, structural rather than delegated to a listener + // callback — never by turnListener.onDelivered() alone (before this ticket, that WAS the + // only path, and any TurnListener wired ahead of the one that mattered could throw and skip + // it; #553 could only make the reachable listener behave, not remove that dependency). + // Completing `sent.delivered()` without also calling `registrar.register` leaves the waiter + // unregistered forever — a hang with a success receipt, which is worse than the plain hang + // #553 was about. So the `finally` below does not just complete the future: on the path + // where the `if (sent != null)` block never ran, it does that block's whole job — + // register, then onDelivered (notification only, may throw), then complete. // // `sentHandled` is NOT allowed to move earlier than this: an earlier comment on #553 // proposed running `if (sent != null)` first, before turnCompleted/turnFailed, and that - // was withdrawn — onDelivered WRITES CompletionResolver's inFlight record for this turn, - // onTurnComplete READS it for the PREVIOUS turn, and running onDelivered first makes + // was withdrawn — `registrar.register` WRITES CompletionResolver's inFlight record for this + // turn, onTurnComplete READS it for the PREVIOUS turn, and running register first makes // onTurnComplete resolve the NEW turn's waiter with the PREVIOUS turn's stale output // (CB-116). The `if (sent != null)` block stays last; the `finally` is a backstop for it, - // not a replacement. + // not a replacement. fleetd #556 keeps this ordering unchanged — it only splits what used + // to be one listener call (register + notify) into two calls in the same position. boolean sentHandled = false; try { if (resubmit) { @@ -585,13 +637,13 @@ public final class Injector { } if (sent != null) { // fleetd #553: record the INTENT to handle before any side effect, not the fact of - // having completed it. If onDelivered() throws part way through — e.g. after - // CompletionResolver's captureBaseline has already done its inFlight.put — a flag - // set only after this block would still read false, and the `finally` below would - // call onDelivered() a SECOND time. A second captureBaseline runs later, once - // onTurnComplete has already thrown and possibly scraped the pane, and can snapshot - // a pane that already absorbed this turn's output — which means the pane's tail - // never differs from that baseline again and CompletionResolver.resolve's own + // having completed it. If register()/onDelivered() throws part way through — e.g. + // after CompletionResolver's captureBaseline has already done its inFlight.put — a + // flag set only after this block would still read false, and the `finally` below + // would re-run this block's job a SECOND time. A second captureBaseline runs later, + // once onTurnComplete has already thrown and possibly scraped the pane, and can + // snapshot a pane that already absorbed this turn's output — which means the pane's + // tail never differs from that baseline again and CompletionResolver.resolve's own // suppression at :364-369 (`return` keeping the in-flight record) drops every future // completion for this turn, permanently. So this flag must go true FIRST. sentHandled = true; @@ -599,32 +651,41 @@ public final class Injector { log.warn("inject to {} failed, dropped message: {}", target, sendError.getMessage()); sent.delivered().completeExceptionally(sendError); } else { - // Baseline the pane's pre-turn content so a misattributed completion (no new output) - // can't resolve this send with the previous turn's stale answer (CB-115). + // fleetd #556: register the waiter FIRST and unconditionally — this is the + // Injector's own invariant, kept structural rather than delegated. Even if + // turnListener.onDelivered below throws (a bug in an unrelated observer, e.g. + // SessionManager's bookkeeping, or a fan-out wired in a different order), the + // waiter is already registered and resolvable — a throwing TurnListener can no + // longer take this invariant down with it. + registrar.register(target, sent.token()); + // Notification only, from here down: baseline the pane's pre-turn content so a + // misattributed completion (no new output) can't resolve this send with the + // previous turn's stale answer (CB-115). Allowed to throw and to fail (the herdr + // scrape inside already fails open with baseline == null on its own). turnListener.onDelivered(target, sent.token()); sent.delivered().complete(null); } } } finally { - // Backstop (fleetd #553): if an earlier block in this try threw before the `if (sent != - // null)` block above ran, `sentHandled` is still false here, and this does that block's - // WHOLE job — not just the future completion. Skipping onDelivered() here would leave - // sent.token().waiter() never registered with CompletionResolver, so a blocking - // fleet_send would be told its message was delivered and then wait out its full timeout - // for an answer that can never resolve — worse than the plain hang, because it now looks - // like success. onDelivered() runs only when sendError == null: nothing was delivered on - // the error path, so there is nothing to register. + // Backstop (fleetd #553, extended #556): if an earlier block in this try threw before the + // `if (sent != null)` block above ran, `sentHandled` is still false here, and this does + // that block's WHOLE job — not just the future completion. Skipping registrar.register() + // here would leave sent.token().waiter() never registered with CompletionResolver, so a + // blocking fleet_send would be told its message was delivered and then wait out its full + // timeout for an answer that can never resolve — worse than the plain hang, because it now + // looks like success. This block runs only when sendError == null: nothing was delivered + // on the error path, so there is nothing to register. // - // fleetd #553 (lead review, comment 16916): `sentHandled` guards ONLY the onDelivered + // fleetd #553 (lead review, comment 16916): `sentHandled` guards ONLY the register/notify // re-call below, never the future completion outside this inner try. One boolean cannot - // carry both meanings — "onDelivered was called" and "the future has been dealt with" — + // carry both meanings — "the turn was handled" and "the future has been dealt with" — // because they come apart exactly when onDelivered throws PART WAY THROUGH: sentHandled // is already true (set before the call, correctly — see the `if (sent != null)` block // above), so a guard on the outer `if` here would skip this whole recovery, including the // completion, and leave sent.delivered() pending forever even though the message really // was typed into the pane and taken off the queue. So `sent != null` alone gates whether - // this target has anything to finish; `!sentHandled` gates only the onDelivered re-call - // inside. CompletableFuture.complete/completeExceptionally are idempotent — on the + // this target has anything to finish; `!sentHandled` gates only the register/notify + // re-call inside. CompletableFuture.complete/completeExceptionally are idempotent — on the // ordinary path (sentHandled == true, no throw) the `if (sent != null)` block above has // already completed this future, so the calls below are a no-op returning false. // @@ -635,6 +696,11 @@ public final class Injector { if (sent != null) { try { if (!sentHandled && sendError == null) { + // fleetd #556: register first here too, same as the ordinary path above — this + // recovery path is reached only when an EARLIER block (resubmit/turnCompleted/ + // turnFailed) threw before the ordinary path ever ran, so nothing has + // registered this turn yet. + registrar.register(target, sent.token()); turnListener.onDelivered(target, sent.token()); } } catch (Throwable recoveryError) { diff --git a/fleetd/src/main/java/dev/ltms/fleet/inject/TurnRegistrar.java b/fleetd/src/main/java/dev/ltms/fleet/inject/TurnRegistrar.java new file mode 100644 index 0000000..111b511 --- /dev/null +++ b/fleetd/src/main/java/dev/ltms/fleet/inject/TurnRegistrar.java @@ -0,0 +1,29 @@ +package dev.ltms.fleet.inject; + +import dev.ltms.fleet.msg.TurnToken; + +/** + * fleetd #556: the {@link Injector}'s own invariant — every delivered turn has a registered + * waiter — kept structural rather than delegated to a {@link TurnListener} callback that can + * throw. The {@link Injector} calls {@link #register} directly and unconditionally as part of + * delivering a turn, before it ever calls {@code turnListener.onDelivered}. A {@link TurnListener} + * that throws from every callback (a bug in an unrelated observer, e.g. session bookkeeping, or a + * fan-out object wired in the wrong order) can therefore never leave a delivered turn unregistered + * — the shape #553 could not close, since that fix could only make the reachable listener behave, + * never remove the Injector's dependency on a listener behaving at all. + * + *

Deliberately narrow: registration only, no I/O, nothing that scrapes a pane or can reasonably + * fail. The CB-115 staleness baseline (a herdr round-trip, allowed to fail) stays a + * {@link TurnListener#onDelivered} concern — fired by the {@link Injector} only after this call has + * already run, so its own failure cannot un-register anything. + */ +@FunctionalInterface +public interface TurnRegistrar { + + /** Record {@code token}'s waiter as the turn currently in flight for {@code target}. */ + void register(String target, TurnToken token); + + /** No-op registrar for callers that don't need CB-106 rendezvous tracking. */ + TurnRegistrar NOOP = (_, _) -> { + }; +} diff --git a/fleetd/src/test/java/dev/ltms/fleet/inject/CompletionResolverTest.java b/fleetd/src/test/java/dev/ltms/fleet/inject/CompletionResolverTest.java index 22dcb6b..2a0f670 100644 --- a/fleetd/src/test/java/dev/ltms/fleet/inject/CompletionResolverTest.java +++ b/fleetd/src/test/java/dev/ltms/fleet/inject/CompletionResolverTest.java @@ -19,6 +19,7 @@ import java.util.regex.Pattern; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; /** Unit behaviour of the CB-106 completion resolver in isolation from the injector. */ @@ -519,6 +520,111 @@ class CompletionResolverTest { assertTrue(rendezvous.isWaiting("term_a"), "turn N+1 is still awaiting its own resolution"); } + // --- fleetd #556 (comment 16908): name the invariant the two-arg remove(target, turn) idiom + // exists to maintain — a superseded turn's terminal handling must not evict its successor's + // registration — and test THAT, not the idiom's literal spelling. Three entry points reach a + // conditional remove: resolve()'s plain completion path, resolve()'s echoed-brief/noReportMessage + // sub-path, and fail(). This is a non-goal-to-break for fleetd #556's redesign: flattening any of + // these removes to the one-arg form must turn one of these three red. ------------------------- + + @Test + void aSupersededTurnsPlainCompletionMustNotEvictItsSuccessorsRegistration() { + // Turn A is overtaken by turn B on the same target (a rapid back-to-back send): B's own + // delivery overwrites the map entry A's delivery installed. A's late completion fallback + // then finally fires and reaches resolve()'s ordinary (non-echoed) completion branch — the + // two-arg remove(target, turnA) there must be a no-op once the map holds B, not a blind + // evict. + FakeHerdr herdr = new FakeHerdr().readText("⏺ A's pre-turn pane\n❯ "); + Rendezvous rendezvous = new Rendezvous(); + CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, + ExhaustedPatternLookup.none(), ExhaustionSink.none()); + + var waiterA = rendezvous.open("term_a"); + var turnA = new CompletionResolver.InFlight(waiterA, null); // baseline null: fails open, never suppressed + rendezvous.close("term_a", waiterA); // A's own send's finally, freeing the session for B + var waiterB = rendezvous.open("term_a"); + resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB)); // the injector registering B + + herdr.readText("⏺ A's late, genuine answer\n❯ "); // what the pane shows when A's fallback fires + resolver.resolve("term_a", turnA); + + assertTrue(waiterA.isDone(), "A's own resolve must still complete A's waiter normally"); + assertEquals(Rendezvous.Kind.COMPLETION, waiterA.getNow(null).kind()); + + CompletionResolver.InFlight afterA = resolver.inFlight("term_a"); + assertNotNull(afterA, "B's registration must still be present — a one-arg remove(target) " + + "here would evict it even though the map no longer holds turnA"); + assertEquals(waiterB, afterA.waiter(), "the surviving entry must still be B's, untouched by A's " + + "terminal handling"); + assertTrue(rendezvous.resolve("term_a", "B replied"), + "B must still resolve normally through the ordinary reply path"); + assertEquals(Rendezvous.Kind.REPLY, waiterB.getNow(null).kind()); + } + + @Test + void aSupersededTurnsEchoedNoReportSubPathMustNotEvictItsSuccessorsRegistration() { + // Same invariant, driven down resolve()'s OTHER branch: a pane that merely echoes the + // injected brief back (no real report) resolves through noReportMessage(), not the plain + // tail — comment 16908 calls this "a sub-path of the first, not a separate method", reaching + // the same terminal remove() call site from different logic above it. + String echoedBrief = "y".repeat(450); // >= CompletionResolver.ECHO_MIN_CHARS normalised chars + FakeHerdr herdr = new FakeHerdr().readText("⏺ " + echoedBrief + "\n❯ "); + Rendezvous rendezvous = new Rendezvous(); + CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous, + ExhaustedPatternLookup.none(), ExhaustionSink.none()); + + var waiterA = rendezvous.open("term_a"); + // Back-dated deliveredAtNanos (like the 2-arg InFlight convenience ctor) so MIN_TURN_NANOS + // never fires; injectedText == the pane's own content so echoesInjectedBrief() is true. + var turnA = new CompletionResolver.InFlight(waiterA, null, Long.MIN_VALUE / 2, echoedBrief); + rendezvous.close("term_a", waiterA); + var waiterB = rendezvous.open("term_a"); + resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB)); + + resolver.resolve("term_a", turnA); + + assertTrue(waiterA.isDone(), "A's own resolve must still complete A's waiter via the echoed " + + "no-report branch"); + assertTrue(waiterA.getNow(null).text().contains(CompletionResolver.NO_REPORT_PREFIX), + "sanity: A must actually have gone down the noReportMessage sub-path, not the plain one"); + + CompletionResolver.InFlight afterA = resolver.inFlight("term_a"); + assertNotNull(afterA, "B's registration must still be present after A's echoed-brief resolve"); + assertEquals(waiterB, afterA.waiter()); + assertTrue(rendezvous.resolve("term_a", "B replied"), + "B must still resolve normally through the ordinary reply path"); + assertEquals(Rendezvous.Kind.REPLY, waiterB.getNow(null).kind()); + } + + @Test + void aSupersededTurnsFailMustNotEvictItsSuccessorsRegistration() { + // The fail() entry point (CB-109 wedge / CB-110 drop): turn A is failed while turn B has + // already superseded it in the map. fail()'s own two-arg remove(target, turnA) must be a + // no-op once the map holds B. + Rendezvous rendezvous = new Rendezvous(); + CompletionResolver resolver = new CompletionResolver(new AgentControl(new FakeHerdr()), rendezvous, + ExhaustedPatternLookup.none(), ExhaustionSink.none()); + + var waiterA = rendezvous.open("term_a"); + var turnA = new CompletionResolver.InFlight(waiterA, null); + rendezvous.close("term_a", waiterA); + var waiterB = rendezvous.open("term_a"); + resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB)); + + resolver.fail("term_a", turnA, "A wedged in an unknown state"); + + assertTrue(waiterA.isDone(), "A's own fail must still complete A's waiter"); + assertEquals(Rendezvous.Kind.FAILED, waiterA.getNow(null).kind()); + + CompletionResolver.InFlight afterA = resolver.inFlight("term_a"); + assertNotNull(afterA, "B's registration must still be present — a one-arg remove(target) in " + + "fail() would evict it even though the map no longer holds turnA"); + assertEquals(waiterB, afterA.waiter()); + assertTrue(rendezvous.resolve("term_a", "B replied"), + "B must still resolve normally through the ordinary reply path"); + assertEquals(Rendezvous.Kind.REPLY, waiterB.getNow(null).kind()); + } + // --- CB-578 stage A: backend-exhausted classification --------------------------------- @Test diff --git a/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java b/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java index 7164db8..4efaf91 100644 --- a/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java +++ b/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java @@ -1197,4 +1197,166 @@ class InjectorTest { + "every path out of onStatus, including the one where onDelivered itself is " + "what threw. Leaving it pending here is a hang, not a fix"); } + + // --- fleetd #556: registration is the Injector's own invariant now, structural rather than + // delegated to a TurnListener callback that can throw. #553 could only make the ONE reachable + // listener behave; this closes the shape itself. --- + + @Test + void aTurnListenerThatThrowsFromEveryCallbackStillLeavesTheDeliveredTurnRegisteredAndResolvable() { + // fleetd #556 acceptance criterion 1 (the ticket's own, restated by comment 16966 as the + // replacement for #561's order-dependent test): a TurnListener that throws from EVERY + // callback it implements — nothing about it is safe to lean on — must still leave a + // delivered turn registered, with its waiter still resolvable through the ordinary path. + // Before this ticket, registration lived INSIDE onDelivered, one of the callbacks that + // throws here — this test is red against that shape, because the only listener callback + // this specific delivery scenario exercises (a first delivery, no previous turn to + // complete) is exactly the one that throws. + RuntimeException boom = new RuntimeException("fleetd #556: throws from every callback"); + TurnListener throwsEverywhere = new TurnListener() { + @Override + public void onTurnComplete(String target) { + throw boom; + } + + @Override + public boolean hasPostTurnAction(String target) { + throw boom; + } + + @Override + public boolean onTurnCompleteWithPostAction(String target) { + throw boom; + } + + @Override + public void onTurnFailed(String target) { + throw boom; + } + + @Override + public void onTurnFailed(String target, String reason) { + throw boom; + } + + @Override + public void onDelivered(String target, TurnToken token) { + throw boom; + } + }; + Rendezvous rendezvous = new Rendezvous(); + CompletionResolver completion = new CompletionResolver(new AgentControl(new FakeHerdr()), rendezvous, + ExhaustedPatternLookup.none(), ExhaustionSink.none()); + + Injector inj = new Injector(new AgentControl(herdr), throwsEverywhere, _ -> true, _ -> { + }, completion::register); + CompletableFuture waiter = rendezvous.open(T); + inj.enqueue(T, "brief", new TurnToken(T, waiter)); + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), // delivers "brief"; onDelivered throws + "the throwing listener's own bug must still escape onStatus — a listener bug stays " + + "loud (fleetd #553's own guarantee, unchanged by this ticket)"); + assertSame(boom, thrown); + + CompletionResolver.InFlight inFlight = completion.inFlight(T); + assertNotNull(inFlight, + "fleetd #556: the delivered turn must be registered even though the ONLY " + + "TurnListener wired throws from every single callback, including " + + "onDelivered — registration must not depend on that call succeeding"); + assertSame(waiter, inFlight.waiter(), + "the registered entry must carry the exact waiter this turn's send opened"); + + assertFalse(waiter.isDone(), "sanity: nothing has resolved this waiter yet"); + assertTrue(rendezvous.resolve(T, "hello"), + "the waiter registrar.register captured must be the real, live rendezvous waiter — " + + "still resolvable through the ordinary fleet_reply path, unaffected by the " + + "throwing listener"); + assertEquals(Rendezvous.Kind.REPLY, waiter.getNow(null).kind()); + } + + @Test + void theNewTurnsRegistrationRunsAfterThePreviousTurnsCompletionIsRead() { + // fleetd #556 acceptance criterion 2 (CB-116 guard, ticket body + comment 16966): the + // ordering constraint from fleetd #553 must survive this redesign — onTurnComplete reads + // CompletionResolver's inFlight entry for the PREVIOUS turn, so the new turn's + // registrar.register() call must not run before that read, or a redesign trades this + // ticket's defect for the CB-116 cross-turn stale reply. Pinned here as a call-ORDER + // assertion so a future redesign that hoists registrar.register() earlier (e.g. "for + // simplicity", ahead of the turnCompleted block) goes red immediately, before it can ever + // reach a real cross-turn scrape race. + List events = new ArrayList<>(); + TurnListener listener = new TurnListener() { + @Override + public void onTurnComplete(String target) { + events.add("onTurnComplete:" + target); + } + }; + TurnRegistrar registrar = (target, token) -> events.add("register:" + target); + + Injector inj = new Injector(new AgentControl(herdr), listener, _ -> true, _ -> { + }, registrar); + inj.enqueue(T, "first", TestTurnTokens.inert(T)); + inj.enqueue(T, "second", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); // delivers "first" -> register:term_a (not asserted below) + inj.onStatus(T, AgentStatus.WORKING); // picked up + events.clear(); + inj.onStatus(T, AgentStatus.IDLE); // "first" completes AND "second" delivers, same cycle + + assertEquals(List.of("onTurnComplete:" + T, "register:" + T), events, + "onTurnComplete for the previous turn (\"first\") must run before register for the " + + "new turn (\"second\") within the same onStatus cycle — reversing this order " + + "is the CB-116 cross-turn stale reply fleetd #553 fixed, and this redesign " + + "must not reopen it"); + } + + @Test + void aRuntimeExceptionFromOnTurnCompleteStillLeavesTheNextDeliveryRegisteredOnTheRecoveryPath() { + // fleetd #556 rework (PR #566, comment 17009): there are TWO registrar.register call sites + // in the delivery method — the ordinary path inside `if (sent != null)`, and the fleetd + // #553 `finally` backstop, reached only when an earlier block (here, onTurnComplete for the + // PREVIOUS turn) throws before the ordinary path ever runs. The existing #553 regression + // test for this exact scenario + // (aRuntimeExceptionFromOnTurnCompleteStillCompletesTheNextDelivery, above) asserts only + // that "second"'s DELIVERED FUTURE completes — never that "second" is REGISTERED with + // CompletionResolver — so a redesign that dropped registrar.register() from the backstop + // passed every existing test while reopening this ticket's own defect on the one path + // fleetd #553 exists for. This test closes that gap: on the recovery path, the backstop is + // the ONLY thing that registers "second", so its waiter must still be resolvable afterward. + RuntimeException boom = new RuntimeException("fleetd #556 rework: boom from onTurnComplete"); + TurnListener throwing = new TurnListener() { + @Override + public void onTurnComplete(String target) { + throw boom; + } + }; + Rendezvous rendezvous = new Rendezvous(); + CompletionResolver completion = new CompletionResolver(new AgentControl(herdr), rendezvous, + ExhaustedPatternLookup.none(), ExhaustionSink.none()); + Injector inj = new Injector(new AgentControl(herdr), throwing, _ -> true, _ -> { + }, completion::register); + + inj.enqueue(T, "first", TestTurnTokens.inert(T)); + CompletableFuture secondWaiter = rendezvous.open(T); + inj.enqueue(T, "second", new TurnToken(T, secondWaiter)); + + inj.onStatus(T, AgentStatus.IDLE); // delivers "first" (inert token — nothing to register) + inj.onStatus(T, AgentStatus.WORKING); // picked up + + RuntimeException thrown = assertThrows(RuntimeException.class, + // "first" completes (onTurnComplete throws) before "second"'s ordinary delivery + // path ever runs; only the finally backstop is left to register "second". + () -> inj.onStatus(T, AgentStatus.IDLE), + "the throwable from onTurnComplete must still escape onStatus"); + assertSame(boom, thrown); + + CompletionResolver.InFlight inFlight = completion.inFlight(T); + assertNotNull(inFlight, + "fleetd #556: \"second\" must be registered by the fleetd #553 finally backstop " + + "even though onTurnComplete threw for \"first\"'s completion before the " + + "ordinary registration path ever ran for \"second\""); + assertSame(secondWaiter, inFlight.waiter(), + "the registered entry must carry \"second\"'s own waiter, not some other value"); + } }