fleetd #437: pin the ack hit/miss contract in the AMQP contract test

AmqpReplyInboxContractTest is the one contract-group class CI actually
runs, and it never asserted on ack()'s return value at all — so the
exact defect this ticket fixes (reporting success for an ack that
removed nothing) was unpinned in the adapter fleetd runs live.

Add ackReportsHitVsMissAgainstARealBroker: a msgId never held for an
owned target returns false without throwing, a real held reply returns
true and is removed, and acking the same msgId again returns false.
Ran against both broker modes the class supports: Testcontainers
(AMQP_URI unset) and an external broker via AMQP_URI (the CI shape,
using a disposable container — not the shared local LavinMQ instance).

.gitea/workflows/ci.yml

@@ -87,18 +87,12 @@ jobs:
           apt-get update && apt-get install -y --no-install-recommends maven
           mvn -version
 
-      # The `contract` profile clears the default-excludes group, so `-Dgroups=contract` runs every
-      # @Tag("contract") test and nothing from the unit suite the `build` job already covered — a
-      # tag selects the whole group, so a test added to it later runs here automatically. A prior
-      # version of this step pinned `-Dtest=AmqpReplyInboxContractTest` by class name instead: that
-      # silently excluded every other contract test (including the herdr ones) from CI, and nobody
-      # noticed until the herdr protocol drifted out from under a test that never ran here
-      # (fleetd #449). If this runner has no herdr socket, the herdr-backed tests in the group
-      # skip on their own `assumeTrue` and only the broker-backed ones actually run — check the
-      # step output rather than assuming which.
+      # The `contract` profile clears the default-excludes group, so the @Tag("contract") AMQP test
+      # runs against the RabbitMQ service container (AMQP_URI). Pinned to the one contract test to
+      # avoid re-running the unit suite already covered by the `build` job.
       - name: Contract tests
         working-directory: fleetd
-        run: mvn -B -Pcontract test -Dgroups=contract
+        run: mvn -B -Pcontract test -Dtest=AmqpReplyInboxContractTest
 
       - name: Failing test output
         if: failure()

fleetd/src/main/java/dev/ltms/fleet/herdr/HerdrClient.java

