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 4d95ddd..f26c82a 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java +++ b/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java @@ -494,54 +494,163 @@ public final class Injector { // Fire listeners / herdr calls after releasing the monitor so nothing runs on the poller // thread while it holds the target lock. - if (resubmit) { - try { - agentsFor(target).submit(target); // nudge a raced Enter so the pending paste submits - } catch (RuntimeException e) { - log.debug("resubmit to {} failed (will retry next poll): {}", target, e.getMessage()); + // + // fleetd #553: wrapped in try/finally. By this point, if `sent != null`, the delivery has + // already happened inside the monitor above — off the queue, p.state == DELIVERED, text + // typed into the target's pane — so `sent`'s future MUST be completed one way or another, + // in every path out of this region, or the caller (a blocking fleet_send, or an async + // ticket) waits forever on a message it actually received. But the finally must not + // swallow whatever escaped: a listener that throws is a defect in THAT listener, and it + // must still reach StatusPoller's catch (Throwable) so log.error fires — converting a + // loud listener bug into a silently orphaned future would be worse than the bug itself. + // + // 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. + // + // `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 + // 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. + boolean sentHandled = false; + try { + if (resubmit) { + try { + agentsFor(target).submit(target); // nudge a raced Enter so the pending paste submits + } catch (Throwable e) { + // fleetd #553: widened from RuntimeException (same reasoning as #549 at :382) — + // this is the FIRST block after the monitor, so an Error escaping it used to skip + // every block below it, including the `sent` completion. + log.debug("resubmit to {} failed (will retry next poll): {}", target, e.getMessage()); + } } - } - if (notReady != null) { - // Worker never became available: forget its (never-set) readiness, unblock every queued - // caller, and route the awaiting send through the same failure path as a stalled turn so - // a blocking or async waiter resolves WORKER_FAILED rather than riding out the timeout. - forget.accept(target); - RuntimeException cause = new IllegalStateException( - target + " never became available (no bridge MCP connection within the boot window)"); - for (Pending p : notReady) { - p.delivered().completeExceptionally(cause); + if (notReady != null) { + // Worker never became available: unblock every queued caller FIRST — these messages' + // fate (NOT_DELIVERED, off the queue) was already decided inside the monitor above, so + // fleetd #553 completes every one of them before calling forget.accept or + // turnListener.onTurnFailed below. Either of those is a listener/consumer callback and + // can throw (an ordinary RuntimeException is enough — the same reasoning as the rest of + // this ticket): completing the futures first means such a throw can no longer leave any + // of them permanently pending, regardless of which one throws or in which order. + RuntimeException cause = new IllegalStateException( + target + " never became available (no bridge MCP connection within the boot window)"); + for (Pending p : notReady) { + p.delivered().completeExceptionally(cause); + } + // route the awaiting send through the same failure path as a stalled turn so a blocking + // or async waiter resolves WORKER_FAILED rather than riding out the timeout. + forget.accept(target); + turnListener.onTurnFailed(target); } - turnListener.onTurnFailed(target); - } - if (turnCompleted) { - if (startPostTurn) { - boolean started = turnListener.onTurnCompleteWithPostAction(target); - synchronized (t) { - t.postTurnPending = false; - if (started) { - t.awaitingPostTurnPickup = true; - t.injectableSincePostTurnPickup = 0; + if (turnCompleted) { + if (startPostTurn) { + // fleetd #553: the listener call is wrapped so `t.postTurnPending` (set true inside + // the monitor above, before this call) is always reset. Before this wrapping, a + // RuntimeException from onTurnCompleteWithPostAction skipped the reset below, + // permanently wedging the target: postTurnPending stayed true forever, so this + // target's delivery guard (:376) would never again pass and no further message to it + // would ever be delivered — a target-level lockup, not just one skipped future. + // `started` defaults to false so an exception is treated as "the post-turn action did + // not start" rather than falsely arming the post-turn pickup latch for an action that + // never ran. + boolean started = false; + try { + started = turnListener.onTurnCompleteWithPostAction(target); + } finally { + synchronized (t) { + t.postTurnPending = false; + if (started) { + t.awaitingPostTurnPickup = true; + t.injectableSincePostTurnPickup = 0; + } + if (t.queue.isEmpty() && !t.awaitingPostTurnPickup) { + targets.remove(target, t); + } + } } - if (t.queue.isEmpty() && !t.awaitingPostTurnPickup) { - targets.remove(target, t); + } else { + turnListener.onTurnComplete(target); + } + } + if (turnFailed) { + turnListener.onTurnFailed(target); + } + 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 + // 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; + if (sendError != null) { + 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). + 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. + // + // fleetd #553 (lead review, comment 16916): `sentHandled` guards ONLY the onDelivered + // 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" — + // 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 + // 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. + // + // This recovery is wrapped in its own try/catch(Throwable) that swallows only ITS OWN + // throwable and logs at WARN — the future is still completed either way — while the + // ORIGINAL throwable from the try block above is left alone to keep unwinding out of + // this method to StatusPoller's catch (Throwable), so a listener bug stays loud. + if (sent != null) { + try { + if (!sentHandled && sendError == null) { + turnListener.onDelivered(target, sent.token()); + } + } catch (Throwable recoveryError) { + log.warn("fleetd #553 backstop: onDelivered failed for {} while recovering from " + + "an earlier listener failure; completing its delivery future " + + "anyway: {}", + target, recoveryError.getMessage()); + } finally { + // No-op (returns false) on the ordinary path, where the `if (sent != null)` block + // above already completed this future — see the comment above this block. + if (sendError != null) { + sent.delivered().completeExceptionally(sendError); + } else { + sent.delivered().complete(null); } } - } else { - turnListener.onTurnComplete(target); - } - } - if (turnFailed) { - turnListener.onTurnFailed(target); - } - if (sent != null) { - if (sendError != null) { - 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). - turnListener.onDelivered(target, sent.token()); - sent.delivered().complete(null); } } } 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 173389e..7164db8 100644 --- a/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java +++ b/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java @@ -8,7 +8,9 @@ import dev.ltms.fleet.herdr.AgentStatus; import dev.ltms.fleet.herdr.FakeHerdr; import dev.ltms.fleet.herdr.HerdrClient; import dev.ltms.fleet.herdr.HerdrException; +import dev.ltms.fleet.msg.Rendezvous; import dev.ltms.fleet.msg.TestTurnTokens; +import dev.ltms.fleet.msg.TurnToken; import dev.ltms.fleet.testing.CapturedLog; import org.junit.jupiter.api.Test; @@ -773,4 +775,426 @@ class InjectorTest { "a HerdrException must still be dropped and marked NOT_DELIVERED, unchanged by the " + "wider Throwable catch"); } + + // --- fleetd #553: a throwable from any listener callback in onStatus must not skip the + // delivered-future completion. The whole post-monitor region is now wrapped in try/finally. --- + + @Test + void theOriginalThrowableFromAListenerStillEscapesOnStatus() { + // fleetd #553's control: the entire point of the finally backstop is that it completes a + // future WITHOUT swallowing whatever escaped. A fix built the wrong way (e.g. catching + // Throwable in the finally, or wrapping the region in try/catch instead of try/finally) + // could make every other test in this group pass while still converting a loud listener + // bug into a silent one — which the ticket calls a worse outcome than the bug itself. This + // is deliberately its own test, not folded into another one's assertion. + RuntimeException boom = new RuntimeException("fleetd #553 control"); + TurnListener throwing = new TurnListener() { + @Override + public void onTurnComplete(String target) { + throw boom; + } + }; + Injector inj = new Injector(new AgentControl(herdr), throwing); + inj.enqueue(T, "only", TestTurnTokens.inert(T)); + inj.onStatus(T, AgentStatus.IDLE); // deliver + inj.onStatus(T, AgentStatus.WORKING); // pickup + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), + "onStatus must still propagate the listener's own throwable, unmodified"); + assertSame(boom, thrown, "must be the EXACT throwable, not a wrapper or a different instance"); + } + + @Test + void aRuntimeExceptionFromOnTurnCompleteStillCompletesTheNextDelivery() { + // fleetd #553, acceptance test 1: onTurnComplete fires for "first"'s completed turn INSIDE + // the same onStatus call that then peeks and delivers "second" — the turnCompleted check + // (Injector.java, inside the released-pickup branch) clears awaitingCompletion just before + // the delivery guard right below it is evaluated, so both happen in one round when there is + // no post-turn action. Before fleetd #553, a RuntimeException thrown here unwound straight + // out of onStatus and skipped the `if (sent != null)` block, leaving "second"'s future + // pending forever even though it was already off the queue, marked DELIVERED, and typed + // into the pane. + RuntimeException boom = new RuntimeException("boom from onTurnComplete"); + TurnListener throwing = new TurnListener() { + @Override + public void onTurnComplete(String target) { + throw boom; + } + }; + Injector inj = new Injector(new AgentControl(herdr), throwing); + inj.enqueue(T, "first", TestTurnTokens.inert(T)); + Injector.Delivery second = inj.enqueue(T, "second", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); // delivers "first" + inj.onStatus(T, AgentStatus.WORKING); // picked up + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), + "the throwable from onTurnComplete must still escape onStatus"); + assertSame(boom, thrown); + + assertEquals(List.of("first", "second"), sent(), + "fleetd #553: \"second\" must still be sent even though onTurnComplete threw for " + + "\"first\"'s completion"); + assertTrue(second.completion().isDone() && !second.completion().isCompletedExceptionally(), + "fleetd #553: \"second\"'s delivered future must still complete normally despite " + + "the throw"); + } + + private static final int READINESS_GRACE_SAMPLES = 240; // Injector.READINESS_GRACE_POLLS is private + + @Test + void aRuntimeExceptionFromForgetStillCompletesTheReadinessFailureFutures() { + // fleetd #553: forget.accept fires as part of the CB-114 readiness-grace path (production + // wires it to presence::forget). Before fleetd #553, that block called forget.accept + // BEFORE completing the queued messages' futures, so a RuntimeException from forget left + // every one of them pending forever even though they were already marked NOT_DELIVERED and + // dropped from the queue. fleetd #553 completes those futures first, so forget.accept + // throwing afterward can no longer un-complete them. + RuntimeException boom = new RuntimeException("boom from forget"); + Injector inj = new Injector(new AgentControl(herdr), TurnListener.NOOP, _ -> false, _ -> { + throw boom; + }); + Injector.Delivery delivery = inj.enqueue(T, "task", TestTurnTokens.inert(T)); + + // Build the not-ready streak up to (but not past) the grace threshold. + for (int i = 0; i < READINESS_GRACE_SAMPLES - 1; i++) inj.onStatus(T, AgentStatus.IDLE); + assertFalse(delivery.completion().isDone(), "must not be resolved before the grace expires"); + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), + "the throwable from forget.accept must still escape onStatus"); + assertSame(boom, thrown); + + assertTrue(delivery.completion().isCompletedExceptionally(), + "fleetd #553: the queued message's future must still be completed even though " + + "forget.accept threw"); + } + + @Test + void aRuntimeExceptionFromOnTurnFailedStillLeavesTheReadinessFailureFuturesCompleted() { + // fleetd #553, acceptance test: same readiness-grace path as the forget test above, but the + // throw comes from onTurnFailed instead. NOTE: onTurnFailed is the LAST statement in this + // block (both before and after this ticket), so the queued messages' futures are already + // completed by the time it runs, regardless of this fix — this test cannot be made to fail + // against the pre-fleetd-553 code the way the onTurnComplete/forget/postAction tests can; I + // could not find a construction where onTurnFailed's own throw is what stands between a + // future and its completion (flagged to the lead via fleet_ask; no reply arrived before the + // ~55s window closed, so recorded here instead). It still pins a real invariant this ticket + // cares about — an ordinary RuntimeException from this callback must not un-complete a + // future that was already decided — so it stays as a regression lock, not a bug-fix proof. + RuntimeException boom = new RuntimeException("boom from onTurnFailed"); + TurnListener throwing = new TurnListener() { + @Override + public void onTurnComplete(String target) { + } + + @Override + public void onTurnFailed(String target) { + throw boom; + } + }; + Injector inj = new Injector(new AgentControl(herdr), throwing, _ -> false, _ -> { + }); + Injector.Delivery delivery = inj.enqueue(T, "task", TestTurnTokens.inert(T)); + + for (int i = 0; i < READINESS_GRACE_SAMPLES - 1; i++) inj.onStatus(T, AgentStatus.IDLE); + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), + "the throwable from onTurnFailed must still escape onStatus"); + assertSame(boom, thrown); + + assertTrue(delivery.completion().isCompletedExceptionally(), + "the queued message's future must remain completed despite onTurnFailed throwing"); + } + + @Test + void aRuntimeExceptionFromOnTurnCompleteWithPostActionStillUnwedgesTheTarget() { + // fleetd #553, acceptance test: before this ticket, a RuntimeException from + // onTurnCompleteWithPostAction skipped the `t.postTurnPending = false` reset that used to + // sit right after it (with nothing guarding it), permanently wedging the target — + // postTurnPending stayed true forever, so the delivery guard never passed again and + // "second" (already queued) was never delivered, no matter how many further onStatus + // rounds ran. fleetd #553 wraps that call so the reset always runs. + RuntimeException boom = new RuntimeException("boom from onTurnCompleteWithPostAction"); + class ThrowingPostTurn implements TurnListener { + @Override + public void onTurnComplete(String target) { + } + + @Override + public boolean hasPostTurnAction(String target) { + return true; + } + + @Override + public boolean onTurnCompleteWithPostAction(String target) { + throw boom; + } + } + Injector inj = new Injector(new AgentControl(herdr), new ThrowingPostTurn()); + inj.enqueue(T, "first", TestTurnTokens.inert(T)); + Injector.Delivery second = inj.enqueue(T, "second", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); // delivers "first" + inj.onStatus(T, AgentStatus.WORKING); // picked up + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), // "first"'s turn completes; post-action throws + "the throwable from onTurnCompleteWithPostAction must still escape onStatus"); + assertSame(boom, thrown); + assertEquals(List.of("first"), sent(), "\"second\" must not be sent in the SAME round as the throw"); + + inj.onStatus(T, AgentStatus.IDLE); // a later, unrelated round + assertEquals(List.of("first", "second"), sent(), + "fleetd #553: the target must not stay wedged — \"second\" must still be delivered " + + "once postTurnPending is reset despite the earlier throw"); + assertTrue(second.completion().isDone() && !second.completion().isCompletedExceptionally()); + } + + /** + * A {@link HerdrClient} that throws a non-{@link RuntimeException} {@link Error} from {@code + * agent.send_keys} instead of delegating — the fleetd #553 case for the resubmit nudge: its own + * catch (Injector.java, the first block after the monitor) was {@code RuntimeException}-only, + * the same one-class-too-narrow shape #546 fixed at the send seam. Records every call it sees + * itself, mirroring {@code ErrorOnPrompt} above, since the delegate's own recording is never + * reached for {@code agent.send_keys}. + */ + private static final class ErrorOnSendKeys implements HerdrClient { + private final FakeHerdr delegate; + private final List calls = new java.util.concurrent.CopyOnWriteArrayList<>(); + + private ErrorOnSendKeys(FakeHerdr delegate) { + this.delegate = delegate; + } + + List calls() { + return calls; + } + + @Override + public JsonNode call(String method, Object params) { + calls.add(new FakeHerdr.Call(method, params)); + if (method.equals("agent.send_keys")) { + throw new AssertionError("simulated non-RuntimeException resubmit failure (fleetd #553)"); + } + return delegate.call(method, params); + } + + @Override + public void close() { + delegate.close(); + } + } + + @Test + void anErrorFromTheResubmitNudgeDoesNotPreventTheSentCompletion() { + // fleetd #553, acceptance test: the resubmit nudge's own catch was RuntimeException-only — + // the same one-class-too-narrow shape #546 fixed at the send seam (now :382). It sits FIRST + // after the monitor, so before this ticket an Error escaping it skipped every block below in + // THAT round, including any `sent` completion that round might otherwise produce. Widened to + // Throwable (matching #549's own widening) so it can no longer escape onStatus at all — the + // delivery that already completed at send time, and the worker's later pickup, are both + // unaffected by the Error in between. + ErrorOnSendKeys throwing = new ErrorOnSendKeys(new FakeHerdr()); + Injector inj = new Injector(new AgentControl(throwing)); + Injector.Delivery delivery = inj.enqueue(T, "task", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); // delivers "task"; awaiting pickup + assertTrue(delivery.completion().isDone() && !delivery.completion().isCompletedExceptionally(), + "the delivery completes at send time, before any resubmit nudge is attempted"); + + assertDoesNotThrow(() -> inj.onStatus(T, AgentStatus.IDLE), // still idle -> resubmit; Error thrown+caught + "fleetd #553: an Error from the resubmit nudge must be caught inside onStatus, not " + + "escape it"); + + inj.onStatus(T, AgentStatus.WORKING); // the worker's real pickup must still resolve cleanly + long promptCalls = throwing.calls().stream().filter(c -> c.method().equals("agent.prompt")).count(); + assertEquals(1, promptCalls, + "sent exactly once, unaffected by the resubmit Error in between"); + } + + @Test + void ordinarySuccessStillCompletesExactlyOnceAfterOnDelivered() { + // Acceptance: the ordinary success path is unchanged by fleetd #553 — the future completes + // normally exactly once, and onDelivered still runs before it (CB-115's pane baseline). + List events = new ArrayList<>(); + TurnListener listener = new TurnListener() { + @Override + public void onTurnComplete(String target) { + } + + @Override + public void onDelivered(String target, dev.ltms.fleet.msg.TurnToken token) { + events.add("onDelivered"); + } + }; + Injector inj = new Injector(new AgentControl(herdr), listener); + CompletableFuture f = inj.enqueue(T, "hello", TestTurnTokens.inert(T)).completion(); + f.whenComplete((v, ex) -> events.add("completed")); + + inj.onStatus(T, AgentStatus.IDLE); + + assertEquals(List.of("onDelivered", "completed"), events, + "onDelivered must still run before the future completes, unchanged by fleetd #553"); + assertTrue(f.isDone() && !f.isCompletedExceptionally()); + } + + // Acceptance: the ordinary failure path (a HerdrException at the send seam still completes the + // future exceptionally with that exception) is already covered, unchanged, by the existing + // aHerdrExceptionFromSendStillProducesNotDeliveredUnchanged test above (fleetd #546) — its + // codepath is untouched by fleetd #553's try/finally, since sendError != null is set before the + // try block begins and that branch never threw to begin with. + + // --- fleetd #553, sharpened acceptance (ticket comments 16870/16884/16890/16903): completing + // sent.delivered() is only HALF the job. See the two tests below. --- + + @Test + void aRuntimeExceptionFromOnTurnCompleteStillRegistersTheRendezvousWaiterForTheNextDelivery() { + // fleetd #553's real invariant, not just the completion of sent.delivered(). There are TWO + // futures at stake: 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() — production wires this to + // CompletionResolver.captureBaseline, which does the inFlight.put() — and that call lives + // inside the very `if (sent != null)` block a plain "complete the future" backstop does not + // reach. A finally that only completes sent.delivered() converts a hang into a HANG WITH A + // SUCCESS RECEIPT: the caller is told the send landed, then waits out its full timeout for + // an answer that can never resolve, because CompletionResolver.resolve (:301-308) finds no + // inFlight entry and returns silently. This test is red on that half-fix, and green only + // once the finally also runs onDelivered() when the normal block never got the chance. + // + // The scenario: onTurnComplete throws while completing "first"'s turn, in the SAME onStatus + // round that (per Injector.java :361-391) then peeks and delivers "second" — the normal + // case, not a corner, since the :365 assignment is what lets the :376 delivery guard pass. + // turnListener.onTurnComplete runs BEFORE the `if (sent != null)` block, so the throw here + // means onDelivered for "second" is never reached on the normal path — only the finally + // backstop can register it. + Rendezvous rendezvous = new Rendezvous(); + CompletionResolver resolver = new CompletionResolver(new AgentControl(new FakeHerdr()), rendezvous, + ExhaustedPatternLookup.none(), ExhaustionSink.none()); + RuntimeException boom = new RuntimeException("boom from onTurnComplete"); + TurnListener throwingOnTurnCompleteOnly = new TurnListener() { + @Override + public void onDelivered(String target, TurnToken token) { + resolver.onDelivered(target, token); // the real registration, exactly as production wires it + } + + @Override + public void onTurnComplete(String target) { + throw boom; // "first"'s completion callback, thrown before "second" is delivered + } + }; + Injector inj = new Injector(new AgentControl(herdr), throwingOnTurnCompleteOnly); + CompletableFuture waiter = rendezvous.open("session-a"); + TurnToken secondToken = new TurnToken(T, waiter); + inj.enqueue(T, "first", TestTurnTokens.inert(T)); + Injector.Delivery second = inj.enqueue(T, "second", secondToken); + + inj.onStatus(T, AgentStatus.IDLE); // delivers "first" + inj.onStatus(T, AgentStatus.WORKING); // picked up + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), // "first" completes (throws); "second" delivered + "the original throwable from onTurnComplete must still escape onStatus"); + assertSame(boom, thrown, "must be the EXACT throwable, not a wrapper or a different instance"); + + assertEquals(List.of("first", "second"), sent(), + "\"second\" must still be sent even though onTurnComplete threw for \"first\""); + assertTrue(second.completion().isDone() && !second.completion().isCompletedExceptionally(), + "\"second\"'s delivery future must still complete despite the throw"); + + CompletionResolver.InFlight inFlight = resolver.inFlight(T); + assertNotNull(inFlight, + "fleetd #553: the finally must register the new turn's waiter (via onDelivered), not " + + "just complete its delivery future — otherwise a blocking fleet_send is told " + + "its message landed and then waits out the full timeout for an answer that " + + "can never resolve"); + assertSame(waiter, inFlight.waiter(), + "the registered waiter must be exactly \"second\"'s waiter, not some other one"); + } + + @Test + void anOnDeliveredThrowAfterItsOwnRegistrationDoesNotRunASecondTime() { + // fleetd #553 comment 16903: `sentHandled` must be set to true BEFORE onDelivered() runs, + // not after. Setting it after would mean a throw from onDelivered() PART WAY THROUGH — e.g. + // after CompletionResolver's own captureBaseline has already done its inFlight.put() — still + // leaves the flag false, so the finally backstop (seeing "not handled") calls onDelivered() a + // SECOND time. That second captureBaseline runs later, over a pane that may already have + // absorbed this turn's output, and CompletionResolver.resolve's own baseline-equals-tail + // suppression (:364-369, which deliberately keeps the in-flight record on a match) can then + // drop every future completion for the turn, permanently. This assertion is red on a + // `sentHandled = true` placed AFTER the onDelivered() call (onDelivered runs twice) and green + // on the correct placement (runs exactly once) — regardless of what onDelivered itself did. + AtomicInteger onDeliveredCalls = new AtomicInteger(); + RuntimeException boom = new RuntimeException("boom from onDelivered, after its own registration ran"); + TurnListener listener = new TurnListener() { + @Override + public void onTurnComplete(String target) { + } + + @Override + public void onDelivered(String target, TurnToken token) { + onDeliveredCalls.incrementAndGet(); // stand-in for CompletionResolver's inFlight.put + throw boom; // then fail, as if a LATER step inside onDelivered blew up + } + }; + Injector inj = new Injector(new AgentControl(herdr), listener); + inj.enqueue(T, "task", TestTurnTokens.inert(T)); + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), // delivers "task"; onDelivered throws + "the original throwable from onDelivered must still escape onStatus"); + assertSame(boom, thrown); + + assertEquals(1, onDeliveredCalls.get(), + "fleetd #553: onDelivered must be called exactly once — a `sentHandled` flag set " + + "AFTER the call (rather than before) would leave it false here and the " + + "finally backstop would call onDelivered a second time"); + } + + @Test + void anOnDeliveredThrowOnTheNormalPathStillCompletesTheDeliveryFuture() { + // fleetd #553, lead review of PR #557 (ticket comment 16916): `sentHandled` must guard ONLY + // the onDelivered RE-CALL in the finally, never the future completion alongside it. Gating + // BOTH behind `!sentHandled` (the shape this test is red against) misses the one path the + // flag's own correct placement creates: `sentHandled` is set to true FIRST, before the + // onDelivered() call, inside the `if (sent != null)` block above (see the previous test — + // that placement is right and must not change). So when onDelivered() itself throws on that + // NORMAL path, `sentHandled` already reads true by the time control reaches the finally, and + // a `sent != null && !sentHandled` guard around the WHOLE recovery — completion included — + // skips it entirely. The message was typed into the target's pane and taken off the queue + // inside the monitor, same as any other delivery, so its future is left pending forever: the + // exact defect this ticket exists to close, just reached from a different throwing call. + // + // The fix splits the one flag's two jobs: `!sentHandled` keeps gating only the onDelivered + // call (so the exactly-once guarantee in the test above still holds — CompletableFuture. + // complete/completeExceptionally are idempotent, so completing unconditionally here is a + // no-op on the ordinary path, where the `if (sent != null)` block already completed it. + RuntimeException boom = new RuntimeException("boom from onDelivered on the normal path"); + TurnListener listener = new TurnListener() { + @Override + public void onTurnComplete(String target) { + } + + @Override + public void onDelivered(String target, TurnToken token) { + throw boom; + } + }; + Injector inj = new Injector(new AgentControl(herdr), listener); + CompletableFuture delivered = inj.enqueue(T, "task", TestTurnTokens.inert(T)).completion(); + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> inj.onStatus(T, AgentStatus.IDLE), // delivers "task"; onDelivered throws + "the original throwable from onDelivered must still escape onStatus"); + assertSame(boom, thrown); + + assertTrue(delivered.isDone(), + "fleetd #553: \"task\" was actually delivered — typed into the pane and taken off " + + "the queue inside the monitor — so its delivery future must be completed on " + + "every path out of onStatus, including the one where onDelivered itself is " + + "what threw. Leaving it pending here is a hang, not a fix"); + } }