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 bf79d8a..2bc3b99 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java +++ b/fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java @@ -2,6 +2,7 @@ package dev.ltms.fleet.inject; import dev.ltms.fleet.herdr.AgentControl; import dev.ltms.fleet.herdr.AgentStatus; +import dev.ltms.fleet.herdr.HerdrException; import dev.ltms.fleet.herdr.HerdrRouter; import dev.ltms.fleet.msg.TurnToken; import org.slf4j.Logger; @@ -46,6 +47,17 @@ import java.util.stream.Collectors; * turn — the pickup-grace path (a turn too fast to sample) unwedges the queue but does not fire * completion, since without a sampled {@code working} there is no trustworthy "the worker just * finished the task" signal to act on. + * + *

Delivery honesty (fleetd #551). The one external send call + * ({@link AgentControl#send}) is irreversible, and its own response can still fail after the text + * has already reached the worker's pane — herdr replied (or may have), so the request was already + * processed, but the reply itself then failed to parse or carried an error. A queued entry is + * polled off the queue and marked {@code ATTEMPTED} before that call is made, not after, + * so a failure from the call itself can never be recorded as a confident {@code NOT_DELIVERED} for + * text that may already be sitting in the pane. {@code NOT_DELIVERED} stays reserved for the cases + * where nothing was ever attempted — the readiness grace expiring, {@link #drop}, or a herdr + * {@code *_not_found} error, which the rest of this codebase already treats as a confirmed absence + * rather than a merely inconclusive failure (see {@code StatusPoller}, {@code AgentControl}). */ public final class Injector { @@ -217,8 +229,23 @@ public final class Injector { /** The result of trying to remove an undelivered message from the injector. */ public enum Cancellation { + /** The message was still queued and this call removed it; the target saw nothing. */ CANCELLED, + /** The message reached the worker's pane and is confirmed delivered. */ DELIVERED, + /** + * fleetd #551: the injector called {@link AgentControl#send} for this message and that call + * threw before its outcome was known — the text may or may not have reached the pane. A + * caller reporting this to an operator must say "uncertain", not "definitely not + * delivered": reading it as a confident negative invites a resend of text that may already + * be sitting in the pane (a double delivery), which is worse than the ambiguity itself. + */ + ATTEMPTED, + /** + * The message never reached the worker's pane — nothing was ever attempted for it (the + * readiness grace expired, the target was dropped, or send failed with a herdr + * {@code *_not_found} error, which is a confirmed absence, not merely inconclusive). + */ NOT_DELIVERED } @@ -241,7 +268,29 @@ public final class Injector { /** A pending message and the future that completes when it has been delivered. */ private static final class Pending { - enum State { QUEUED, DELIVERED, NOT_DELIVERED, CANCELLED } + enum State { + /** Still in the target's queue, not yet attempted. */ + QUEUED, + /** + * fleetd #551: {@link AgentControl#send} has been called for this entry and its outcome + * is not yet known — recorded BEFORE the call (see {@link #onStatus}), so a Throwable + * from send() can never leave the entry either still QUEUED or falsely marked + * NOT_DELIVERED. Upgraded to DELIVERED on success; left as ATTEMPTED on an ordinary + * failure, since reaching the catch does not prove the text never reached the pane. + */ + ATTEMPTED, + /** {@code send()} returned normally: the text is confirmed to have reached the pane. */ + DELIVERED, + /** + * Confirmed — not merely inconclusive — that nothing was ever sent for this entry: the + * worker never became ready ({@link #onStatus}'s readiness-grace expiry), the target was + * dropped ({@link #drop}), or send failed with a herdr {@code *_not_found} error. Never + * written for an entry whose send() outcome is unknown; see ATTEMPTED. + */ + NOT_DELIVERED, + /** Removed from the queue by {@link #cancel} before it was ever attempted. */ + CANCELLED + } final String target; final String text; @@ -333,7 +382,14 @@ public final class Injector { } private static Cancellation cancellationOf(Pending p) { - return p.state == Pending.State.DELIVERED ? Cancellation.DELIVERED : Cancellation.NOT_DELIVERED; + // fleetd #551 (comment 17037): a two-way split on a now-three-way question folded ATTEMPTED + // into NOT_DELIVERED with no compiler error and no failing test — the exact defect this + // ticket exists to fix, one layer up. ATTEMPTED gets its own answer instead. + return switch (p.state) { + case DELIVERED -> Cancellation.DELIVERED; + case ATTEMPTED -> Cancellation.ATTEMPTED; + case QUEUED, NOT_DELIVERED, CANCELLED -> Cancellation.NOT_DELIVERED; + }; } private static boolean isQuiescent(Target t) { @@ -426,9 +482,18 @@ public final class Injector { if (p != null && ready.test(target)) { t.notReadySincePoll = 0; t.notReadySinceMillis = 0; + // fleetd #551: poll and record BEFORE the irreversible send, not after. + // The entry comes off the queue and its state is set to ATTEMPTED here, + // unconditionally — so a Throwable escaping the send call below (caught or + // not) can never leave the entry QUEUED at the head of t.queue (the fleetd + // #546 hazard, since peek() alone would let the next onStatus round re-enter + // this block and send the same text again), and no path can write a + // confident DELIVERED or NOT_DELIVERED before we actually know which one + // happened. + t.queue.poll(); + p.state = Pending.State.ATTEMPTED; try { agentsFor(target).send(target, p.text()); - t.queue.poll(); p.state = Pending.State.DELIVERED; t.awaitingPickup = true; t.awaitingCompletion = true; @@ -436,15 +501,24 @@ public final class Injector { t.injectableSincePickup = 0; sent = p; } catch (Throwable e) { - // Delivery failed at herdr; drop the poisoned message and surface it - // rather than blocking the queue behind it. Catches Throwable, not just - // RuntimeException: fleetd #546 — an Error escaping this send (e.g. a - // NoClassDefFoundError, see #413) would otherwise leave the entry QUEUED - // at the head of t.queue. Line :378 peeks rather than polls, so the next - // onStatus round would re-enter this try and send the same text again, - // typing the same brief into the member's pane a second time. - t.queue.poll(); - p.state = Pending.State.NOT_DELIVERED; + // fleetd #551: leave p.state == ATTEMPTED (recorded above, before the + // call) rather than downgrading it to NOT_DELIVERED here — reaching this + // catch does not prove the text never reached the pane. Three of the + // four HerdrException throw sites in HerdrCodec fire only after herdr + // has already replied (so it processed the request), and the fourth (a + // transport IOException) leaves it genuinely unknown whether herdr even + // received the bytes — see #551 comment 16867. The one exception is a + // herdr `*_not_found` error: that family is already read as "definitely + // absent, not merely inconclusive" everywhere else in this codebase + // (StatusPoller, AgentControl's own retry, WorkspaceControl, + // HerdrPeerLauncher, FleetApp, ReplyPushLoop) because it means the + // target pane/agent does not exist at all, so nothing could have been + // pasted anywhere — #551 keeps the new state consistent with that + // existing vocabulary rather than inventing a second one. + if (e instanceof HerdrException he && he.code() != null + && he.code().endsWith("_not_found")) { + p.state = Pending.State.NOT_DELIVERED; + } sent = p; sendError = e; } diff --git a/fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java b/fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java index 9ce814d..8773b5d 100644 --- a/fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java +++ b/fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java @@ -94,16 +94,22 @@ public final class MessageService { TIMED_OUT_WORKING, /** * Timed out with no confirmed delivery. Despite the name, this does not mean the message - * is sitting in a queue — on every route it will not arrive later. {@link #send} reaches - * this outcome through {@link Injector#cancel}, whose result tells the routes apart: + * is sitting in a queue. {@link #send} reaches this outcome through {@link Injector#cancel}, + * whose result tells three routes apart: * {@link Injector.Cancellation#CANCELLED} means the message was still queued and this call - * removed it, so the target saw nothing; {@link Injector.Cancellation#NOT_DELIVERED} means - * an earlier attempt already decided the message's fate — the injector's call to the - * target's terminal ({@link dev.ltms.fleet.herdr.AgentControl#send}) threw, or the queue - * was cleared because the target never became ready or was abandoned — and {@code cancel} - * is only reporting that pre-existing state. On the failed-attempt route the terminal may - * already hold a partial paste from before the call threw, so only the {@code CANCELLED} - * case establishes that the target saw nothing. + * removed it, so the target saw nothing and it will not arrive later; + * {@link Injector.Cancellation#NOT_DELIVERED} means nothing was ever sent — the queue was + * cleared because the target never became ready or was abandoned, or the injector's call to + * the target's terminal ({@link dev.ltms.fleet.herdr.AgentControl#send}) failed with a herdr + * error that this codebase already treats as a confirmed absence — so this route too + * establishes that the target saw nothing and it will not arrive later; but + * {@link Injector.Cancellation#ATTEMPTED} (fleetd #551) means that call was made and its + * outcome is unknown. {@code agent.prompt} pastes and submits in one call, so on + * this route the target may hold a complete, already-submitted turn and be working on it + * right now — {@link Outcome#TIMED_OUT_WORKING}'s meaning, reported here as + * {@code TIMED_OUT_QUEUED} only because this caller never observed the pickup. Only + * {@code CANCELLED} and {@code NOT_DELIVERED} establish that the target saw nothing; + * {@code ATTEMPTED} does not. */ TIMED_OUT_QUEUED, /** Another send to this session was in flight for the whole window. */ @@ -313,16 +319,18 @@ public final class MessageService { /** * Targets whose last send timed out with {@link Outcome#TIMED_OUT_QUEUED} (CB-640) — {@link * #send} called {@link Injector#cancel} and got back something other than {@code DELIVERED}. - * That covers more than one history: the message may have still been queued and {@code cancel} - * removed it right there, or an earlier attempt may already have failed (the call to the - * target's terminal threw) or been abandoned (the target never became ready, or was torn - * down). What holds on every route: the message will not arrive later, and it is not sitting - * in any queue. What does NOT hold on every route: that the target saw nothing — a failed - * delivery attempt can leave a partial paste behind. Set where {@link #send} already computes - * {@code wasDelivered} for that outcome; no queue is kept here, only the fact that the send - * ended with no confirmed delivery. Cleared the same way as {@link #strandedReplies}: the next - * accepted delivery for the target ({@link #send} opening a fresh waiter) or a teardown - * ({@link #abandon}). + * That covers three histories, not one: {@link Injector.Cancellation#CANCELLED} — the message + * was still queued and {@code cancel} removed it right there; {@link + * Injector.Cancellation#NOT_DELIVERED} — nothing was ever sent, because the target never became + * ready, was torn down, or the call to its terminal failed with a herdr error this codebase + * already treats as a confirmed absence; or {@link Injector.Cancellation#ATTEMPTED} (fleetd + * #551) — the call to the target's terminal was made and its outcome is unknown, so the target + * may already hold a complete, submitted turn. Only the first two mean the message will not + * arrive later and the target saw nothing; on the third it may already have arrived in full. + * Set where {@link #send} already computes {@code wasDelivered} for that outcome; no queue is + * kept here, only the fact that the send ended with no confirmed delivery. Cleared the same way + * as {@link #strandedReplies}: the next accepted delivery for the target ({@link #send} opening + * a fresh waiter) or a teardown ({@link #abandon}). */ private final ConcurrentHashMap queuedDeliveries = new ConcurrentHashMap<>(); private final AtomicLong ticketSeq = new AtomicLong(); @@ -412,14 +420,18 @@ public final class MessageService { * with no confirmed delivery — the caller saw {@link Outcome#TIMED_OUT_QUEUED} (see the * {@code TimeoutException} branch of {@link #send}). Despite the method's name, this is not * proof that a message is sitting in a queue: {@link Injector#cancel} reports this outcome - * either because the message was still queued and got removed right there, or because an - * earlier attempt already failed (the call to the target's terminal threw) or was abandoned - * (the target never became ready, or was torn down). Either way the message will not arrive - * later. It does NOT follow that the target saw nothing — on the failed-attempt route the - * terminal may already hold a partial paste. Distinct from {@link Outcome#TIMED_OUT_WORKING}, - * where delivery already happened and only the reply is outstanding. Cleared the next time - * this target's delivery is accepted or the target is abandoned — see - * {@link #queuedDeliveries}. + * through three routes. {@link Injector.Cancellation#CANCELLED} means the message was still + * queued and got removed right there. {@link Injector.Cancellation#NOT_DELIVERED} means + * nothing was ever sent — the target never became ready, was torn down, or the call to its + * terminal failed with a herdr error this codebase already treats as a confirmed absence. + * Only these two routes mean the message will not arrive later. {@link + * Injector.Cancellation#ATTEMPTED} (fleetd #551) means the call to the target's terminal was + * made and its outcome is unknown: {@code agent.prompt} pastes and submits in one + * call, so on this route the target may already hold a complete, submitted turn and be + * working on it right now — it does NOT follow that the target saw nothing. Distinct from + * {@link Outcome#TIMED_OUT_WORKING}, where delivery already happened and only the reply is + * outstanding. Cleared the next time this target's delivery is accepted or the target is + * abandoned — see {@link #queuedDeliveries}. */ public boolean hasQueuedDelivery(String target) { return target != null && queuedDeliveries.containsKey(target); @@ -974,11 +986,12 @@ public final class MessageService { } log.debug("send to {} timed out (delivered={})", target, wasDelivered); if (!wasDelivered) { - // CB-640: record that delivery did not happen for fleet health (see + // CB-640: record that delivery is not confirmed, for fleet health (see // queuedDeliveries). Whatever injector.cancel() reported above — this call - // removed a still-queued Pending, or an earlier attempt already failed or - // was abandoned — the send ends with no confirmed delivery and will not - // arrive later. + // removed a still-queued Pending (CANCELLED), an earlier attempt already + // failed with a confirmed absence (NOT_DELIVERED), or an earlier attempt was + // made and its outcome is unknown (ATTEMPTED, fleetd #551 — the message may + // already have arrived in full) — the send ends with no confirmed delivery. queuedDeliveries.put(target, Boolean.TRUE); } return recorded(new Reply( 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 4efaf91..bf20c79 100644 --- a/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java +++ b/fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java @@ -720,10 +720,13 @@ class InjectorTest { } @Test - void anErrorFromSendRemovesTheMessageAndMarksItNotDelivered() { + void anErrorFromSendRemovesTheMessageAndMarksItAttempted() { // fleetd #546, acceptance test 1: an Error (not a RuntimeException) escaping the send seam - // at Injector.java:383 must still be caught, the message dropped from the queue, and its - // state set to NOT_DELIVERED — never left QUEUED at the head. + // must still be caught and the message dropped from the queue, never left QUEUED at the + // head. fleetd #551 updates what it is marked: an AssertionError from the send call itself + // proves nothing about whether the text reached the pane, so it is recorded as ATTEMPTED + // (uncertain), not a confident NOT_DELIVERED — see anErrorFromSendMarksItNotDeliveredOnlyForANotFoundCode + // below for the one case that still gets NOT_DELIVERED. ErrorOnPrompt throwing = new ErrorOnPrompt(new FakeHerdr()); Injector inj = new Injector(new AgentControl(throwing)); Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T)); @@ -734,9 +737,9 @@ class InjectorTest { assertTrue(delivery.completion().isCompletedExceptionally(), "the delivery's future must surface the send failure"); - assertEquals(Injector.Cancellation.NOT_DELIVERED, inj.cancel(delivery), - "the message must be dropped and marked NOT_DELIVERED, not left QUEUED at the " - + "head of the queue"); + assertEquals(Injector.Cancellation.ATTEMPTED, inj.cancel(delivery), + "fleetd #551: the message must be dropped and marked ATTEMPTED (not a confident " + + "NOT_DELIVERED), and never left QUEUED at the head of the queue"); } @Test @@ -758,10 +761,14 @@ class InjectorTest { } @Test - void aHerdrExceptionFromSendStillProducesNotDeliveredUnchanged() { + void aHerdrExceptionFromSendStillSurfacesButNowReportsAttempted() { // fleetd #546, acceptance test 3 (control): HerdrException extends RuntimeException, so it - // was already caught before this ticket's widening. This pins that the ordinary path is - // unchanged — still dropped, still NOT_DELIVERED, still surfaced to the caller. + // was already caught before #546's widening. The send-failure/surface behaviour is + // unchanged by #551 — still dropped, still surfaced to the caller — but fleetd #551 + // deliberately changes WHAT it is marked: "send_failed" is not a herdr `*_not_found` code, + // so this is exactly the response-half case #551 exists to fix. See + // aHerdrExceptionAfterThePasteIsNeverRecordedAsConfidentlyNotDelivered for the acceptance + // test this ticket was filed for. FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed"); Injector inj = new Injector(new AgentControl(failing)); Injector.Delivery delivery = inj.enqueue(T, "boom", TestTurnTokens.inert(T)); @@ -771,9 +778,90 @@ class InjectorTest { assertTrue(delivery.completion().isCompletedExceptionally(), "a HerdrException at the send seam must still surface to the caller, unchanged by " + "fleetd #546's widening"); + assertEquals(Injector.Cancellation.ATTEMPTED, inj.cancel(delivery), + "fleetd #551: a non-*_not_found HerdrException must now report ATTEMPTED, not a " + + "confident NOT_DELIVERED"); + } + + // --- fleetd #551: the injector records the delivery attempt BEFORE the irreversible send, not + // after, so a failure in the send's response window can never be written down as a confident + // NOT_DELIVERED for text that may already be sitting in the worker's pane. --- + + @Test + void aHerdrExceptionAfterThePasteIsNeverRecordedAsConfidentlyNotDelivered() { + // fleetd #551, the ticket's acceptance test 1, and comment 17037's added test ("a test + // pinning that a caller cannot receive a confident NOT_DELIVERED for a message that reached + // the paste"): a HerdrException thrown from the response half of the send seam — herdr + // replied, with an error, so it definitely processed the request — must not leave a record + // claiming the text was never delivered. Red before the fix: on main at ba2f4d1 this + // asserted (and got) Injector.Cancellation.NOT_DELIVERED. + FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed"); + Injector inj = new Injector(new AgentControl(failing)); + Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); + + assertTrue(delivery.completion().isCompletedExceptionally(), + "the caller must still see the send failure"); + assertEquals(Injector.Cancellation.ATTEMPTED, inj.cancel(delivery), + "fleetd #551: a HerdrException from the response half of the send seam must not be " + + "recorded as a confident NOT_DELIVERED — the text may already be sitting " + + "in the pane"); + } + + @Test + void ordinarySuccessStillReportsDeliveredExactlyOnce() { + // fleetd #551, the ticket's acceptance test 2: the record-before-send reorder must not + // change the ordinary success path — exactly one send, and the caller-facing Cancellation + // for it is still DELIVERED, never left at the new ATTEMPTED value. + Injector.Delivery delivery = injector.enqueue(T, "hello", TestTurnTokens.inert(T)); + + injector.onStatus(T, AgentStatus.IDLE); + + assertEquals(List.of("hello"), sent(), "the text must be sent exactly once"); + assertTrue(delivery.completion().isDone() && !delivery.completion().isCompletedExceptionally(), + "the ordinary success path must still complete normally"); + assertEquals(Injector.Cancellation.DELIVERED, injector.cancel(delivery), + "fleetd #551: the ordinary success path must still report DELIVERED, unaffected by " + + "the record-before-send reorder"); + } + + @Test + void transportDownStillSurfacesTheErrorAndDropsTheEntry() { + // fleetd #551, the ticket's acceptance test 3: the ordinary transport-down path (herdr + // unreachable — a plain HerdrException with no error code) must be unaffected by the #551 + // reorder — the failure still surfaces to the caller, and the entry is not left sitting in + // the queue for a second onStatus round to resend. + FakeHerdr unreachable = new FakeHerdr().healthy(false); + Injector inj = new Injector(new AgentControl(unreachable)); + Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); + + assertTrue(delivery.completion().isCompletedExceptionally(), + "a transport-down failure must still surface to the caller"); + assertTrue(inj.activeTargets().isEmpty(), + "the entry must not be left queued for a second round to resend"); + } + + @Test + void aNotFoundCodeFromSendStillReportsAConfidentNotDelivered() { + // fleetd #551 (comment 17037, point 3): a herdr `*_not_found` error means the target + // pane/agent does not exist at all, so nothing could have been pasted anywhere — the + // codebase already treats this family as a confirmed absence everywhere else (StatusPoller, + // AgentControl's own retry). #551's new ATTEMPTED state must not swallow this case: it stays + // a confident NOT_DELIVERED. + FakeHerdr failing = new FakeHerdr().agentSendFailsWith("agent_not_found"); + Injector inj = new Injector(new AgentControl(failing)); + Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T)); + + inj.onStatus(T, AgentStatus.IDLE); + + assertTrue(delivery.completion().isCompletedExceptionally(), + "an agent_not_found failure must still surface to the caller"); assertEquals(Injector.Cancellation.NOT_DELIVERED, inj.cancel(delivery), - "a HerdrException must still be dropped and marked NOT_DELIVERED, unchanged by the " - + "wider Throwable catch"); + "fleetd #551: a herdr *_not_found error is a confirmed absence, not merely " + + "inconclusive — it must stay NOT_DELIVERED, not the new ATTEMPTED"); } // --- fleetd #553: a throwable from any listener callback in onStatus must not skip the