@@ -3,7 +3,7 @@ package dev.ltms.fleet.herdr;
 import com.fasterxml.jackson.databind.JsonNode;
 
 /**
- * Client face onto the herdr daemon (protocol 19, herdr 0.8.0).
+ * Client face onto the herdr daemon (protocol 14, herdr 0.7.0).
  *
  * <p>This is the ONLY thing in {@code fleetd} that speaks to herdr. Every method
  * maps to a herdr JSON-RPC call over its Unix domain socket. Requests are

fleetd/src/main/java/dev/ltms/fleet/herdr/HerdrCodec.java

@@ -8,7 +8,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.nio.charset.StandardCharsets;
 
 /**
- * Wire codec for herdr's newline-delimited JSON-RPC (protocol 19).
+ * Wire codec for herdr's newline-delimited JSON-RPC (protocol 14).
  *
  * <p>Split out from the socket so the framing rules — the ones that actually bit us
  * during the spike (id MUST be a string; response carries {@code result} or

fleetd/src/main/java/dev/ltms/fleet/mcp/FleetMcp.java

@@ -994,7 +994,11 @@ public final class FleetMcp {
         if (isBlank(target) || isBlank(msgId)) {
             return error("target and msgId are required");
         }
-        messages.ackReply(target, msgId);
+        if (!messages.ackReply(target, msgId)) {
+            return error(msgId + " is not in " + target + "'s reply inbox (wrong id, wrong target, "
+                    + "or already acked). Held lead-to-lead (peer) mail cannot be acked this way — "
+                    + "read it with fleet_poll{coordId}.");
+        }
         return text("acknowledged " + msgId);
     }
 
@@ -1766,7 +1770,10 @@ public final class FleetMcp {
                         + "has processed a reply and wants to confirm it, leaving other pending replies "
                         + "in the inbox for later drain.",
                 objectSchema(Map.of(
-                        "target", stringProp("Worker session id whose inbox to ack from"),
+                        "target", stringProp("Worker session id whose inbox to ack from. Must name a "
+                                + "reply actually queued for it — an id in no inbox, or a coord-id "
+                                + "(peer held mail, read with fleet_poll{coordId} instead), errors "
+                                + "rather than reporting a false success"),
                         "msgId", stringProp("The message id to acknowledge")),
                         List.of("target", "msgId")));
     }

fleetd/src/main/java/dev/ltms/fleet/msg/AmqpReplyInbox.java

@@ -394,17 +394,17 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
     }
 
     @Override
-    public void ack(String target, String msgId) {
+    public boolean ack(String target, String msgId) {
         var perTarget = held.get(target);
         if (perTarget == null || perTarget == RELEASED) {
-            return;
+            return false;
         }
         Held h;
         synchronized (perTarget) {
             h = perTarget.remove(msgId);
         }
         if (h == null) {
-            return;                                // never held (or already acked) — no-op
+            return false;                          // never held (or already acked) — no-op
         }
         try {
             synchronized (channelLock) {
@@ -418,6 +418,7 @@ public final class AmqpReplyInbox implements ReplyInbox, AutoCloseable {
             }
             throw new IllegalStateException("cannot ack reply " + msgId + " on " + queueName(target), e);
         }
+        return true;
     }
 
     private DeliverCallback deliverCallback(String target) {

fleetd/src/main/java/dev/ltms/fleet/msg/InMemoryReplyInbox.java

@@ -59,16 +59,17 @@ public final class InMemoryReplyInbox implements ReplyInbox {
     }
 
     @Override
-    public void ack(String target, String msgId) {
+    public boolean ack(String target, String msgId) {
         if (!owned.contains(target)) {
-            return;
+            return false;
         }
         var perTarget = store.get(target);
-        if (perTarget != null) {
-            //noinspection SynchronizationOnLocalVariableOrMethodParameter
-            synchronized (perTarget) {
-                perTarget.remove(msgId);
-            }
+        if (perTarget == null) {
+            return false;
+        }
+        //noinspection SynchronizationOnLocalVariableOrMethodParameter
+        synchronized (perTarget) {
+            return perTarget.remove(msgId) != null;
         }
     }
 }

fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java

@@ -826,9 +826,13 @@ public final class MessageService {
     /**
      * Acknowledge a specific reply by {@code msgId} for {@code target}. Removes it from the inbox
      * so that a subsequent drain or peek no longer returns it.
+     *
+     * @return {@code true} if an entry was actually removed, {@code false} if {@code msgId} was not
+     *     in {@code target}'s inbox (wrong id, wrong target, or already acked). The caller —
+     *     {@link dev.ltms.fleet.mcp.FleetMcp#ack} — must not report success on {@code false}.
      */
-    public void ackReply(String target, String msgId) {
-        inbox.ack(target, msgId);
+    public boolean ackReply(String target, String msgId) {
+        return inbox.ack(target, msgId);
     }
 
     /**

fleetd/src/main/java/dev/ltms/fleet/msg/ReplyInbox.java

@@ -46,6 +46,13 @@ public interface ReplyInbox {
     /** Non-destructive snapshot of pending replies for {@code target} (FIFO), empty list if none. */
     List<InboxMessage> peek(String target);
 
-    /** Remove the reply {@code msgId} for {@code target} once the primary has taken it. No-op if absent. */
-    void ack(String target, String msgId);
+    /**
+     * Remove the reply {@code msgId} for {@code target} once the primary has taken it.
+     *
+     * @return {@code true} if an entry was actually removed, {@code false} if there was nothing to
+     *     remove (unknown {@code target}, unowned {@code target}, or a {@code msgId} not held for
+     *     it). A {@code false} is not an error — acking a {@code target} this daemon does not own is
+     *     part of the normal contract, not a failure.
+     */
+    boolean ack(String target, String msgId);
 }

