Merge #549: widen Injector's delivery catch to Throwable (#546)

Closes the re-delivery window that merging #543 opened. I caused that; this closes it the same day.

Verified by me on the branch at 87871ea, base 0b032f5 (not stale).

Production diff is 11 lines: `catch (RuntimeException)` -> `catch (Throwable)` at Injector.java:391,
and the `sendError` local widened to `Throwable` so it compiles. I checked every use of `sendError`
myself — :533 `getMessage()` and :534 `completeExceptionally(Throwable)` — so the wider type
reaches nothing that needed the narrow one.

Build on the branch: exit 0, Tests run: 1719, Failures: 0, Errors: 0, Skipped: 0 (1716 on main plus
the 3 new tests). #459's javadoc reference gate: exit 0, 0 reference errors. Gitea CI run 1803 on
87871ea: success.

Three mutations of my own, none of them the ones the worker used:
1. Reverted the catch to `RuntimeException`. RED: `anErrorFromSendDoesNotRedeliverOnASecondRound`
   "expected: <1> but was: <2>" prompt calls, and `anErrorFromSendRemovesTheMessage...` reporting
   the Error escaping `onStatus`. That second message is the defect itself, stated by the test.
2. Deleted the `t.queue.poll()` in the catch arm. RED on the new test AND on the pre-existing
   `sendFailureDropsMessageAndFailsItsFuture` — so the new test is not carrying that behaviour alone.
3. Wrote `DELIVERED` instead of `NOT_DELIVERED` at :400, the catch arm only. RED on the new test and
   on `aHerdrExceptionFromSendStillProducesNotDeliveredUnchanged`, which is the control that proves
   the widening did not quietly change the ordinary path.

All three restored; sha256 back to 97c560b6e33fc49a1772abec92e5bbab613f8991deba2220d30221a2f546ba14.
Green control after the restores: InjectorTest 37/37, exit 0.

My third mutation did not apply on its first attempt — it asserted a unique match on
`p.state = Pending.State.NOT_DELIVERED;`, which occurs twice (:400 and :419), so the script wrote
nothing and the test run came back exit 0. That is not a surviving mutant, it is a non-result
wearing the same clothes. The pristine-anchor count catching it is the only reason I noticed.

Deliberately NOT fixed here, filed as #551: the catch arm assumes that reaching it means nothing was
sent, and nothing establishes that. `agent.prompt` pastes and submits in one call, and every failure
in the response half of `UnixSocketHerdrClient.call()` — dropped connection, malformed line, error
result — is a `HerdrException`, which is a `RuntimeException`, which this catch arm already caught
before today. So "records NOT_DELIVERED for a delivery that happened" is older than this PR and is
not created by it. The fleet01 lead argued it was a trap inside this change and asked to be argued
out of it before the merge; the measurement above is the argument, and their underlying diagnosis is
right and is now #551 with their wording on it.

fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java

