fleetd #656: add regression tests for the two cases #639's redact() fix covers

Criterion 19 covers a block-scalar body whose key line falls outside
diff -u's default 3-line context (an 8-line body with only the 6th
line changed). Criterion 20 covers a blank line inside the value,
which used to reset the old indentation-anchored mask.

Both are RED against the pre-#639 redact() (git show 28ea0de) and
GREEN against the current one; each asserts both the secret's
absence and a non-secret control line's presence.

fleetd/src/test/java/dev/ltms/fleet/FleetdAssemblyFleetAppTest.java

@@ -58,20 +58,31 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
  * {@code HttpClient} — no accessor needed for this half.
  *
  * <p><strong>{@code GET /sessions} could not be driven the same way</strong>, so this class does
- * not pin the merge half of the deleted test's javadoc. {@code /sessions} requires
- * {@code Authz.Action.READ}, which — through the REAL assembly's real {@code
- * CallerResolver}/{@code ConnectionIdentity} (built with a hardcoded {@code
- * new LsofPeerPidLookup()}) — needs {@code Caller.resolved()}, i.e. a real positive pid from
- * {@code lsof}. {@code LsofPeerPidLookup} excludes its own pid (see its javadoc), and a JUnit
- * test's HTTP client and the daemon under test share one JVM pid, so the resolved pid is always
- * {@code -1} and every such request is refused as {@code ANONYMOUS} (fleetd #317's fail-closed
- * rule) before the route handler — and its {@code memberHerdr} merge — is ever reached. Verified
- * directly: driving {@code GET /sessions} here returns {@code 401 unauthenticated}, not the
- * merged body. {@code FleetAppTwoDaemonTest} avoids this because it builds {@code FleetApp} with
- * {@code callers: null}, which is not what the real assembly passes. The {@code /healthz} pin
- * below is what this class relies on for CB-185's {@code FleetApp} half; {@code
- * FleetAppTwoDaemonTest} remains the full behavioural proof that {@code FleetApp} itself merges
- * {@code /sessions} correctly once handed two clients.
+ * not pin the merge half of the deleted test's javadoc. This class configures no {@code auth:}
+ * block, so it runs under the default {@code loopback-trust} mode ({@code FleetConfig}). Under
+ * that mode, {@code /sessions} requires {@code Authz.Action.READ}, which — through the REAL
+ * assembly's real {@code CallerResolver}/{@code ConnectionIdentity} (built with a hardcoded
+ * {@code new LsofPeerPidLookup()}) — needs {@code Caller.resolved()}, i.e. a real positive pid
+ * from {@code lsof}. {@code LsofPeerPidLookup} excludes its own pid (see its javadoc), and a
+ * JUnit test's HTTP client and the daemon under test share one JVM pid, so the resolved pid is
+ * always {@code -1} and every such request is refused as {@code ANONYMOUS} (fleetd #317's
+ * fail-closed rule) before the route handler — and its {@code memberHerdr} merge — is ever
+ * reached. Verified directly: driving {@code GET /sessions} here returns {@code 401
+ * unauthenticated}, not the merged body. {@code FleetAppTwoDaemonTest} avoids this because it
+ * builds {@code FleetApp} with {@code callers: null}, which is not what the real assembly
+ * passes. The {@code /healthz} pin below is what this class relies on for CB-185's {@code
+ * FleetApp} half; {@code FleetAppTwoDaemonTest} remains the full behavioural proof that
+ * {@code FleetApp} itself merges {@code /sessions} correctly once handed two clients.
+ *
+ * <p><strong>This refusal is {@code loopback-trust}-specific, not a property of {@code
+ * CallerResolver} in general.</strong> Under {@code auth.mode: token}, {@code
+ * CallerResolver#resolve} returns before ever consulting {@code Caller.resolved()} or {@code
+ * Caller.scanComplete()}: a request carrying a valid bearer token in its {@code Authorization}
+ * header resolves to {@code Role#PRIMARY} with no pid lookup at all, so the same-JVM-pid
+ * exclusion above never comes into play. {@code FleetdQuarantineOutageDualWindowAssemblyTest}
+ * and {@code FleetdListReportingSourcesAssemblyTest} both drive {@code Authz.Action.READ} this
+ * way, over a real {@code McpSyncClient}/{@code HttpClient} against a real {@code
+ * FleetdAssembly#assembleAndStart}, and both get the real response rather than a refusal.
  *
  * <p><strong>fleetd #629 follow-up.</strong> The fix below (see {@link TwoHerdrResourcePorts})
  * makes {@link #healthzGoesRedWhenTheLeadDaemonIsDownEvenThoughTheMemberIsUp}'s fake {@code

fleetd/src/test/java/dev/ltms/fleet/FleetdAssemblyLoopHealthTest.java

@@ -0,0 +1,180 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.HerdrClient;
+import dev.ltms.fleet.inject.LoopWatchdog;
+import dev.ltms.fleet.mcp.FleetMcp;
+import io.javalin.Javalin;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.reflect.Field;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.LongSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #612 Shape A, rank 10: {@link FleetdAssembly} creates one {@link FleetMcp.LoopHealthSource}
+ * from the real started {@code StatusPoller} and {@code SessionReaper}, then gives it to two operator
+ * windows. These tests reach the real assembled objects through {@link FleetdRuntime}, rather than
+ * building a second source beside them. A hardcoded {@code RUNNING} source would pass a simple
+ * "running" test, so the mutation proof also mis-wires the source to never-started loops: both
+ * windows must then report {@code STOPPED} and these assertions go red.
+ */
+class FleetdAssemblyLoopHealthTest {
+
+    private static final class RecordingResourcePorts implements ResourcePorts {
+        final FakeHerdr herdr = new FakeHerdr();
+        Runnable shutdownHook;
+
+        @Override
+        public Map<String, String> environment() {
+            return Map.of();
+        }
+
+        @Override
+        public HerdrClient connectHerdr(Path socketPath) {
+            return herdr;
+        }
+
+        @Override
+        public Fleetd.AmqpOpener replyInboxOpener() {
+            return (uri, prefetch) -> new dev.ltms.fleet.msg.InMemoryReplyInbox();
+        }
+
+        @Override
+        public Fleetd.LeadMailboxOpener leadMailboxOpener() {
+            return (uri, selfCoordId, prefetch) -> {
+                throw new UnsupportedOperationException("no coordinator is configured");
+            };
+        }
+
+        @Override
+        public Runnable herdrPollWait() {
+            return () -> {
+                throw new UnsupportedOperationException("healthy FakeHerdr must not be polled");
+            };
+        }
+
+        @Override
+        public LongSupplier nanoClock() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public LongSupplier wallClockNanos() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public ScheduledExecutorService newScheduler(String purpose) {
+            return Executors.newSingleThreadScheduledExecutor();
+        }
+
+        @Override
+        public void addShutdownHook(Runnable hook) {
+            shutdownHook = hook;
+        }
+
+        @Override
+        public void startHttp(Javalin app, String host, int port) {
+            // Bind runtime.app() to an ephemeral port only in the REST assertion below.
+        }
+    }
+
+    private final HttpClient http = HttpClient.newHttpClient();
+    private RecordingResourcePorts ports;
+    private FleetdRuntime runtime;
+    private Javalin boundApp;
+
+    @AfterEach
+    void tearDown() {
+        if (boundApp != null) {
+            boundApp.stop();
+        }
+        if (runtime != null) {
+            runtime.close();
+        }
+        if (ports != null) {
+            assertNotNull(ports.shutdownHook, "the assembly must register its shutdown hook");
+            assertTrue(ports.herdr.closed, "FleetdRuntime.close must close the real assembled herdr client");
+        }
+    }
+
+    private static FleetConfig writeConfig(Path dir) throws Exception {
+        Path file = dir.resolve("fleetd.yaml");
+        Files.writeString(file, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                lifecycle:
+                  idleTtlSeconds: 600
+                idleSleepGuard:
+                  enabled: false
+                health:
+                  enabled: false
+                broker:
+                  uri: "amqp://fake-test-broker/vh"
+                """);
+        return FleetConfig.load(file);
+    }
+
+    private void assemble(Path dir) throws Exception {
+        FleetConfig cfg = writeConfig(dir);
+        ports = new RecordingResourcePorts();
+        runtime = FleetdAssembly.assembleAndStart(
+                new AssemblyInputs(cfg, new ConfigRef(dir.resolve("fleetd.yaml"), cfg),
+                        new SubscriptionGuard(cfg.guard().hostSet())), ports);
+    }
+
+    @Test
+    void fleetListUsesTheRunningLoopsInTheRealAssembledMcp(@TempDir Path dir) throws Exception {
+        assemble(dir);
+
+        // The field is the exact source captured by FleetMcp's fleet_list handler. Reflection is
+        // necessary because FleetMcp has no public source accessor; it is not a source-text check.
+        FleetMcp.LoopHealthSource loopHealth = loopHealthOf(runtime.mcp());
+        assertRunning(loopHealth, "FleetMcp's real fleet_list source");
+    }
+
+    @Test
+    void healthzUsesTheRunningLoopsInTheRealAssembledApp(@TempDir Path dir) throws Exception {
+        assemble(dir);
+        boundApp = runtime.app().start("127.0.0.1", 0);
+
+        HttpRequest request = HttpRequest.newBuilder(
+                URI.create("http://127.0.0.1:" + boundApp.port() + "/healthz")).GET().build();
+        HttpResponse<String> response = http.send(request, HttpResponse.BodyHandlers.ofString());
+        assertEquals(200, response.statusCode(), response.body());
+        assertTrue(response.body().contains("\"statusPoller\":\"RUNNING\""), response.body());
+        assertTrue(response.body().contains("\"sessionReaper\":\"RUNNING\""), response.body());
+    }
+
+    private static FleetMcp.LoopHealthSource loopHealthOf(FleetMcp mcp) throws Exception {
+        Field field = FleetMcp.class.getDeclaredField("loopHealth");
+        field.setAccessible(true);
+        return (FleetMcp.LoopHealthSource) field.get(mcp);
+    }
+
+    private static void assertRunning(FleetMcp.LoopHealthSource source, String consumer) {
+        assertEquals(LoopWatchdog.State.RUNNING, source.statusPoller().get(),
+                consumer + " must report the started real StatusPoller as RUNNING");
+        assertEquals(LoopWatchdog.State.RUNNING, source.sessionReaper().get(),
+                consumer + " must report the started real SessionReaper as RUNNING");
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/FleetdAssemblyTurnRegistrarBehaviouralTest.java

@@ -0,0 +1,184 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.AgentStatus;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.HerdrClient;
+import dev.ltms.fleet.inject.Injector;
+import dev.ltms.fleet.inject.TurnListener;
+import dev.ltms.fleet.msg.Rendezvous;
+import dev.ltms.fleet.msg.TurnToken;
+import io.javalin.Javalin;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.reflect.Field;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #612 rank 12 — {@code FleetdAssembly} supplies the {@link Injector}'s {@code TurnRegistrar}
+ * with {@code Fleetd.turnRegistrar(completion)}. The normal delivery path cannot distinguish that
+ * registrar from {@code TurnRegistrar.NOOP}: {@link dev.ltms.fleet.inject.CompletionResolver#onDelivered}
+ * registers the same turn shortly afterwards. The distinction matters when a delivery listener throws
+ * after the pane received the message but before normal completion runs. The registrar must already have
+ * registered the waiter with the REAL assembled resolver, so a later completion can still resolve it.
+ *
+ * <p>The test installs a throwing wrapper around the real assembled listener. It does not replace the
+ * registrar or resolver. This constructs the narrow failure condition without a sleep, then observes the
+ * real resolver through {@link FleetdRuntime#completion()}. An inert registrar, or a registrar wired to a
+ * throwaway resolver, leaves this waiter's turn absent from the real resolver and makes the final assertion
+ * fail.
+ */
+class FleetdAssemblyTurnRegistrarBehaviouralTest {
+
+    private static final String TARGET = "term_a";
+
+    private static final class ControllableResourcePorts implements ResourcePorts {
+        final FakeHerdr herdr = new FakeHerdr();
+        final AtomicLong nowNanos = new AtomicLong(1_000_000_000L);
+        Runnable shutdownHook;
+
+        void advanceSeconds(long seconds) {
+            nowNanos.addAndGet(TimeUnit.SECONDS.toNanos(seconds));
+        }
+
+        @Override
+        public Map<String, String> environment() {
+            return Map.of();
+        }
+
+        @Override
+        public HerdrClient connectHerdr(Path socketPath) {
+            return herdr;
+        }
+
+        @Override
+        public Fleetd.AmqpOpener replyInboxOpener() {
+            return (uri, prefetch) -> {
+                throw new UnsupportedOperationException("no broker: block is configured");
+            };
+        }
+
+        @Override
+        public Fleetd.LeadMailboxOpener leadMailboxOpener() {
+            return (uri, selfCoordId, prefetch) -> {
+                throw new UnsupportedOperationException("no coordinator: block is configured");
+            };
+        }
+
+        @Override
+        public LongSupplier nanoClock() {
+            return nowNanos::get;
+        }
+
+        @Override
+        public LongSupplier wallClockNanos() {
+            return nowNanos::get;
+        }
+
+        @Override
+        public ScheduledExecutorService newScheduler(String purpose) {
+            return Executors.newSingleThreadScheduledExecutor();
+        }
+
+        @Override
+        public void addShutdownHook(Runnable hook) {
+            shutdownHook = hook;
+        }
+
+        @Override
+        public void startHttp(Javalin app, String host, int port) {
+            // Do not bind a real port in this assembly test.
+        }
+
+        @Override
+        public Runnable herdrPollWait() {
+            return () -> {
+                throw new UnsupportedOperationException("FakeHerdr is healthy; no poll wait is expected");
+            };
+        }
+    }
+
+    private static FleetConfig writeConfig(Path dir) throws Exception {
+        Path file = dir.resolve("fleetd.yaml");
+        Files.writeString(file, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                idleSleepGuard:
+                  enabled: false
+                fleet:
+                  leaders:
+                    primary:
+                      tab: "lead: primary"
+                      profile: sonnet
+                profiles:
+                  sonnet:
+                    subscription: true
+                    argv: ["ccs", "sonnet"]
+                """);
+        return FleetConfig.load(file);
+    }
+
+    @Test
+    void realAssembledResolverStillResolvesAfterDeliveredListenerThrows(@TempDir Path dir) throws Exception {
+        FleetConfig cfg = writeConfig(dir);
+        ControllableResourcePorts ports = new ControllableResourcePorts();
+        ports.herdr.withTab("w2", "w2:t7", "lead: primary");
+        FleetdRuntime runtime = FleetdAssembly.assembleAndStart(new AssemblyInputs(cfg,
+                new ConfigRef(dir.resolve("fleetd.yaml"), cfg), new SubscriptionGuard(cfg.guard().hostSet())), ports);
+        try {
+            Injector injector = runtime.injector();
+            installThrowingDeliveredListener(injector);
+
+            CompletableFuture<Rendezvous.Resolution> waiter = new CompletableFuture<>();
+            injector.enqueue(TARGET, "brief", new TurnToken(TARGET, waiter));
+
+            IllegalStateException thrown = assertThrows(IllegalStateException.class,
+                    () -> injector.onStatus(TARGET, AgentStatus.IDLE),
+                    "control: delivery must reach the installed listener and it must throw after delivery");
+            assertTrue(thrown.getMessage().contains("listener failure"), thrown::getMessage);
+
+            ports.herdr.readText("worker report after the listener failure");
+            ports.advanceSeconds(3);
+            runtime.completion().resolveBeforePostAction(TARGET);
+
+            assertTrue(waiter.isDone(),
+                    "FleetdAssembly must wire the Injector registrar to this runtime's real CompletionResolver: "
+                            + "after a delivered listener throws, resolveBeforePostAction must still find and "
+                            + "resolve the registered waiter");
+        } finally {
+            assertTrue(ports.shutdownHook != null, "control: assembly must capture its shutdown hook");
+            ports.shutdownHook.run();
+            assertTrue(ports.herdr.closed, "teardown control: the captured shutdown hook must close herdr");
+        }
+    }
+
+    private static void installThrowingDeliveredListener(Injector injector) throws Exception {
+        Field field = Injector.class.getDeclaredField("turnListener");
+        field.setAccessible(true);
+        field.set(injector, new TurnListener() {
+            @Override
+            public void onTurnComplete(String target) {
+            }
+
+            @Override
+            public void onDelivered(String target, TurnToken token) {
+                throw new IllegalStateException("listener failure after delivery");
+            }
+        });
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/FleetdConfigRefWiringTest.java

@@ -24,8 +24,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
  * {@code ConfigRefTest} and {@code FleetdConfigRefCharterToolSurfaceWiringTest} case — because
  * neither of those tests constructs its {@code ConfigRef} through {@code main}; both build their own
  * instance directly, wired with the check by hand. That silent regression is exactly the shape
- * {@link FleetdBackendQuarantineWiringTest}, {@link FleetdLeadSeatWiringTest} and {@link
- * FleetdCompletionResolverWiringTest} already guard against for their own constructor arguments —
+ * {@link FleetdBackendQuarantineAssemblyTest}, {@link FleetdLeadSeatAssemblyTest} and {@link
+ * FleetdCompletionResolverAssemblyTest} already guard against for their own constructor arguments —
  * this class is the same class of gap for fleetd #474's {@code extraValidation} argument, following
  * their approach.
  *

fleetd/src/test/java/dev/ltms/fleet/FleetdHerdrControlConstructionTest.java

@@ -2,16 +2,67 @@ package dev.ltms.fleet;
 
 import java.nio.file.Files;
 import java.nio.file.Path;
+import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+/**
+ * {@code AgentControl} caches {@code paneByTerminal}, so {@code HerdrRouter} must be its only
+ * production factory — a second instance means a second cache; the same reasoning applies to
+ * {@code WorkspaceControl}. {@code HerdrRouter}'s constructor is the one place both are built.
+ *
+ * <p><b>This test checks source text, not runtime behaviour.</b> It never constructs a {@code
+ * HerdrRouter} and never runs {@code FleetdAssembly.assembleAndStart} — a green result proves only
+ * that neither watched file's text contains {@code new AgentControl(} or {@code new
+ * WorkspaceControl(}. It does not prove the instances {@code HerdrRouter} does build are the ones
+ * actually wired through the rest of the daemon, and it does not cover a bypass written into a
+ * production file other than the two this test reads.
+ */
 class FleetdHerdrControlConstructionTest {
+
+    private static String source(String relativePath) throws Exception {
+        return Files.readString(Path.of(relativePath));
+    }
+
     @Test
+    @DisplayName("[SOURCE TEXT] Fleetd.java never constructs AgentControl or WorkspaceControl directly")
     void fleetdDelegatesStatefulControlsToTheRouter() throws Exception {
-        // AgentControl caches paneByTerminal, so the router must be its only production factory.
-        String source = Files.readString(Path.of("src/main/java/dev/ltms/fleet/Fleetd.java"));
-        assertFalse(source.contains("new AgentControl("));
-        assertFalse(source.contains("new WorkspaceControl("));
+        String source = source("src/main/java/dev/ltms/fleet/Fleetd.java");
+
+        // A broken read (wrong working directory, wrong path, a file that came back empty) would
+        // make the assertFalse checks below pass vacuously — a "clean" negative check that actually
+        // checked nothing. Guard against that first, with an anchor that has nothing to do with
+        // this mutation, so a bad read fails loudly here instead of silently proving nothing below.
+        assertTrue(source.contains("public final class Fleetd"),
+                "the read of Fleetd.java did not come back containing its own class declaration — "
+                        + "the assertFalse checks below would pass vacuously on a broken read; fix the "
+                        + "read before trusting this test.");
+
+        assertFalse(source.contains("new AgentControl("),
+                "Fleetd.java must not construct AgentControl directly — HerdrRouter is its only "
+                        + "production factory");
+        assertFalse(source.contains("new WorkspaceControl("),
+                "Fleetd.java must not construct WorkspaceControl directly — HerdrRouter is its only "
+                        + "production factory");
+    }
+
+    @Test
+    @DisplayName("[SOURCE TEXT] FleetdAssembly.java never constructs AgentControl or WorkspaceControl directly")
+    void fleetdAssemblyDelegatesStatefulControlsToTheRouter() throws Exception {
+        String source = source("src/main/java/dev/ltms/fleet/FleetdAssembly.java");
+
+        assertTrue(source.contains("final class FleetdAssembly"),
+                "the read of FleetdAssembly.java did not come back containing its own class "
+                        + "declaration — the assertFalse checks below would pass vacuously on a broken "
+                        + "read; fix the read before trusting this test.");
+
+        assertFalse(source.contains("new AgentControl("),
+                "FleetdAssembly.java must not construct AgentControl directly — HerdrRouter is its "
+                        + "only production factory");
+        assertFalse(source.contains("new WorkspaceControl("),
+                "FleetdAssembly.java must not construct WorkspaceControl directly — HerdrRouter is "
+                        + "its only production factory");
     }
 }

fleetd/src/test/java/dev/ltms/fleet/FleetdLeadConfigDirSourceAssemblyTest.java

@@ -0,0 +1,213 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.HerdrClient;
+import dev.ltms.fleet.mcp.FleetMcp;
+import dev.ltms.fleet.msg.ReplyInbox;
+import io.javalin.Javalin;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.reflect.Field;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.LongSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * fleetd #612 Shape A, unit r5 — {@code FleetdAssembly.java:488} wires {@link
+ * FleetMcp.LeadConfigDirSource} with {@code Fleetd.leadConfigDirSource(() -> config.get().profiles(),
+ * leaders)}. {@link FleetdLeadConfigDirSourceWiringTest} already pins that the FACTORY itself
+ * delegates to the real {@link Fleetd#leadConfigDirLookup} — but, by its own javadoc, it "does not
+ * and structurally cannot cover" whether the real call site in {@code FleetdAssembly} still calls
+ * that factory at all. Measured there: swapping that one-line call for a bare {@code
+ * FleetMcp.LeadConfigDirSource.none()} compiles with 0 errors and leaves the full suite green.
+ *
+ * <p>This is the literal fleetd #602/#606 defect, one call site away from its own fix: {@code main}
+ * (now {@code FleetdAssembly}) used to build {@code LeadConfigDirSource.none()} inline, the whole
+ * suite passed, and the live daemon reported {@code "state":"unknown"} for every lead's context,
+ * forever, with no test noticing. The fix extracted the factory; this test is the one that proves
+ * {@code FleetdAssembly}'s own call site still reaches it.
+ *
+ * <p>This test drives the REAL {@link FleetMcp} the real {@link FleetdAssembly#assembleAndStart}
+ * builds, reached through {@link FleetdRuntime#mcp()}, and reads the {@code leadConfigDirs} field it
+ * was constructed with via reflection — {@code FleetMcp} exposes no public accessor for it (unlike
+ * {@code quarantineSource()}/{@code leadSeatSource()}), so there is no non-reflective route to the
+ * live instance. The assertion resolves a REAL lead name against a REAL configured {@code
+ * configDir:}: {@link FleetMcp.LeadConfigDirSource#none()} (the historical defect, and the
+ * mis-wire this test's mutation cycles reintroduce) always returns {@code null} regardless of the
+ * input, so a non-null, config-matching answer is a property {@code none()} can never produce by
+ * accident.
+ */
+class FleetdLeadConfigDirSourceAssemblyTest {
+
+    private static final String LEAD_NAME = "opus";
+    private static final String LEAD_TAB = "lead: opus";
+    private static final String LEAD_PROFILE = "sonnet";
+
+    private static final class RecordingResourcePorts implements ResourcePorts {
+
+        final FakeHerdr herdr = new FakeHerdr();
+        final SentinelReplyInbox replyInbox = new SentinelReplyInbox();
+
+        @Override
+        public Map<String, String> environment() {
+            return Map.of();
+        }
+
+        @Override
+        public HerdrClient connectHerdr(Path socketPath) {
+            return herdr;
+        }
+
+        @Override
+        public Fleetd.AmqpOpener replyInboxOpener() {
+            return (uri, prefetch) -> replyInbox;
+        }
+
+        @Override
+        public Fleetd.LeadMailboxOpener leadMailboxOpener() {
+            return (uri, selfCoordId, prefetch) -> {
+                throw new UnsupportedOperationException(
+                        "leadMailboxOpener must not be called — no coordinator: block is configured");
+            };
+        }
+
+        @Override
+        public LongSupplier nanoClock() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public LongSupplier wallClockNanos() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public ScheduledExecutorService newScheduler(String purpose) {
+            return Executors.newSingleThreadScheduledExecutor();
+        }
+
+        @Override
+        public void addShutdownHook(Runnable hook) {
+        }
+
+        @Override
+        public void startHttp(Javalin app, String host, int port) {
+        }
+
+        @Override
+        public Runnable herdrPollWait() {
+            // Never invoked: this test's FakeHerdr answers immediately, so awaitHerdr never polls.
+            return () -> {
+                throw new UnsupportedOperationException("herdrPollWait must not be called — herdr is healthy");
+            };
+        }
+    }
+
+    private static final class SentinelReplyInbox implements ReplyInbox, AutoCloseable {
+        @Override
+        public void own(String target) {
+        }
+
+        @Override
+        public void release(String target) {
+        }
+
+        @Override
+        public void publish(String target, String msgId, String content) {
+        }
+
+        @Override
+        public List<InboxMessage> peek(String target) {
+            return List.of();
+        }
+
+        @Override
+        public boolean ack(String target, String msgId) {
+            return false;
+        }
+
+        @Override
+        public void close() {
+        }
+    }
+
+    private static FleetConfig writeConfig(Path dir, String configDir) throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        Files.writeString(f, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                idleSleepGuard:
+                  enabled: false
+                broker:
+                  uri: "amqp://fake-test-broker/vh"
+                fleet:
+                  leaders:
+                    %s:
+                      tab: "%s"
+                      profile: %s
+                profiles:
+                  %s:
+                    subscription: true
+                    argv: ["ccs", "sonnet"]
+                    configDir: "%s"
+                """.formatted(LEAD_NAME, LEAD_TAB, LEAD_PROFILE, LEAD_PROFILE, configDir));
+        return FleetConfig.load(f);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static FleetMcp.LeadConfigDirSource leadConfigDirSourceOf(FleetMcp mcp) throws Exception {
+        Field field = FleetMcp.class.getDeclaredField("leadConfigDirs");
+        field.setAccessible(true);
+        return (FleetMcp.LeadConfigDirSource) field.get(mcp);
+    }
+
+    @Test
+    @DisplayName("[BEHAVIOURAL] the real assembled LeadConfigDirSource resolves a lead's REAL "
+            + "configured configDir, not the none() stand-in's hardcoded null")
+    void assembledLeadConfigDirSourceResolvesTheRealConfiguredConfigDir(@TempDir Path dir) throws Exception {
+        String configuredConfigDir = "/mnt/fake-lead-configdir";
+        FleetConfig cfg = writeConfig(dir, configuredConfigDir);
+        ConfigRef config = new ConfigRef(dir.resolve("fleetd.yaml"), cfg);
+        SubscriptionGuard guard = new SubscriptionGuard(cfg.guard().hostSet());
+        RecordingResourcePorts ports = new RecordingResourcePorts();
+        // Label FakeHerdr's own default pane's tab (term_a / w2:p7 / w2:t7, already carrying a live
+        // agent) to match fleet.leaders.opus.tab exactly, so LeadLauncher.ensureLeads() sees the
+        // lead as already live and does not try to auto-launch a second one.
+        ports.herdr.withTab("w2", "w2:t7", LEAD_TAB);
+
+        FleetdRuntime runtime = FleetdAssembly.assembleAndStart(new AssemblyInputs(cfg, config, guard), ports);
+        try {
+            FleetMcp.LeadConfigDirSource source = leadConfigDirSourceOf(runtime.mcp());
+
+            assertEquals(configuredConfigDir, source.configDirFor().apply(LEAD_NAME),
+                    "fleet.leaders." + LEAD_NAME + ".profile (" + LEAD_PROFILE + ") configures "
+                            + "configDir: " + configuredConfigDir + " — the real assembled source must "
+                            + "resolve it. FleetMcp.LeadConfigDirSource.none() (the inert stand-in "
+                            + "this test's mutation cycles swap the call site for, and the historical "
+                            + "fleetd #602/#606 defect) always reports null here, whatever the input");
+
+            // A lead name the config does not recognise still resolves to null, not a crash — the
+            // same source, applied to an input that must stay at the inert answer even on the real,
+            // non-inert instance.
+            assertNull(source.configDirFor().apply("no-such-lead"));
+        } finally {
+            // Surefire runs the whole suite in one JVM fork (fleetd/pom.xml sets no forkCount /
+            // reuseForks), so the scheduler/loops this assembly starts must be torn down here, on the
+            // failure path too — hence try/finally rather than a bare statement at the end.
+            runtime.close();
+        }
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/FleetdListReportingSourcesAssemblyTest.java

@@ -0,0 +1,243 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.HerdrClient;
+import dev.ltms.fleet.msg.LeadChannelHandle;
+import dev.ltms.fleet.msg.LeadMessage;
+import dev.ltms.fleet.msg.ReplyInbox;
+import io.javalin.Javalin;
+import io.modelcontextprotocol.client.McpClient;
+import io.modelcontextprotocol.client.McpSyncClient;
+import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
+import io.modelcontextprotocol.spec.McpClientTransport;
+import io.modelcontextprotocol.spec.McpSchema;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.net.http.HttpRequest;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #612 Shape A ranks 9 and 11: drives the real {@code fleet_list} MCP route through a
+ * token-authenticated primary caller. The assertions read the response from the exact {@link
+ * dev.ltms.fleet.mcp.FleetMcp} instance assembled by {@link FleetdAssembly}, rather than a source
+ * scrape or a separately built reporting source.
+ *
+ * <p>The coordinator fixture contains a peer on purpose. Both a missing coordinator and an
+ * incorrectly wired peers argument can render as an empty list, so the non-empty peer assertion
+ * distinguishes the real call-site value from that inert result. Its fake mailbox never contacts a
+ * broker.
+ */
+class FleetdListReportingSourcesAssemblyTest {
+
+    private static final String TOKEN = "fleetd-r9-r11-test-token";
+    private static final String TOKEN_ENV = "FLEETD_R9_TEST_TOKEN";
+    private static final String PROFILE = "capacity-profile";
+    private static final String PEER = "peer-fleet";
+
+    private static final class FakeLeadChannel implements LeadChannelHandle {
+        @Override
+        public void publish(String toCoordId, LeadMessage message) {
+        }
+
+        @Override
+        public List<LeadMessage> peek() {
+            return List.of();
+        }
+
+        @Override
+        public void ack(String msgId) {
+        }
+
+        @Override
+        public String selfCoordId() {
+            return "test-lead";
+        }
+
+        @Override
+        public boolean heldDurable() {
+            return true;
+        }
+
+        @Override
+        public MailboxState inspect(String coordId) {
+            return MailboxState.unknown(coordId);
+        }
+
+        @Override
+        public void close() {
+        }
+    }
+
+    private static final class TestResourcePorts implements ResourcePorts {
+        private final FakeHerdr herdr = new FakeHerdr();
+        private Runnable shutdownHook;
+
+        @Override
+        public Map<String, String> environment() {
+            return Map.of(TOKEN_ENV, TOKEN);
+        }
+
+        @Override
+        public HerdrClient connectHerdr(Path socketPath) {
+            return herdr;
+        }
+
+        @Override
+        public Fleetd.AmqpOpener replyInboxOpener() {
+            return (uri, prefetch) -> new ReplyInbox() {
+                @Override public void own(String target) { }
+                @Override public void release(String target) { }
+                @Override public void publish(String target, String msgId, String content) { }
+                @Override public List<InboxMessage> peek(String target) { return List.of(); }
+                @Override public boolean ack(String target, String msgId) { return false; }
+            };
+        }
+
+        @Override
+        public Fleetd.LeadMailboxOpener leadMailboxOpener() {
+            return (uri, selfCoordId, prefetch) -> new FakeLeadChannel();
+        }
+
+        @Override
+        public LongSupplier nanoClock() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public LongSupplier wallClockNanos() {
+            return System::nanoTime;
+        }
+
+        @Override
+        public ScheduledExecutorService newScheduler(String purpose) {
+            return Executors.newSingleThreadScheduledExecutor();
+        }
+
+        @Override
+        public void addShutdownHook(Runnable hook) {
+            shutdownHook = hook;
+        }
+
+        @Override
+        public void startHttp(Javalin app, String host, int port) {
+            // The test binds runtime.app() itself, below.
+        }
+
+        @Override
+        public Runnable herdrPollWait() {
+            return () -> {
+                throw new UnsupportedOperationException("healthy FakeHerdr must not poll");
+            };
+        }
+    }
+
+    private FleetdRuntime runtime;
+    private TestResourcePorts ports;
+    private Javalin boundApp;
+
+    @AfterEach
+    void tearDown() {
+        if (boundApp != null) {
+            boundApp.stop();
+        }
+        if (ports != null && ports.shutdownHook != null) {
+            ports.shutdownHook.run();
+        }
+    }
+
+    private static FleetConfig writeConfig(Path dir) throws Exception {
+        Path file = dir.resolve("fleetd.yaml");
+        Files.writeString(file, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                auth:
+                  mode: token
+                  tokenEnv: %s
+                idleSleepGuard:
+                  enabled: false
+                health:
+                  enabled: true
+                  notifications:
+                    mode: webhook
+                profiles:
+                  %s:
+                    baseUrl: http://capacity.test:8000
+                    model: test-model
+                    maxLoad: 7
+                coordinator:
+                  uri: amqp://fake-coordinator/vh
+                  selfId: test-lead
+                  peers:
+                    - %s
+                """.formatted(TOKEN_ENV, PROFILE, PEER));
+        return FleetConfig.load(file);
+    }
+
+    private int assembleAndBind(Path dir) throws Exception {
+        FleetConfig cfg = writeConfig(dir);
+        ConfigRef config = new ConfigRef(dir.resolve("fleetd.yaml"), cfg);
+        ports = new TestResourcePorts();
+        runtime = FleetdAssembly.assembleAndStart(new AssemblyInputs(cfg, config,
+                new SubscriptionGuard(cfg.guard().hostSet())), ports);
+        boundApp = runtime.app().start("127.0.0.1", 0);
+        return boundApp.port();
+    }
+
+    private static String fleetList(int port) {
+        HttpRequest.Builder requestTemplate = HttpRequest.newBuilder()
+                .header("Authorization", "Bearer " + TOKEN);
+        McpClientTransport transport = HttpClientStreamableHttpTransport.builder("http://127.0.0.1:" + port)
+                .endpoint("/mcp")
+                .requestBuilder(requestTemplate)
+                .build();
+        try (McpSyncClient client = McpClient.sync(transport).build()) {
+            client.initialize();
+            McpSchema.CallToolResult response = client.callTool(McpSchema.CallToolRequest.builder("fleet_list")
+                    .arguments(Map.of()).build());
+            return ((McpSchema.TextContent) response.content().getFirst()).text();
+        }
+    }
+
+    @Test
+    @Timeout(value = 15, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    void fleetListReportsTheAssembledCapacitySource(@TempDir Path dir) throws Exception {
+        String response = fleetList(assembleAndBind(dir));
+
+        assertTrue(response.contains("\"capacity\""), response);
+        assertTrue(response.contains("\"" + PROFILE + "\""), response);
+        assertTrue(response.contains("\"maxLoad\":7"), response);
+    }
+
+    @Test
+    @Timeout(value = 15, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    void fleetListReportsTheAssembledHealthCoverageSource(@TempDir Path dir) throws Exception {
+        String response = fleetList(assembleAndBind(dir));
+
+        assertTrue(response.contains("\"healthCoverage\":\"full\""), response);
+    }
+
+    @Test
+    @Timeout(value = 15, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    void fleetListReportsTheAssembledCoordinatorPeers(@TempDir Path dir) throws Exception {
+        String response = fleetList(assembleAndBind(dir));
+
+        assertTrue(response.contains("\"coordinator\""), response);
+        assertTrue(response.contains("\"coordId\":\"" + PEER + "\""), response);
+    }
+}

fleetd/src/test/java/dev/ltms/fleet/FleetdQuarantineOutageDualWindowAssemblyTest.java

@@ -0,0 +1,368 @@
+package dev.ltms.fleet;
+
+import dev.ltms.fleet.config.ConfigRef;
+import dev.ltms.fleet.config.FleetConfig;
+import dev.ltms.fleet.guard.SubscriptionGuard;
+import dev.ltms.fleet.herdr.FakeHerdr;
+import dev.ltms.fleet.herdr.HerdrClient;
+import dev.ltms.fleet.inject.CompletionResolver;
+import dev.ltms.fleet.msg.Rendezvous;
+import dev.ltms.fleet.msg.TurnToken;
+import dev.ltms.fleet.session.MemberSession;
+import io.javalin.Javalin;
+import io.modelcontextprotocol.client.McpClient;
+import io.modelcontextprotocol.client.McpSyncClient;
+import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
+import io.modelcontextprotocol.spec.McpClientTransport;
+import io.modelcontextprotocol.spec.McpSchema;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * fleetd #612 Shape A, unit r4 — {@link FleetdAssembly}'s {@code quarantineSource} (lines 471-472)
+ * and {@code outageSource} (lines 473-476 at {@code main} = {@code 141ae3b}), EACH of which feeds
+ * two separate consumers: {@code FleetMcp} ({@code fleet_profiles}) and {@code FleetApp}
+ * ({@code GET /profiles}), one call site ({@code :484}/{@code :486}) into the MCP constructor and
+ * the SAME shared local again ({@code :529}) into the REST constructor.
+ *
+ * <p><strong>The property pinned here</strong> (from the ticket): when the real assembly has built
+ * a real {@link dev.ltms.fleet.placement.BackendQuarantine} holding a quarantined credential, and a
+ * real {@link dev.ltms.fleet.placement.BackendOutagePolicy} holding a cooling-off credential, BOTH
+ * operator windows must report that state — the real assembled {@code FleetMcp} (reached through
+ * {@link FleetdRuntime#mcp()}) and the real assembled REST surface (reached through {@link
+ * FleetdRuntime#app()}). A mutation that starves one consumer while leaving the other wired must
+ * make only that consumer's assertion go red.
+ *
+ * <p><strong>No source-text assertion anywhere in this file.</strong> Both windows are read off the
+ * REAL running objects: {@code fleet_profiles} is called through a real MCP client over a real
+ * HTTP connection to the servlet {@link FleetdAssembly} actually mounted, and {@code GET /profiles}
+ * is called through a real {@link java.net.http.HttpClient} against the real bound {@link
+ * FleetdRuntime#app()}. Neither is a copy built alongside the assembly for this test's benefit.
+ *
+ * <p><strong>How this gets past CB-185's own pid-resolution dead end.</strong> {@code
+ * FleetdAssemblyFleetAppTest}'s class javadoc explains that {@code GET /sessions} cannot be driven
+ * over real HTTP here because {@code LsofPeerPidLookup} excludes its own pid and an in-process test
+ * client/server share one JVM pid — every such request resolves {@code ANONYMOUS} and is refused
+ * before the handler runs. {@code GET /profiles} and {@code fleet_profiles} sit behind the exact
+ * same {@code Authz.Action.READ} gate. This test sidesteps the dead end instead of hitting it:
+ * {@code auth.mode: token} (see {@link dev.ltms.fleet.auth.CallerResolver#resolve}) resolves a
+ * caller to {@code PRIMARY} from a valid {@code Authorization: Bearer} header ALONE, with no pid
+ * resolution involved at all — the same technique {@code FleetMcpContextExtractorTest} already uses
+ * to drive a real {@code fleet_whoami} call through the real transport.
+ *
+ * <p><strong>How the quarantined/cooling-off state is set up.</strong> Both {@code BackendQuarantine}
+ * and {@code BackendOutagePolicy} are private to the collaborators the assembly wires them into, and
+ * (unlike {@code quarantineSource()}) neither {@code FleetMcp} nor {@code FleetApp} exposes a public
+ * accessor for the live {@code BackendOutagePolicy} instance. Rather than add one (acceptance
+ * criterion 1: no production change), this test drives the REAL production classification path —
+ * exactly the recipe {@code FleetdExhaustedPatternAssemblyTest} (quarantine) and {@code
+ * FleetdCompletionResolverAssemblyTest} (cool-off) already proved works end to end against this same
+ * {@link FleetdAssembly#assembleAndStart}: acquire a real {@link MemberSession}, feed the real {@link
+ * CompletionResolver} a pane scrape matching the profile's configured {@code exhaustedPattern} /
+ * {@code errorPattern}, and let the real {@code exhaustionSink}/{@code backendErrorSink} write into
+ * the real, shared tracker. Each setup step asserts its own {@link Rendezvous.Kind} as a CONTROL —
+ * if the resolver were never actually exercised, the setup itself fails loudly before either window
+ * is ever read.
+ */
+class FleetdQuarantineOutageDualWindowAssemblyTest {
+
+    private static final String TOKEN = "s3cret-r4-token";
+    private static final String TOKEN_ENV = "FLEETD_R4_TEST_TOKEN";
+
+    private static final class ControllableResourcePorts implements ResourcePorts {
+
+        final FakeHerdr herdr;
+        final AtomicLong nowNanos = new AtomicLong(1_000_000_000L); // arbitrary non-zero start
+        Runnable shutdownHook;
+
+        ControllableResourcePorts(FakeHerdr herdr) {
+            this.herdr = herdr;
+        }
+
+        void advanceSeconds(long seconds) {
+            nowNanos.addAndGet(TimeUnit.SECONDS.toNanos(seconds));
+        }
+
+        @Override
+        public Map<String, String> environment() {
+            return Map.of(TOKEN_ENV, TOKEN);
+        }
+
+        @Override
+        public HerdrClient connectHerdr(Path socketPath) {
+            return herdr;
+        }
+
+        @Override
+        public Fleetd.AmqpOpener replyInboxOpener() {
+            // Never invoked: this test's config has no `broker:` block.
+            return (uri, prefetch) -> {
+                throw new UnsupportedOperationException("replyInboxOpener must not be called — no broker: block");
+            };
+        }
+
+        @Override
+        public Fleetd.LeadMailboxOpener leadMailboxOpener() {
+            // Never invoked: this test's config has no `coordinator:` block.
+            return (uri, selfCoordId, prefetch) -> {
+                throw new UnsupportedOperationException(
+                        "leadMailboxOpener must not be called — no coordinator: block is configured");
+            };
+        }
+
+        @Override
+        public LongSupplier nanoClock() {
+            return nowNanos::get;
+        }
+
+        @Override
+        public LongSupplier wallClockNanos() {
+            return nowNanos::get;
+        }
+
+        @Override
+        public ScheduledExecutorService newScheduler(String purpose) {
+            return Executors.newSingleThreadScheduledExecutor();
+        }
+
+        @Override
+        public void addShutdownHook(Runnable hook) {
+            this.shutdownHook = hook;
+        }
+
+        @Override
+        public void startHttp(Javalin app, String host, int port) {
+            // Deliberately never bind here — this test binds runtime.app() itself, for real, below.
+        }
+
+        @Override
+        public Runnable herdrPollWait() {
+            // Never invoked: this test's FakeHerdr answers immediately, so awaitHerdr never polls.
+            return () -> {
+                throw new UnsupportedOperationException("herdrPollWait must not be called — herdr is healthy");
+            };
+        }
+    }
+
+    private FleetdRuntime runtime;
+    private ControllableResourcePorts ports;
+    private Javalin boundApp;
+
+    @AfterEach
+    void tearDown() {
+        if (boundApp != null) {
+            boundApp.stop();
+        }
+        if (ports != null && ports.shutdownHook != null) {
+            ports.shutdownHook.run();
+        }
+    }
+
+    private static FleetConfig writeConfig(Path dir, int cooldownSeconds) throws Exception {
+        Path f = dir.resolve("fleetd.yaml");
+        Files.writeString(f, """
+                bind:
+                  host: 127.0.0.1
+                  port: 8765
+                auth:
+                  mode: token
+                  tokenEnv: %s
+                idleSleepGuard:
+                  enabled: false
+                quarantineCooldownSeconds: %d
+                profiles:
+                  exhaustprofile:
+                    baseUrl: http://exhausthost.local:8000
+                    model: sonnet
+                    exhaustedPattern: "usage limit reached"
+                  coolprofile:
+                    baseUrl: http://coolhost.local:8000
+                    model: sonnet
+                    errorPattern: "credential outage"
+                guard:
+                  offSubscriptionHosts:
+                    - exhausthost.local
+                    - coolhost.local
+                """.formatted(TOKEN_ENV, cooldownSeconds));
+        return FleetConfig.load(f);
+    }
+
+    /** Assembles the real graph, then binds the real {@code Javalin app} to an ephemeral port. */
+    private int assembleAndBind(Path dir) throws Exception {
+        FleetConfig cfg = writeConfig(dir, 120);
+        ConfigRef config = new ConfigRef(dir.resolve("fleetd.yaml"), cfg);
+        SubscriptionGuard guard = new SubscriptionGuard(cfg.guard().hostSet());
+        ports = new ControllableResourcePorts(new FakeHerdr());
+        runtime = FleetdAssembly.assembleAndStart(new AssemblyInputs(cfg, config, guard), ports);
+        boundApp = runtime.app().start("127.0.0.1", 0);
+        return boundApp.port();
+    }
+
+    /**
+     * Drives the real assembled {@link CompletionResolver} through a scrape matching {@code
+     * exhaustprofile}'s configured {@code exhaustedPattern}, exactly {@code
+     * FleetdExhaustedPatternAssemblyTest}'s own recipe, so the real {@code exhaustionSink} quarantines
+     * the credential ({@code effectiveCredentialId() == "exhaustprofile"}, no explicit credentialId
+     * configured).
+     */
+    private void quarantineExhaustProfile(Path dir) {
+        MemberSession session = runtime.sessions().acquire("exhaustprofile", null, dir.toString(), null);
+        String target = session.terminalId();
+
+        CompletionResolver completion = runtime.completion();
+        CompletableFuture<Rendezvous.Resolution> waiter = new CompletableFuture<>();
+
+        ports.herdr.readText("idle, nothing yet");
+        completion.onDelivered(target, new TurnToken(target, waiter, null));
+        ports.herdr.readText("usage limit reached: try again in a few hours");
+        ports.advanceSeconds(3); // clear CompletionResolver.MIN_TURN_NANOS (2s), no real sleep
+        completion.resolveBeforePostAction(target);
+
+        Rendezvous.Resolution resolution = waiter.getNow(null);
+        assertTrue(resolution != null && resolution.kind() == Rendezvous.Kind.BACKEND_EXHAUSTED,
+                "CONTROL: setup must classify as BACKEND_EXHAUSTED before either window is read — "
+                        + "if this fails, the assembled resolver was never actually exercised: " + resolution);
+    }
+
+    /**
+     * Drives the real assembled {@link CompletionResolver} with TWO distinct targets on {@code
+     * coolprofile}, each matching its configured {@code errorPattern}, exactly {@code
+     * FleetdCompletionResolverAssemblyTest}'s own recipe, so the real {@code backendErrorSink} cools
+     * the credential off ({@code effectiveCredentialId() == "coolprofile"}).
+     */
+    private void coolOffCoolProfile(Path dir) {
+        MemberSession s1 = runtime.sessions().acquire("coolprofile", null, dir.toString(), null);
+        MemberSession s2 = runtime.sessions().acquire("coolprofile", null, dir.toString(), null);
+        CompletionResolver completion = runtime.completion();
+
+        String t1 = s1.terminalId();
+        CompletableFuture<Rendezvous.Resolution> w1 = new CompletableFuture<>();
+        ports.herdr.readText("idle 1");
+        completion.onDelivered(t1, new TurnToken(t1, w1, null));
+        ports.herdr.readText("credential outage: upstream 503");
+        ports.advanceSeconds(3);
+        completion.resolveBeforePostAction(t1);
+        Rendezvous.Resolution r1 = w1.getNow(null);
+        assertTrue(r1 != null && r1.kind() == Rendezvous.Kind.FAILED,
+                "CONTROL: target1's setup must classify FAILED (backend error): " + r1);
+
+        String t2 = s2.terminalId();
+        CompletableFuture<Rendezvous.Resolution> w2 = new CompletableFuture<>();
+        ports.herdr.readText("idle 2");
+        completion.onDelivered(t2, new TurnToken(t2, w2, null));
+        ports.herdr.readText("credential outage: upstream 503 again");
+        ports.advanceSeconds(3);
+        completion.resolveBeforePostAction(t2);
+        Rendezvous.Resolution r2 = w2.getNow(null);
+        assertTrue(r2 != null && r2.kind() == Rendezvous.Kind.FAILED,
+                "CONTROL: target2's setup must classify FAILED (backend error) — two distinct "
+                        + "targets are required to cross BackendOutagePolicy.THRESHOLD: " + r2);
+    }
+
+    /** Calls the real {@code fleet_profiles} tool over a real MCP client, token-authenticated as PRIMARY. */
+    private static McpSchema.CallToolResult callProfilesViaMcp(int port) {
+        HttpRequest.Builder requestTemplate = HttpRequest.newBuilder().header("Authorization", "Bearer " + TOKEN);
+        McpClientTransport transport = HttpClientStreamableHttpTransport.builder("http://127.0.0.1:" + port)
+                .endpoint("/mcp")
+                .requestBuilder(requestTemplate)
+                .build();
+        try (McpSyncClient client = McpClient.sync(transport).build()) {
+            client.initialize();
+            return client.callTool(McpSchema.CallToolRequest.builder("fleet_profiles").arguments(Map.of()).build());
+        }
+    }
+
+    private static String textOf(McpSchema.CallToolResult r) {
+        return ((McpSchema.TextContent) r.content().getFirst()).text();
+    }
+
+    /** Calls the real {@code GET /profiles} route over a real {@link HttpClient}, same token. */
+    private static String getProfilesViaRest(int port) throws Exception {
+        HttpClient http = HttpClient.newHttpClient();
+        HttpRequest req = HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + port + "/profiles"))
+                .header("Authorization", "Bearer " + TOKEN)
+                .GET().build();
+        HttpResponse<String> res = http.send(req, HttpResponse.BodyHandlers.ofString());
+        assertEquals(200, res.statusCode(), "GET /profiles must succeed with the real token: " + res.body());
+        return res.body();
+    }
+
+    // --- quarantineSource (FleetdAssembly.java :471-472) --------------------------------------
+
+    @Test
+    @Timeout(value = 15, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    void quarantinedCredentialIsReportedByTheRealAssembledFleetMcp(@TempDir Path dir) throws Exception {
+        int port = assembleAndBind(dir);
+        quarantineExhaustProfile(dir);
+
+        String out = textOf(callProfilesViaMcp(port));
+
+        assertTrue(out.contains("\"quarantined\""), "fleet_profiles must report a quarantined "
+                + "section once the real BackendQuarantine holds a quarantined credential: " + out);
+        assertTrue(out.contains("\"exhaustprofile\""), out);
+        assertTrue(out.contains("\"quarantinedForSeconds\""), out);
+    }
+
+    @Test
+    @Timeout(value = 15, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    void quarantinedCredentialIsReportedByTheRealAssembledFleetApp(@TempDir Path dir) throws Exception {
+        int port = assembleAndBind(dir);
+        quarantineExhaustProfile(dir);
+
+        String out = getProfilesViaRest(port);
+
+        assertTrue(out.contains("\"quarantined\""), "GET /profiles must report a quarantined "
+                + "section once the real BackendQuarantine holds a quarantined credential: " + out);
+        assertTrue(out.contains("\"exhaustprofile\""), out);
+        assertTrue(out.contains("\"quarantinedForSeconds\""), out);
+    }
+
+    // --- outageSource (FleetdAssembly.java :473-476) -------------------------------------------
+
+    @Test
+    @Timeout(value = 15, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    void coolingOffCredentialIsReportedByTheRealAssembledFleetMcp(@TempDir Path dir) throws Exception {
+        int port = assembleAndBind(dir);
+        coolOffCoolProfile(dir);
+
+        String out = textOf(callProfilesViaMcp(port));
+
+        assertTrue(out.contains("\"coolingOff\""), "fleet_profiles must report a coolingOff "
+                + "section once the real BackendOutagePolicy holds a cooling-off credential: " + out);
+        assertTrue(out.contains("\"coolprofile\""), out);
+        assertTrue(out.contains("\"coolingOffForSeconds\""), out);
+    }
+
+    @Test
+    @Timeout(value = 15, unit = TimeUnit.SECONDS, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
+    void coolingOffCredentialIsReportedByTheRealAssembledFleetApp(@TempDir Path dir) throws Exception {
+        int port = assembleAndBind(dir);
+        coolOffCoolProfile(dir);
+
+        String out = getProfilesViaRest(port);
+
+        assertTrue(out.contains("\"coolingOff\""), "GET /profiles must report a coolingOff "
+                + "section once the real BackendOutagePolicy holds a cooling-off credential: " + out);
+        assertTrue(out.contains("\"coolprofile\""), out);
+        assertTrue(out.contains("\"coolingOffForSeconds\""), out);
+    }
+}

scripts/config-edit.sh

@@ -2,6 +2,10 @@
 #
 # The one auditable way to edit the live fleetd.yaml.
 #
+# `--set` uses yq and rewrites the whole YAML document in yq's output style. Use `--from` for a
+# candidate whose comment alignment or other formatting carries meaning: it copies that file
+# verbatim while keeping this script's backup, parse check, atomic install, and verdict read-back.
+#
 # fleetd ticket #635 — why this exists at all: fleetd.yaml is gitignored and holds the live
 # fleet's settings. A bad raw edit reaches a daemon that is already serving, so a direct `Edit`
 # on it is refused by policy. This script is the allow-listed alternative, and it is not just
@@ -46,6 +50,11 @@
 #   scripts/config-edit.sh --dry-run --set <yq-path>=<value>
 #   scripts/config-edit.sh --restore
 #
+# `--set` rewrites the whole file in yq's output style, not only the requested keys. The script
+# warns before installation when the candidate changes more lines than its number of --set pairs.
+# Use `--from <candidate.yaml>` when comment alignment or other formatting is meaningful: --from
+# copies the candidate verbatim, with no yq round-trip.
+#
 # `--set .a.b=` (an empty value — a forgotten typo) is REFUSED, not accepted as "clear the
 # field": a null value falls back to its default rather than erroring, which is silent, not
 # safe. To clear a key on purpose, write a literal null: `--set .a.b=null`. Every other value
@@ -154,12 +163,10 @@ done
 # story: a YAML block scalar (`|`, `|-`, `>`, `>-`, ...) puts the VALUE on the lines that follow
 # the key, each indented deeper than it. The key-name match above only ever sees the key line
 # itself, so those continuation lines used to flow straight through unredacted while the key line
-# right above them printed a reassuring "<redacted>" — an incomplete redactor that looks complete
-# is worse than one that visibly does nothing, because it stops a reviewer from looking further.
-# The fix is structural, not another name to match: once a key line is masked, every following
-# line indented STRICTLY DEEPER than that key is masked too, by indentation alone, until the
-# indentation returns to the key's own level or shallower. This needs no knowledge of the key's
-# name and so protects a block scalar under any masked key, present or future.
+# right above them printed a reassuring "<redacted>". The redactor maps masked continuation lines
+# from each complete file before it reads the diff. It then masks a printed line when that file
+# line is inside a masked key's value. This covers block-scalar bodies even when the key line is
+# outside the printed hunk, and it keeps blank lines inside the value masked.
 #
 # `redact` is always fed `diff -u` output, and every line of a unified diff starts with exactly
 # one of ' ', '+', '-' (the three body markers; '@'/'-'/'+' for the three header-line kinds too).
@@ -168,37 +175,102 @@ done
 # column shallower than it really is, and either wrongly escapes a continuation mask or wrongly
 # ends one early. Tabs are out of scope: YAML forbids them for indentation, and this is a bounded
 # fix, not a YAML parser.
+map_masked_lines() {
+  local file="$1" side="$2" line content indent lead key line_number=0
+  local masked=0 masked_indent=0
+
+  case "$side" in
+    old) OLD_MASKED_LINES=() ;;
+    new) NEW_MASKED_LINES=() ;;
+    *) die "internal error: unknown redaction map side $side" ;;
+  esac
+
+  while IFS= read -r line || [ -n "$line" ]; do
+    line_number=$((line_number + 1))
+    content="$line"
+    indent=0
+    while [ "${content:$indent:1}" = " " ]; do indent=$((indent + 1)); done
+
+    if [ "$masked" = 1 ]; then
+      if [ -z "${content// /}" ] || [ "$indent" -gt "$masked_indent" ]; then
+        case "$side" in
+          old) OLD_MASKED_LINES[$line_number]=1 ;;
+          new) NEW_MASKED_LINES[$line_number]=1 ;;
+        esac
+        continue
+      fi
+      masked=0
+    fi
+
+    if [[ "$content" =~ ^([[:space:]]*)([A-Za-z0-9_.-]+:) ]]; then
+      lead="${BASH_REMATCH[1]}"
+      key="${BASH_REMATCH[2]}"
+      if [[ "$key" =~ (TOKEN|SECRET|PASSWORD|PASSWD|PASSPHRASE|CREDENTIAL|URI|_KEY) ]]; then
+        masked=1
+        masked_indent="$indent"
+      fi
+    fi
+  done < "$file"
+}
+
 redact() {
-  local line prefix content indent lead key
-  local masked=0 masked_indent=0 saved_nocasematch=0
+  local old_file="$1" new_file="$2"
+  local line prefix content indent lead key old_line=0 new_line=0 in_hunk=0
+  local old_masked new_masked saved_nocasematch=0
   shopt -q nocasematch && saved_nocasematch=1
   shopt -s nocasematch
-  sed -E 's#://[^@]*@#://<redacted>@#g' | while IFS= read -r line || [ -n "$line" ]; do
+
+  map_masked_lines "$old_file" old
+  map_masked_lines "$new_file" new
+
+  while IFS= read -r line || [ -n "$line" ]; do
+    if [[ "$line" =~ ^@@\ -([0-9]+)(,([0-9]+))?\ \+([0-9]+)(,([0-9]+))?\ @@ ]]; then
+      old_line="${BASH_REMATCH[1]}"
+      new_line="${BASH_REMATCH[4]}"
+      in_hunk=1
+      printf '%s\n' "$line"
+      continue
+    fi
+
     case "$line" in
       [\ +-]*) prefix="${line:0:1}"; content="${line:1}" ;;
       *)       prefix="";            content="$line" ;;
     esac
 
+    old_masked=0
+    new_masked=0
+    if [ "$in_hunk" = 1 ]; then
+      case "$prefix" in
+        ' ')
+          [ "${OLD_MASKED_LINES[$old_line]:-}" = 1 ] && old_masked=1
+          [ "${NEW_MASKED_LINES[$new_line]:-}" = 1 ] && new_masked=1
+          old_line=$((old_line + 1)); new_line=$((new_line + 1)) ;;
+        -)
+          [ "${OLD_MASKED_LINES[$old_line]:-}" = 1 ] && old_masked=1
+          old_line=$((old_line + 1)) ;;
+        +)
+          [ "${NEW_MASKED_LINES[$new_line]:-}" = 1 ] && new_masked=1
+          new_line=$((new_line + 1)) ;;
+      esac
+    fi
+
     indent=0
     while [ "${content:$indent:1}" = " " ]; do indent=$((indent + 1)); done
 
-    if [ "$masked" = 1 ] && [ "$indent" -gt "$masked_indent" ]; then
+    if [ "$old_masked" = 1 ] || [ "$new_masked" = 1 ]; then
       printf '%s%*s<redacted>\n' "$prefix" "$indent" ""
       continue
     fi
-    masked=0
 
     if [[ "$content" =~ ^([[:space:]]*)([A-Za-z0-9_.-]+:) ]]; then
       lead="${BASH_REMATCH[1]}"
       key="${BASH_REMATCH[2]}"
       if [[ "$key" =~ (TOKEN|SECRET|PASSWORD|PASSWD|PASSPHRASE|CREDENTIAL|URI|_KEY) ]]; then
         printf '%s%s%s <redacted>\n' "$prefix" "$lead" "$key"
-        masked=1
-        masked_indent="$indent"
         continue
       fi
     fi
-    printf '%s\n' "$line"
+    printf '%s\n' "$line" | sed -E 's#://[^@]*@#://<redacted>@#g'
   done
   [ "$saved_nocasematch" = 1 ] || shopt -u nocasematch
 }
@@ -457,6 +529,50 @@ parse_check() {
   yq eval '.' "$1" >/dev/null 2>&1
 }
 
+# Count logical changed lines in a unified diff. A replacement counts once, while an added or
+# deleted line also counts once. One changed `--set` value normally produces one changed line.
+changed_line_count() {
+  local before="$1" after="$2" line count=0 old_count=0 new_count=0
+  while IFS= read -r line || [ -n "$line" ]; do
+    case "$line" in
+      ---\ *|+++\ *|@@\ *)
+        if [ "$old_count" -gt "$new_count" ]; then
+          count=$((count + old_count))
+        else
+          count=$((count + new_count))
+        fi
+        old_count=0
+        new_count=0
+        ;;
+      -*) old_count=$((old_count + 1)) ;;
+      +*) new_count=$((new_count + 1)) ;;
+      *)
+        if [ "$old_count" -gt "$new_count" ]; then
+          count=$((count + old_count))
+        else
+          count=$((count + new_count))
+        fi
+        old_count=0
+        new_count=0
+        ;;
+    esac
+  done < <(diff -u "$before" "$after" || true)
+  if [ "$old_count" -gt "$new_count" ]; then
+    count=$((count + old_count))
+  else
+    count=$((count + new_count))
+  fi
+  printf '%s' "$count"
+}
+
+warn_set_reformat() {
+  local before="$1" after="$2" changed
+  changed="$(changed_line_count "$before" "$after")"
+  if [ "$changed" -gt "${#SETS[@]}" ]; then
+    warn "--set changed $changed candidate lines for ${#SETS[@]} pair(s); yq reformatted the whole file. Use --from for meaningful comment alignment or formatting."
+  fi
+}
+
 install_candidate() {
   local cand="$1" live="$2"
   mv -f "$cand" "$live"
@@ -618,10 +734,14 @@ run_edit() {
   fi
   ok "candidate parses"
 
+  if [ "$MODE" = "set" ]; then
+    warn_set_reformat "$backup" "$cand"
+  fi
+
   apply_mode "$cand" "$orig_mode"
 
   say "change (redacted)"
-  diff -u "$backup" "$cand" | redact || true
+  diff -u "$backup" "$cand" | redact "$backup" "$cand" || true
 
   say "install"
   install_candidate "$cand" "$CONFIG" \
@@ -649,8 +769,11 @@ dry_run_diff() {
     rm -f "$cand"; CAND=""
     die "candidate does not parse as valid YAML — this was a --dry-run, nothing would have been installed either"
   fi
+  if [ "$MODE" = "set" ]; then
+    warn_set_reformat "$CONFIG" "$cand"
+  fi
   say "dry run — diff (redacted), nothing installed"
-  diff -u "$CONFIG" "$cand" | redact || true
+  diff -u "$CONFIG" "$cand" | redact "$CONFIG" "$cand" || true
   rm -f "$cand"; CAND=""
   return 0
 }

scripts/test-config-edit.sh

@@ -474,6 +474,90 @@ test_passphrase_key_is_redacted() {
   assert_not_contains "FAKELEAK-PASSPHRASE" "$RUN_OUTPUT" "passphrase case: the passphrase VALUE must never leak"
 }
 
+# ------------------- acceptance criterion 19: the key line falls outside the printed hunk
+# fleetd #656 — criteria 15a/15b both put the edit right next to the key line, so the key line is
+# always inside diff -u's default 3-line context. Neither covers the actual case #639 fixed: an
+# 8-line block-scalar body with only its SIXTH line changed, so the printed hunk (3 lines of
+# context on each side of the change) covers body lines 3-8 and never includes the "token:" key
+# line at all. The old, line-by-line redact() only ever masks after it has SEEN the key line go
+# past; with the key line outside the hunk it never sets its mask, and the whole body — the
+# changed line included — passes through raw. The control key sits right after the body, inside
+# the same hunk, so the positive control below proves the fix is not simply printing nothing.
+new_fixture_hunk_without_key_line() {
+  local dir
+  dir="$(mktemp -d "$TMP/fixture.XXXXXX")"
+  cat > "$dir/fleetd.yaml" <<'YAML'
+bind:
+  host: 127.0.0.1
+  port: 19999
+broker:
+  uri: amqp://user:hunter2@host/vhost
+auth:
+  token: |
+    SECRET-LINE-1
+    SECRET-LINE-2
+    SECRET-LINE-3
+    SECRET-LINE-4
+    SECRET-LINE-5
+    SECRET-LINE-6
+    SECRET-LINE-7
+    SECRET-LINE-8
+  control: CTRL-MUST-APPEAR
+profiles:
+  sonnet:
+    weight: 3
+    maxLoad: 5
+YAML
+  : > "$dir/fleetd.out"
+  printf '%s' "$dir"
+}
+
+test_key_line_outside_hunk_is_still_redacted() {
+  local dir
+  dir="$(new_fixture_hunk_without_key_line)"
+  sed 's/SECRET-LINE-6$/SECRET-LINE-6-CHANGED/' "$dir/fleetd.yaml" > "$dir/candidate.yaml"
+
+  start_run "$dir" 5 --from "$dir/candidate.yaml"
+  sleep 1
+  printf 'config reloaded\n' >> "$dir/fleetd.out"
+  collect_run "$dir"
+
+  assert_equals 0 "$RUN_RC" "hunk-without-key-line case reload exit code"
+  # Positive control FIRST: without this, a diff that printed nothing at all would pass the
+  # negative assertion right below identically to a correctly redacted one.
+  assert_contains "CTRL-MUST-APPEAR" "$RUN_OUTPUT" "hunk-without-key-line case: the non-secret control line must still print unmasked"
+  assert_not_contains "SECRET-LINE-6-CHANGED" "$RUN_OUTPUT" "hunk-without-key-line case: the changed body line must never leak, even with the key line outside the printed hunk"
+}
+
+# ------------------------------- acceptance criterion 20: a blank line inside the value
+# fleetd #656 — the old, line-by-line redact() reset its mask on any line whose indentation was
+# not STRICTLY greater than the key's, and a wholly blank line has indentation 0, so it reset the
+# mask exactly like the "control:" line that legitimately ends the block scalar. Everything after
+# the blank line then printed raw. The current fix tracks masked lines by FILE line number instead
+# of by indentation seen so far, so a blank line inside the value stays masked.
+new_fixture_blank_line_in_value() {
+  local dir
+  dir="$(mktemp -d "$TMP/fixture.XXXXXX")"
+  printf 'bind:\n  host: 127.0.0.1\n  port: 19999\nbroker:\n  uri: amqp://user:hunter2@host/vhost\nauth:\n  token: |\n    LEAK-BEFORE-BLANK\n\n    LEAK-AFTER-BLANK\n  control: CTRL-MUST-APPEAR\nprofiles:\n  sonnet:\n    weight: 3\n    maxLoad: 5\n' > "$dir/fleetd.yaml"
+  : > "$dir/fleetd.out"
+  printf '%s' "$dir"
+}
+
+test_blank_line_inside_value_is_still_redacted() {
+  local dir
+  dir="$(new_fixture_blank_line_in_value)"
+  sed 's/LEAK-AFTER-BLANK$/LEAK-AFTER-BLANK-CHANGED/' "$dir/fleetd.yaml" > "$dir/candidate.yaml"
+
+  start_run "$dir" 5 --from "$dir/candidate.yaml"
+  sleep 1
+  printf 'config reloaded\n' >> "$dir/fleetd.out"
+  collect_run "$dir"
+
+  assert_equals 0 "$RUN_RC" "blank-line-in-value case reload exit code"
+  assert_contains "CTRL-MUST-APPEAR" "$RUN_OUTPUT" "blank-line-in-value case: the non-secret control line must still print unmasked"
+  assert_not_contains "LEAK-AFTER-BLANK-CHANGED" "$RUN_OUTPUT" "blank-line-in-value case: the line after the blank must never leak"
+}
+
 # ----------------------------------- acceptance criterion 16: a failing --set must not echo value
 # fleetd #635 follow-up (ticket comment 17673, defect 8) — apply_set_pairs used to echo the FULL
 # "$kv" (path=value, exactly as typed) in its yq-failure messages, so a broken --set with a
@@ -545,6 +629,57 @@ test_refusal_shape_from_parse_failure_wording_is_recognised() {
   assert_equals 4 "$RUN_RC" "the parse-failure refusal shape must also exit 4, not be read as silence"
 }
 
+# --set runs yq over the whole candidate. It warns when that changes more lines than the requested
+# pairs, but a simple file with only the intended changed line must stay quiet.
+new_fixture_reformat_sensitive() {
+  local dir
+  dir="$(mktemp -d "$TMP/fixture.XXXXXX")"
+  cat > "$dir/fleetd.yaml" <<'YAML'
+# A section comment that documents the next block.
+bind:
+  host: 127.0.0.1 # Keep this aligned with the port note.
+  port: 19999     # A fixture port.
+
+# These comments use their placement as documentation.
+profiles:
+  sonnet:
+    weight: 3
+    bootstrapText: >-
+      First line.
+      Second line.
+YAML
+  : > "$dir/fleetd.out"
+  printf '%s' "$dir"
+}
+
+test_set_warns_when_yq_reformats_extra_lines() {
+  local dir
+  dir="$(new_fixture_reformat_sensitive)"
+
+  start_run "$dir" 5 --set '.profiles.sonnet.weight=4'
+  sleep 1
+  printf 'config reloaded\n' >> "$dir/fleetd.out"
+  collect_run "$dir"
+
+  assert_equals 0 "$RUN_RC" "reformat warning case reload exit code"
+  assert_contains "yq reformatted the whole file" "$RUN_OUTPUT" \
+    "a --set that changes extra candidate lines must warn before installation"
+}
+
+test_set_stays_quiet_without_formatting_churn() {
+  local dir
+  dir="$(new_fixture)"
+
+  start_run "$dir" 5 --set '.profiles.sonnet.weight=4'
+  sleep 1
+  printf 'config reloaded\n' >> "$dir/fleetd.out"
+  collect_run "$dir"
+
+  assert_equals 0 "$RUN_RC" "no-reformat warning case reload exit code"
+  assert_not_contains "yq reformatted the whole file" "$RUN_OUTPUT" \
+    "a --set that changes only its requested candidate line must not warn"
+}
+
 echo "== acceptance criterion 1: refusal restores byte for byte =="
 test_refusal_restores_byte_for_byte
 echo "== acceptance criterion 2: clean reload keeps the edit =="
@@ -573,6 +708,10 @@ echo "== acceptance criterion 15a: a block scalar's continuation lines are redac
 test_block_scalar_continuation_is_redacted
 echo "== acceptance criterion 15b: a passphrase key is also recognised =="
 test_passphrase_key_is_redacted
+echo "== acceptance criterion 19: the key line falls outside the printed hunk =="
+test_key_line_outside_hunk_is_still_redacted
+echo "== acceptance criterion 20: a blank line inside the value =="
+test_blank_line_inside_value_is_still_redacted
 echo "== acceptance criterion 16: a failing --set must not echo its value =="
 test_failing_set_does_not_echo_its_value
 echo "== extra: dry-run never installs, and redacts =="
@@ -581,5 +720,9 @@ echo "== extra: --check is read-only and always exits 0 =="
 test_check_is_read_only_and_exits_zero
 echo "== extra: the parse-failure refusal shape is also recognised =="
 test_refusal_shape_from_parse_failure_wording_is_recognised
+echo "== acceptance criterion 17: --set warns about yq formatting churn =="
+test_set_warns_when_yq_reformats_extra_lines
+echo "== acceptance criterion 18: --set stays quiet without formatting churn =="
+test_set_stays_quiet_without_formatting_churn
 
 printf 'PASS: config-edit acceptance criteria\n'