fleetd/src/main/java/dev/ltms/fleet/peer/PeerLauncher.java

@@ -255,18 +255,7 @@ public interface PeerLauncher {
      *
      * <p>Default implementation for a launcher with no placement concept of its own: delegates to
      * {@link #spawn(SpawnRequest)} with the decision's profile named explicitly — its only spawn
-     * contract, since there is no separate routing path to honor. This default is correct ONLY for
-     * a launcher that spawns a single profile of its own (e.g. {@code HerdrPeerLauncher}), where
-     * the explicit-profile branch it re-enters and the routing branch {@link #place} would have
-     * used are the same thing. <strong>A launcher that routes across more than one profile — the
-     * way {@code CompositePeerLauncher} routes across every configured adapter — MUST override
-     * this method instead of inheriting this default.</strong> Re-entering {@link
-     * #spawn(SpawnRequest)} re-applies that single-argument method's explicit-profile checks
-     * ({@code enforceNotQuarantined}, {@code enforceNotCoolingOff}, {@code enforceMaxLoad}, {@code
-     * enforceModelEnabled} in {@code CompositePeerLauncher}), which can refuse the very profile
-     * {@link #place} just chose, if the underlying placement state moved in the window between the
-     * {@link #place} call and this one — the exact window this method and {@link PlacementDecision}
-     * exist to close (fleetd #444).
+     * contract, since there is no separate routing path to honor.
      *
      * @throws IllegalArgumentException if the decision names an unknown profile
      */

fleetd/src/test/java/dev/ltms/fleet/herdr/AgentControlContractTest.java

@@ -17,70 +17,15 @@ import static org.junit.jupiter.api.Assumptions.assumeTrue;
  * SHELL directly (never {@code claude}, so no subscription/token involvement) and always tears
  * the throwaway space down.
  *
- * <p>The seed shell's own startup (restoring its session, printing its banner) is asynchronous
- * and its length is not a fleetd contract — measured here at ~2.5s on one host (fleetd #449). A
- * fixed sleep before typing raced that startup: input typed before the shell reached its prompt
- * was swallowed by the shell's own startup, and the pane showed the typed line followed by the
- * startup banner with no command output at all — indistinguishable, at a glance, from the env
- * map never reaching the shell. So this polls for a real signal (the pane's visible text
- * settling, then the expected output appearing) instead of guessing a sleep length.
- *
  * <p>Tagged {@code contract}; run with {@code mvn test -Pcontract}.
  */
 @Tag("contract")
 class AgentControlContractTest {
 
-    private static final long POLL_INTERVAL_MS = 150;
-    /** Bound for the seed shell to settle: observed ~2.5s three times running; this leaves headroom. */
-    private static final long SHELL_READY_TIMEOUT_MS = 8_000;
-    /** Bound for the typed command's output to appear once the shell is ready: observed ~0.2s. */
-    private static final long OUTPUT_TIMEOUT_MS = 5_000;
-
     private boolean noSocket() {
         return !Files.exists(UnixSocketHerdrClient.defaultSocketPath());
     }
 
-    private static String readPane(UnixSocketHerdrClient herdr, String paneId) {
-        return herdr.call("pane.read", Map.of("pane_id", paneId, "source", "visible"))
-                .path("read").path("text").asText("");
-    }
-
-    /**
-     * Poll {@code pane.read} until two consecutive reads come back identical — the shell's own
-     * startup output (restore banner, prompt) has stopped changing — or {@code timeoutMs} elapses.
-     * Never asserts by itself; the caller's own assertion is what actually verifies the outcome,
-     * this only avoids sending input into a shell still mid-startup.
-     */
-    private static String waitUntilSettled(UnixSocketHerdrClient herdr, String paneId, long timeoutMs)
-            throws InterruptedException {
-        long deadline = System.currentTimeMillis() + timeoutMs;
-        String previous = null;
-        while (System.currentTimeMillis() < deadline) {
-            Thread.sleep(POLL_INTERVAL_MS);
-            String current = readPane(herdr, paneId);
-            if (current.equals(previous) && !current.isBlank()) {
-                return current;
-            }
-            previous = current;
-        }
-        return previous == null ? "" : previous;
-    }
-
-    /** Poll {@code pane.read} until {@code needle} appears or {@code timeoutMs} elapses. */
-    private static String waitForText(UnixSocketHerdrClient herdr, String paneId, String needle, long timeoutMs)
-            throws InterruptedException {
-        long deadline = System.currentTimeMillis() + timeoutMs;
-        String last = "";
-        while (System.currentTimeMillis() < deadline) {
-            last = readPane(herdr, paneId);
-            if (last.contains(needle)) {
-                return last;
-            }
-            Thread.sleep(POLL_INTERVAL_MS);
-        }
-        return last;
-    }
-
     @Test
     void tabCreateInjectsEnvIntoTheSeedShell() throws Exception {
         assumeTrue(!noSocket(), "no herdr socket — skipping");
@@ -91,13 +36,15 @@ class AgentControlContractTest {
                     Map.of("ANTHROPIC_BASE_URL", "http://gx00.gw:8000"));
             try {
                 assertNotNull(tab.rootPaneId(), "tab.create must return the seed pane");
-                waitUntilSettled(herdr, tab.rootPaneId(), SHELL_READY_TIMEOUT_MS);
+                Thread.sleep(1000); // let the seed shell reach its prompt
                 herdr.call("pane.send_input", Map.of(
                         "pane_id", tab.rootPaneId(),
                         "text", "printf 'PROBE_BASE=[%s]\\n' \"$ANTHROPIC_BASE_URL\"",
                         "keys", List.of("enter")));
-                String visible = waitForText(herdr, tab.rootPaneId(),
-                        "PROBE_BASE=[http://gx00.gw:8000]", OUTPUT_TIMEOUT_MS);
+                Thread.sleep(800);
+                String visible = herdr.call("pane.read",
+                                Map.of("pane_id", tab.rootPaneId(), "source", "visible"))
+                        .path("read").path("text").asText("");
                 assertTrue(visible.contains("PROBE_BASE=[http://gx00.gw:8000]"),
                         "env map must reach the seed shell; saw: " + visible);
             } finally {

fleetd/src/test/java/dev/ltms/fleet/herdr/HerdrContractTest.java

@@ -14,7 +14,7 @@ import static org.junit.jupiter.api.Assumptions.assumeTrue;
  * Contract test against a REAL running herdr. Tagged {@code contract} so it is
  * excluded from {@code mvn test}; run it with {@code mvn test -Pcontract}. It fails
  * loudly if herdr drifts from the protocol {@code fleetd} was built against
- * (0.8.0, protocol 19) — catching breakage that unit tests with canned frames cannot.
+ * (0.7.0, protocol 14) — catching breakage that unit tests with canned frames cannot.
  */
 @Tag("contract")
 class HerdrContractTest {
@@ -24,13 +24,13 @@ class HerdrContractTest {
     }
 
     @Test
-    void pingReturnsProtocol19() {
+    void pingReturnsProtocol14() {
         assumeTrue(Files.exists(socket()), "no herdr socket at " + socket() + " — skipping");
         try (UnixSocketHerdrClient herdr = UnixSocketHerdrClient.connect()) {
             JsonNode pong = herdr.call("ping");
             assertEquals("pong", pong.get("type").asText());
-            assertEquals(19, pong.get("protocol").asInt(),
-                    "fleetd is built against herdr protocol 19");
+            assertEquals(14, pong.get("protocol").asInt(),
+                    "fleetd is built against herdr protocol 14");
             assertFalse(pong.get("version").asText().isBlank());
         }
     }

fleetd/src/test/java/dev/ltms/fleet/mcp/FleetMcpTest.java

@@ -352,7 +352,7 @@ class FleetMcpTest {
                 fail("a lead fleet_reply must not publish to the worker inbox");
             }
             @Override public List<InboxMessage> peek(String target) { return List.of(); }
-            @Override public void ack(String target, String msgId) { }
+            @Override public boolean ack(String target, String msgId) { return false; }
         };
         MessageService leadMessages = new MessageService(agents, new Injector(agents), new Rendezvous(),
                 inboxThatRejectsPublishes);
@@ -1386,6 +1386,10 @@ class FleetMcpTest {
 
     @Test
     void bridgeAckReturnsConfirmationForValidArgs() {
+        // fleet_ack only reports success for a msgId actually queued in the target's inbox
+        // (fleetd #437) — publish one via the inbox directly rather than asserting on a
+        // fabricated id nothing ever queued.
+        inbox.publish("term_a", "msg-1", "queued reply");
         McpSchema.CallToolResult res = FleetMcp.ack(messages, "term_a", "msg-1");
         assertNotEquals(Boolean.TRUE, res.isError());
         assertTrue(textOf(res).contains("msg-1"), "response should mention the msgId");
@@ -1398,6 +1402,26 @@ class FleetMcpTest {
         assertTrue(FleetMcp.ack(messages, "  ", "msg-1").isError());
     }
 
+    @Test
+    void bridgeAckOfAnIdInNoInboxIsAnError() {
+        // fleetd #437: fleet_ack used to say "acknowledged <msgId>" for a message it never
+        // touched, because nothing in the chain reported hit vs. miss. "never-queued" is in no
+        // inbox at all, so this must error rather than claim success.
+        McpSchema.CallToolResult res = FleetMcp.ack(messages, "term_a", "never-queued");
+        assertTrue(res.isError());
+        assertTrue(textOf(res).contains("never-queued"), textOf(res));
+    }
+
+    @Test
+    void bridgeAckOfACoordIdTargetIsAnErrorNamingFleetPoll() {
+        // A coord-id names a peer lead's held mailbox (LeadChannel/LeadMailbox), never a
+        // worker's ReplyInbox — fleet_ack has no route to it and must say so, pointing at
+        // fleet_poll{coordId} instead of reporting a false "acknowledged".
+        McpSchema.CallToolResult res = FleetMcp.ack(messages, "coord-some-peer", "msg-1");
+        assertTrue(res.isError());
+        assertTrue(textOf(res).contains("fleet_poll{coordId}"), textOf(res));
+    }
+
     @Test
     void bridgeAckRemovesSpecificReply() {
         // Queue a reply and capture its msgId.
@@ -1411,8 +1435,18 @@ class FleetMcpTest {
         var peeked = messages.drainReplies("term_a");
         assertEquals(1, peeked.size(), "one fresh reply in the inbox");
 
-        // ackReply works (no-op since published with a different UUID, but callable).
-        assertDoesNotThrow(() -> messages.ackReply("term_a", msgId));
+        // fleetd #437: msgId was already drained above (a fresh UUID each publish), so it is no
+        // longer in the inbox — ackReply must now report that miss instead of pretending to ack.
+        assertFalse(messages.ackReply("term_a", msgId));
+    }
+
+    @Test
+    void bridgeAckRemovingARealQueuedReplyReportsSuccessAndRemovesIt() {
+        // The worker path must not change behaviour: acking a reply that IS still in the inbox
+        // still succeeds and still removes it (fleetd #437).
+        inbox.publish("term_a", "real-1", "still queued");
+        assertTrue(messages.ackReply("term_a", "real-1"), "ack of a real queued reply must report true");
+        assertTrue(inbox.peek("term_a").isEmpty(), "the acked reply must be gone from the inbox");
     }
 
     @Test

fleetd/src/test/java/dev/ltms/fleet/member/CompositePeerLauncherTest.java

@@ -20,7 +20,6 @@ import dev.ltms.fleet.peer.PeerUnreachableException;
 import dev.ltms.fleet.peer.SpawnRequest;
 import dev.ltms.fleet.placement.BackendOutagePolicy;
 import dev.ltms.fleet.placement.BackendQuarantine;
-import dev.ltms.fleet.placement.PlacementDecision;
 import dev.ltms.fleet.placement.PlacementException;
 import dev.ltms.fleet.placement.PlacementPolicies;
 import org.junit.jupiter.api.Test;
@@ -1124,68 +1123,6 @@ class CompositePeerLauncherTest {
         assertEquals(0, adapter.spawnCount("b"), "routedProfileFor never spawns anything");
     }
 
-    /**
-     * fleetd #444: {@link PlacementDecision} exists to close the window between {@link
-     * CompositePeerLauncher#place} and {@link CompositePeerLauncher#spawn(SpawnRequest,
-     * PlacementDecision)} — the placement state must be free to move in that window without the
-     * held decision being re-checked against the new state. Every quarantine test above resolves
-     * and spawns in one call, so none of them ever open that window; this test is the one that
-     * does: "sol" is placed FIRST, while nothing is quarantined yet, and only THEN is its
-     * credential quarantined, before the held decision is spawned.
-     *
-     * <p>This is the test that tells the real override apart from the alternative body the ticket
-     * measured: routing {@code decision.profile()} straight to its adapter (the real override)
-     * never re-runs {@code enforceNotQuarantined}, so the spawn against the held decision still
-     * succeeds on sol. Re-entering {@code spawn(req.withProfile(decision.profile()))} instead
-     * lands in the explicit-profile branch, which refuses a now-quarantined sol outright — before
-     * this test existed, replacing the real override's body with that re-entering call left the
-     * whole suite green.
-     */
-    @Test
-    void spawnHonorsAPlacementDecisionEvenAfterItsProfileIsQuarantinedInTheWindowAfterPlace() {
-        FakeHerdr herdr = new FakeHerdr();
-        Map<String, FleetConfig.Profile> profiles = ordered(
-                "sol", stubWorker("sol", "shared-openai"),
-                "b", stubWorker("b"));
-        // The adapter's OWN fallback default is "b", deliberately different from the profile place()
-        // decides ("sol") — see the note below on why this must not be "sol" too.
-        StubLauncher adapter = new StubLauncher("claude", herdr, profiles, "b", Set.of());
-        BackendQuarantine quarantine = new BackendQuarantine(() -> 0L, TimeUnit.MINUTES.toNanos(30));
-        CompositePeerLauncher composite = new CompositePeerLauncher(List.of(adapter), "sol", profiles,
-                PlacementPolicies.fixed(), _ -> 0, null, quarantine);
-
-        // 1. Resolve BEFORE anything is quarantined — sol (definition order first, fixed policy) wins.
-        // composite's own defaultProfile ("sol", the constructor arg above) never enters this: the
-        // pool poolFor(DEV) resolves to is never empty here, so place() only ever reads that field as
-        // a fallback for an empty pool, which this test does not exercise.
-        PlacementDecision decision = composite.place(MemberRole.DEV);
-        assertEquals("sol", decision.profile(), "sanity: nothing is quarantined yet, so sol is placed");
-
-        // 2. Move the placement state IN THE WINDOW between place() and spawn() — sol's credential
-        // is now quarantined. A fresh place()/spawn(req) pair would fall through to b instead; the
-        // held decision must not be re-evaluated against this new state at all.
-        quarantine.quarantine("shared-openai");
-
-        // 3. Spawn against the HELD decision, not a fresh resolve.
-        SpawnRequest req = new SpawnRequest(null, null, null, null, null, MemberRole.DEV);
-        PeerHandle handle = composite.spawn(req, decision);
-
-        assertEquals("sol", handle.profile(),
-                "the decision from place() is honored even though sol is now quarantined");
-        // A fixture whose adapter falls back to "sol" too would let an UNSTAMPED request (one
-        // routed but never given req.withProfile("sol")) land on spawnCount("sol") == 1 by
-        // COINCIDENCE, since StubLauncher.spawn falls back to its own defaultProfile whenever
-        // req.profileName() is blank. Giving the adapter "b" as its fallback instead means only an
-        // actually-stamped request can produce this count — an unstamped one would count against
-        // "b" and this assertion would fail.
-        assertEquals(1, adapter.spawnCount("sol"),
-                "the request that reached the delegate actually carried sol as its profile "
-                        + "(the adapter's own fallback default is 'b', so this can't happen by accident)");
-        assertEquals(0, adapter.spawnCount("b"),
-                "b must never be touched — neither as the decision's profile nor as an unstamped "
-                        + "request's accidental fallback");
-    }
-
     @Test
     void aQuarantineLiftsOnTheInjectedClockAndTheProfileBecomesSpawnableAgain() {
         FakeHerdr herdr = new FakeHerdr();

fleetd/src/test/java/dev/ltms/fleet/msg/AmqpReplyInboxContractTest.java

@@ -16,6 +16,7 @@ import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -221,6 +222,31 @@ class AmqpReplyInboxContractTest {
         }
     }
 
+    @Test
+    void ackReportsHitVsMissAgainstARealBroker() throws Exception {
+        // fleetd #437: fleet_ack said "acknowledged <msgId>" for a message it never touched,
+        // because ReplyInbox.ack() (void) could not tell a hit from a miss. Pin the fixed
+        // boolean contract against a real broker — the adapter fleetd actually runs live.
+        String target = "worker-ack-contract-" + System.nanoTime();
+        try (AmqpReplyInbox inbox = AmqpReplyInbox.open(uri())) {
+            inbox.own(target);
+
+            // Never held for this target at all: must report false, not throw.
+            assertFalse(inbox.ack(target, "never-held"),
+                    "acking a msgId never held for an owned target must report false");
+
+            // A real message: first ack removes it and reports true...
+            inbox.publish(target, "m1", "ack me");
+            assertEquals(1, awaitPeek(inbox, target).size(), "the published reply should be held");
+            assertTrue(inbox.ack(target, "m1"), "acking a held reply must report true");
+            assertTrue(inbox.peek(target).isEmpty(), "an acked reply is dropped");
+
+            // ...and the second ack of the SAME msgId has nothing left to remove: false.
+            assertFalse(inbox.ack(target, "m1"),
+                    "acking the same msgId twice must report false the second time");
+        }
+    }
+
     @Test
     void confirmedPublishDeliversNormally() throws Exception {
         String target = "worker-confirm-" + System.nanoTime();

fleetd/src/test/java/dev/ltms/fleet/msg/InMemoryReplyInboxTest.java

@@ -41,20 +41,20 @@ class InMemoryReplyInboxTest {
     @Test
     void ackRemovesTheMessage() {
         inbox.publish("term_a", "m1", "hello");
-        inbox.ack("term_a", "m1");
+        assertTrue(inbox.ack("term_a", "m1"), "fleetd #437: ack of a real entry must report true");
         assertTrue(inbox.peek("term_a").isEmpty(), "after ack, the message is gone");
     }
 
     @Test
     void ackForUnknownMsgIdIsNoOp() {
         inbox.publish("term_a", "m1", "hello");
-        inbox.ack("term_a", "no-such-id"); // no-op
+        assertFalse(inbox.ack("term_a", "no-such-id"), "fleetd #437: a miss must report false"); // no-op
         assertEquals(1, inbox.peek("term_a").size(), "the published message is still there");
     }
 
     @Test
     void ackForUnknownTargetIsNoOp() {
-        inbox.ack("no-such-target", "m1"); // no-op, should not throw
+        assertFalse(inbox.ack("no-such-target", "m1"), "fleetd #437: a miss must report false"); // no-op, should not throw
     }
 
     @Test
@@ -167,7 +167,7 @@ class InMemoryReplyInboxTest {
     @Test
     void peekAndAckAreNoOpsForUnownedTarget() {
         assertTrue(inbox.peek("term_not_owned").isEmpty());
-        inbox.ack("term_not_owned", "m1"); // no-op, should not throw
+        assertFalse(inbox.ack("term_not_owned", "m1"), "fleetd #437: a miss must report false"); // no-op, should not throw
     }
 
     @Test