@@ -306,7 +306,7 @@ public final class Injector {
         if (t == null) return;
 
         Pending sent = null;
-        RuntimeException sendError = null;
+        Throwable sendError = null;
         boolean turnCompleted = false;
         boolean turnFailed = false;
         boolean resubmit = false;
@@ -388,9 +388,14 @@ public final class Injector {
                                 t.turnObserved = false;
                                 t.injectableSincePickup = 0;
                                 sent = p;
-                            } catch (RuntimeException e) {
+                            } catch (Throwable e) {
                                 // Delivery failed at herdr; drop the poisoned message and surface it
-                                // rather than blocking the queue behind 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;
                                 sent = p;

fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java

@@ -2,9 +2,11 @@ package dev.ltms.fleet.inject;
 
 import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.spi.ILoggingEvent;
+import com.fasterxml.jackson.databind.JsonNode;
 import dev.ltms.fleet.herdr.AgentControl;
 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.TestTurnTokens;
 import dev.ltms.fleet.testing.CapturedLog;
@@ -664,4 +666,111 @@ class InjectorTest {
         ExecutionException ex = assertThrows(ExecutionException.class, f::get);
         assertInstanceOf(HerdrException.class, ex.getCause());
     }
+
+    /**
+     * A {@link HerdrClient} that throws a non-{@link RuntimeException} {@link Error} from {@code
+     * agent.prompt} instead of delegating — the fleetd #546 case: a stray non-RuntimeException
+     * throwable (e.g. a {@code NoClassDefFoundError}, fleetd #413) escaping the send seam at
+     * {@code Injector.java:383}. Records every call it sees itself, including the ones it throws
+     * for, since the delegate's own recording is never reached for {@code agent.prompt} — so a test
+     * can assert on exactly what this fake actually received.
+     */
+    private static final class ErrorOnPrompt implements HerdrClient {
+        private final FakeHerdr delegate;
+        private final List<FakeHerdr.Call> calls = new java.util.concurrent.CopyOnWriteArrayList<>();
+
+        private ErrorOnPrompt(FakeHerdr delegate) {
+            this.delegate = delegate;
+        }
+
+        List<FakeHerdr.Call> calls() {
+            return calls;
+        }
+
+        @Override
+        public JsonNode call(String method, Object params) {
+            calls.add(new FakeHerdr.Call(method, params));
+            if (method.equals("agent.prompt")) {
+                throw new AssertionError("simulated non-RuntimeException send failure (fleetd #546)");
+            }
+            return delegate.call(method, params);
+        }
+
+        @Override
+        public void close() {
+            delegate.close();
+        }
+    }
+
+    /**
+     * Mirrors StatusPoller's own per-target {@code catch (Throwable)} (fleetd #538 / PR #543): the
+     * production polling loop already swallows whatever escapes one target's round and comes back
+     * for the next one. A unit test that calls {@code onStatus} directly (bypassing StatusPoller)
+     * needs the same survival so it can observe what a SECOND round does, regardless of whether
+     * fleetd #546's fix is present.
+     */
+    private static void pollOnceSurviving(Injector inj, String target, AgentStatus status) {
+        try {
+            inj.onStatus(target, status);
+        } catch (Throwable ignored) {
+            // matches StatusPoller.loop's own catch (Throwable) added by PR #543
+        }
+    }
+
+    @Test
+    void anErrorFromSendRemovesTheMessageAndMarksItNotDelivered() {
+        // 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.
+        ErrorOnPrompt throwing = new ErrorOnPrompt(new FakeHerdr());
+        Injector inj = new Injector(new AgentControl(throwing));
+        Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T));
+
+        assertDoesNotThrow(() -> inj.onStatus(T, AgentStatus.IDLE),
+                "fleetd #546: an Error from the send seam must be caught inside onStatus, not "
+                        + "escape it");
+
+        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");
+    }
+
+    @Test
+    void anErrorFromSendDoesNotRedeliverOnASecondRound() {
+        // fleetd #546, acceptance test 2 — the test this ticket exists for. Before the fix, an
+        // Error at the send seam left the message QUEUED (Injector.java:378 peeks, not polls), so a
+        // second onStatus round re-entered the same try and sent the same text again: the member's
+        // pane got the same brief typed into it twice.
+        ErrorOnPrompt throwing = new ErrorOnPrompt(new FakeHerdr());
+        Injector inj = new Injector(new AgentControl(throwing));
+        inj.enqueue(T, "brief", TestTurnTokens.inert(T));
+
+        pollOnceSurviving(inj, T, AgentStatus.IDLE);
+        pollOnceSurviving(inj, T, AgentStatus.IDLE);
+
+        long promptCalls = throwing.calls().stream().filter(c -> c.method().equals("agent.prompt")).count();
+        assertEquals(1, promptCalls, "fleetd #546: the poisoned text must be sent exactly once — a "
+                + "second onStatus round must not re-enter send for the same message");
+    }
+
+    @Test
+    void aHerdrExceptionFromSendStillProducesNotDeliveredUnchanged() {
+        // 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.
+        FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
+        Injector inj = new Injector(new AgentControl(failing));
+        Injector.Delivery delivery = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
+
+        inj.onStatus(T, AgentStatus.IDLE);
+
+        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.NOT_DELIVERED, inj.cancel(delivery),
+                "a HerdrException must still be dropped and marked NOT_DELIVERED, unchanged by the "
+                        + "wider Throwable catch");
+    }
 }