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 6dd05bd..638d0c9 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java +++ b/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java @@ -157,6 +157,7 @@ public final class Injector { boolean awaitingCompletion; // a delivered message's turn is not yet known-complete boolean turnObserved; // saw a real `working` sample since that delivery (turn ran) int unknownSinceTurn; // consecutive `unknown` samples while a delegation is outstanding (CB-109) + int unknownSincePostTurn; // the same, for the post-turn housekeeping phase (fleetd #306) int notReadySincePoll; // consecutive injectable samples a queued message waited on the readiness gate (CB-114) boolean postTurnPending; // completion observed; adapter housekeeping has not started yet boolean awaitingPostTurnPickup; @@ -217,10 +218,12 @@ public final class Injector { t.awaitingPickup = false; t.injectableSincePickup = 0; t.unknownSinceTurn = 0; + t.unknownSincePostTurn = 0; t.notReadySincePoll = 0; if (t.awaitingCompletion) t.turnObserved = true; } else if (status.injectable()) { // IDLE or BLOCKED t.unknownSinceTurn = 0; + t.unknownSincePostTurn = 0; if (t.awaitingPostTurnPickup) { if (++t.injectableSincePostTurnPickup >= PICKUP_GRACE_POLLS) { t.awaitingPostTurnPickup = false; @@ -315,6 +318,24 @@ public final class Injector { t.unknownSinceTurn = 0; turnFailed = true; } + // fleetd #306: the same escape for the post-turn housekeeping phase. Four latches + // gate delivery (awaitingCompletion, postTurnPending, awaitingPostTurnPickup, + // postTurnObserved) and only the first had a way out of a sustained unknown streak — + // a gate that closed one direction only. The other two below are released here as + // well; postTurnPending needs no escape because it is cleared unconditionally on the + // line after the listener call that sets it. + // + // This does NOT set turnFailed. The delegated turn already completed and its waiter + // already resolved — what is outstanding is adapter housekeeping (the `/clear`). + // Reporting a turn failure here would drive SessionManager.onFailed on a session + // that genuinely finished its work, which is a worse lie than the wedge. + if ((t.awaitingPostTurnPickup || t.postTurnObserved) + && ++t.unknownSincePostTurn >= TURN_STALL_GRACE_POLLS) { + t.awaitingPostTurnPickup = false; + t.postTurnObserved = false; + t.injectableSincePostTurnPickup = 0; + t.unknownSincePostTurn = 0; + } } // Reclaim the entry once the worker is fully quiescent (nothing queued, no pickup or 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 54d87fe..72e0c90 100644 --- a/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java +++ b/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java @@ -297,6 +297,68 @@ class InjectorTest { assertTrue(inj.activeTargets().isEmpty(), "the wedged target is reclaimed, not polled forever"); } + /** A listener whose post-turn housekeeping always starts, as SessionManager's does with clearAfterTurn on. */ + private static final class PostTurnListener implements TurnListener { + @Override public void onTurnComplete(String target) { } + @Override public boolean hasPostTurnAction(String target) { return true; } + @Override public boolean onTurnCompleteWithPostAction(String target) { return true; } + } + + @Test + void aWorkerThatWedgesInUnknownAwaitingPostTurnPickupIsReleased() { + // fleetd #306: the post-turn phase had no way out of a sustained unknown streak, so the + // pickup latch stayed set, the target was polled forever, and every later message to it was + // blocked by the delivery gate — while the session still looked healthy. + Captor cap = new Captor(); + Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener()); + inj.enqueue(T, "task", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); // deliver + inj.onStatus(T, AgentStatus.WORKING); // turn starts + inj.onStatus(T, AgentStatus.IDLE); // turn completes; housekeeping dispatched + for (int i = 0; i < STALL_SAMPLES; i++) inj.onStatus(T, AgentStatus.UNKNOWN); // then wedges + + assertTrue(inj.activeTargets().isEmpty(), + "a target wedged awaiting post-turn pickup must be reclaimed, not polled forever"); + assertEquals(List.of(), cap.failed, + "the delegated turn already completed — a stuck /clear must not be reported as a failed turn"); + } + + @Test + void aWorkerThatWedgesInUnknownAfterPickingUpTheResetIsReleased() { + // The sibling latch. postTurnObserved is set when the reset is seen picked up (WORKING) and + // is cleared only on a later injectable sample, so a wedge right after pickup sticks too. + Captor cap = new Captor(); + Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener()); + inj.enqueue(T, "task", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); + inj.onStatus(T, AgentStatus.WORKING); + inj.onStatus(T, AgentStatus.IDLE); // turn complete; reset dispatched + inj.onStatus(T, AgentStatus.WORKING); // reset picked up -> postTurnObserved + for (int i = 0; i < STALL_SAMPLES; i++) inj.onStatus(T, AgentStatus.UNKNOWN); + + assertTrue(inj.activeTargets().isEmpty(), "a wedge after reset pickup must also be reclaimed"); + assertEquals(List.of(), cap.failed, "still not a turn failure"); + } + + @Test + void aBriefUnknownDuringPostTurnHousekeepingDoesNotDropTheLatch() { + // The other direction: the escape must not fire on a glitch, or the queued next delegation + // would overtake housekeeping that is still running. + Injector inj = new Injector(new AgentControl(herdr), new PostTurnListener()); + inj.enqueue(T, "first", TestTurnTokens.inert(T)); + inj.enqueue(T, "second", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); + inj.onStatus(T, AgentStatus.WORKING); + inj.onStatus(T, AgentStatus.IDLE); // first completes; reset dispatched + for (int i = 0; i < 10; i++) inj.onStatus(T, AgentStatus.UNKNOWN); // well under the grace + + assertFalse(inj.activeTargets().isEmpty(), "a brief glitch must not release the post-turn latch"); + assertEquals(List.of("first"), sent(), "the queued delegation must not overtake housekeeping"); + } + @Test void aTransientUnknownGlitchNeitherFailsNorBlocksCompletion() { Captor cap = new Captor();