CB-201: make unmapped target notice truthful

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

@@ -9,8 +9,10 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
@@ -68,6 +70,12 @@ public final class ReplyPushLoop {
     static final String QUESTIONS_NUDGE_FORMAT =
             "%d workers are paused on a question — run fleet_poll(ticket=...) for each, then answer "
                     + "with fleet_send(turnId=..., content=...): %s";
+    static final String BACKEND_INCIDENT_NUDGE_FORMAT =
+            "Backend credential %s is cooling for %d remaining seconds; affected profiles: %s; "
+                    + "affected workers: %s. Run fleet_list to see more.";
+    static final String BACKEND_TARGET_UNMAPPED_NUDGE_FORMAT =
+            "A backend error on worker %s could not be mapped to a credential (%s) — no cool-off "
+                    + "was applied. Run fleet_list to check that worker.";
 
     private final PrimaryRegistry primaryRegistry;
     private final AgentControl agents;
@@ -93,6 +101,13 @@ public final class ReplyPushLoop {
      * how depleted an older, still-open question's count is.
      */
     private final ConcurrentHashMap<String, PendingQuestion> pendingQuestions = new ConcurrentHashMap<>();
+    /** Backend incidents awaiting one successful delivery, keyed by incident id and owning lead. */
+    private final ConcurrentHashMap<IncidentLead, PendingIncident> pendingIncidents = new ConcurrentHashMap<>();
+    /** Incident/lead pairs already delivered. They make repeated incident reports one-shot. */
+    private final Set<IncidentLead> deliveredIncidents = ConcurrentHashMap.newKeySet();
+    private final ConcurrentHashMap<UnmappedTargetLead, PendingUnmappedTarget> pendingUnmappedTargets =
+            new ConcurrentHashMap<>();
+    private final Set<UnmappedTargetLead> deliveredUnmappedTargets = ConcurrentHashMap.newKeySet();
     /** CB-590: leads with an active combined reminder schedule (replies and/or tickets and/or questions). */
     private final ConcurrentHashMap<String, Boolean> activeLeads = new ConcurrentHashMap<>();
 
@@ -178,7 +193,20 @@ public final class ReplyPushLoop {
      * tracked per question, not per lead per source).
      */
     private record PendingQuestion(String turnId, String ticket, String target, String lead,
-                                   String question, int nudgeCount) {
+                                    String question, int nudgeCount) {
+    }
+
+    private record IncidentLead(String incidentId, String lead) {
+    }
+
+    private record PendingIncident(IncidentLead key, String credential, List<String> profiles,
+                                   List<String> targets, int remainingCoolOffSeconds, int nudgeCount) {
+    }
+
+    private record UnmappedTargetLead(String target, String reason, String lead) {
+    }
+
+    private record PendingUnmappedTarget(UnmappedTargetLead key, int nudgeCount) {
     }
 
     /** Questions still open for {@code lead}, snapshotted fresh for one tick. */
@@ -192,6 +220,24 @@ public final class ReplyPushLoop {
                 .collect(Collectors.toUnmodifiableSet());
     }
 
+    private List<PendingIncident> pendingIncidentsFor(String lead) {
+        return pendingIncidents.values().stream().filter(i -> lead.equals(i.key().lead())).toList();
+    }
+
+    private Set<IncidentLead> pendingIncidentKeysFor(String lead) {
+        return pendingIncidentsFor(lead).stream().map(PendingIncident::key)
+                .collect(Collectors.toUnmodifiableSet());
+    }
+
+    private List<PendingUnmappedTarget> pendingUnmappedTargetsFor(String lead) {
+        return pendingUnmappedTargets.values().stream().filter(i -> lead.equals(i.key().lead())).toList();
+    }
+
+    private Set<UnmappedTargetLead> pendingUnmappedTargetKeysFor(String lead) {
+        return pendingUnmappedTargetsFor(lead).stream().map(PendingUnmappedTarget::key)
+                .collect(Collectors.toUnmodifiableSet());
+    }
+
     /**
      * The reply-source reminder count {@link #decide} should see for {@code lead} on this tick:
      * the <em>minimum</em> nudge count among the reply targets currently pending for it (CB-598).
@@ -236,6 +282,15 @@ public final class ReplyPushLoop {
         return min == Integer.MAX_VALUE ? 0 : min;
     }
 
+    private int minIncidentNudgeCountFor(String lead) {
+        return pendingIncidentsFor(lead).stream().mapToInt(PendingIncident::nudgeCount).min().orElse(0);
+    }
+
+    private int minUnmappedTargetNudgeCountFor(String lead) {
+        return pendingUnmappedTargetsFor(lead).stream().mapToInt(PendingUnmappedTarget::nudgeCount)
+                .min().orElse(0);
+    }
+
     /**
      * Pure decision function: examine everything pending for {@code lead} — reply targets and
      * tickets alike — and return what the loop should do.
@@ -272,17 +327,35 @@ public final class ReplyPushLoop {
      * @param questionReminderCount the lowest nudge count among questions open for this lead
      */
     Action decide(String lead, int replyReminderCount, int ticketReminderCount, int questionReminderCount) {
+        return decide(lead, replyReminderCount, ticketReminderCount, questionReminderCount,
+                minIncidentNudgeCountFor(lead));
+    }
+
+    /** As above, with backend incidents as a fourth, independently bounded source. */
+    Action decide(String lead, int replyReminderCount, int ticketReminderCount, int questionReminderCount,
+                  int incidentReminderCount) {
+        return decide(lead, replyReminderCount, ticketReminderCount, questionReminderCount,
+                incidentReminderCount, minUnmappedTargetNudgeCountFor(lead));
+    }
+
+    /** As above, with unmapped backend targets as a fifth, independently bounded source. */
+    Action decide(String lead, int replyReminderCount, int ticketReminderCount, int questionReminderCount,
+                  int incidentReminderCount, int unmappedTargetReminderCount) {
         boolean hasReplyWork = !pendingReplyTargetsFor(lead).isEmpty();
         boolean hasTicketWork = !pendingTicketIdsFor(lead).isEmpty();
         boolean hasQuestionWork = !pendingQuestionTurnIdsFor(lead).isEmpty();
-        if (!hasReplyWork && !hasTicketWork && !hasQuestionWork) {
+        boolean hasIncidentWork = !pendingIncidentKeysFor(lead).isEmpty();
+        boolean hasUnmappedTargetWork = !pendingUnmappedTargetKeysFor(lead).isEmpty();
+        if (!hasReplyWork && !hasTicketWork && !hasQuestionWork && !hasIncidentWork && !hasUnmappedTargetWork) {
             log.debug("push: nothing pending for lead {}, stopping reminder", lead);
             return Action.STOP;
         }
         boolean replyEligible = hasReplyWork && replyReminderCount < maxReminders;
         boolean ticketEligible = hasTicketWork && ticketReminderCount < maxReminders;
         boolean questionEligible = hasQuestionWork && questionReminderCount < maxReminders;
-        if (!replyEligible && !ticketEligible && !questionEligible) {
+        boolean incidentEligible = hasIncidentWork && incidentReminderCount < maxReminders;
+        boolean unmappedTargetEligible = hasUnmappedTargetWork && unmappedTargetReminderCount < maxReminders;
+        if (!replyEligible && !ticketEligible && !questionEligible && !incidentEligible && !unmappedTargetEligible) {
             log.debug("push: reminder cap ({}) reached for lead {} on every source with pending work, stopping",
                     maxReminders, lead);
             countNudge("exhausted");
@@ -394,6 +467,48 @@ public final class ReplyPushLoop {
         pendingQuestions.remove(turnId);
     }
 
+    /**
+     * Queue a one-shot backend credential outage notice for every distinct lead that owns an
+     * affected worker. A successful injection records its {@code (incidentId, lead)} key, so a
+     * repeat report never reminds that lead again.
+     */
+    public void onBackendIncident(String incidentId, Collection<String> targets, String credential,
+                                  Collection<String> profiles, int remainingCoolOffSeconds) {
+        Map<String, List<String>> targetsByLead = new ConcurrentHashMap<>();
+        for (String target : targets) {
+            var lead = primaryRegistry.nudgeTargetFor(target);
+            if (lead.isEmpty()) {
+                log.warn("push: backend incident {} has no known lead for target {}", incidentId, target);
+                continue;
+            }
+            targetsByLead.computeIfAbsent(lead.get(), _ -> new ArrayList<>()).add(target);
+        }
+        List<String> profileNames = profiles.stream().sorted().toList();
+        for (var entry : targetsByLead.entrySet()) {
+            IncidentLead key = new IncidentLead(incidentId, entry.getKey());
+            if (deliveredIncidents.contains(key)) continue;
+            pendingIncidents.putIfAbsent(key, new PendingIncident(key, credential, profileNames,
+                    entry.getValue().stream().sorted().toList(), remainingCoolOffSeconds, 0));
+            startOrCoalesce(entry.getKey());
+        }
+    }
+
+    /**
+     * Tell the owning lead that a classified backend error could not be tied to a credential.
+     * Without an owning lead, emit a warning because no control can act on the target.
+     */
+    public void onBackendTargetUnmapped(String target, String reason) {
+        var lead = primaryRegistry.nudgeTargetFor(target);
+        if (lead.isEmpty()) {
+            log.warn("push: backend target {} could not map to a credential: {}", target, reason);
+            return;
+        }
+        UnmappedTargetLead key = new UnmappedTargetLead(target, reason, lead.get());
+        if (deliveredUnmappedTargets.contains(key)) return;
+        pendingUnmappedTargets.putIfAbsent(key, new PendingUnmappedTarget(key, 0));
+        startOrCoalesce(lead.get());
+    }
+
     // --- the schedule ----------------------------------------------------------------------------
 
     /** Start a reminder schedule for {@code lead}, or join the one already running. */
@@ -425,18 +540,25 @@ public final class ReplyPushLoop {
         Set<String> repliesBefore = pendingReplyTargetsFor(lead);
         Set<String> ticketsBefore = pendingTicketIdsFor(lead);
         Set<String> questionsBefore = pendingQuestionTurnIdsFor(lead);
+        Set<IncidentLead> incidentsBefore = pendingIncidentKeysFor(lead);
+        Set<UnmappedTargetLead> unmappedTargetsBefore = pendingUnmappedTargetKeysFor(lead);
         int replyReminderCount = minReplyNudgeCountFor(lead);
         int ticketReminderCount = minTicketNudgeCountFor(lead);
         int questionReminderCount = minQuestionNudgeCountFor(lead);
-        var action = decide(lead, replyReminderCount, ticketReminderCount, questionReminderCount);
+        int incidentReminderCount = minIncidentNudgeCountFor(lead);
+        int unmappedTargetReminderCount = minUnmappedTargetNudgeCountFor(lead);
+        var action = decide(lead, replyReminderCount, ticketReminderCount, questionReminderCount,
+                incidentReminderCount, unmappedTargetReminderCount);
         switch (action) {
             case INJECT -> {
-                injectNudge(lead, replyReminderCount, ticketReminderCount, questionReminderCount);
+                injectNudge(lead, replyReminderCount, ticketReminderCount, questionReminderCount,
+                        incidentReminderCount, unmappedTargetReminderCount);
                 scheduleNext(lead);
             }
             // Re-check after the configured backoff; the lead may become injectable soon.
             case WAIT_BUSY -> scheduleNext(lead);
-            case STOP -> stopOrRestart(lead, repliesBefore, ticketsBefore, questionsBefore);
+            case STOP -> stopOrRestart(lead, repliesBefore, ticketsBefore, questionsBefore, incidentsBefore,
+                    unmappedTargetsBefore);
         }
     }
 
@@ -480,11 +602,25 @@ public final class ReplyPushLoop {
      * decision-to-release window reclaims the schedule slot exactly like a raced-in reply or ticket.
      */
     void stopOrRestart(String lead, Set<String> repliesBefore, Set<String> ticketsBefore,
-                       Set<String> questionsBefore) {
+                        Set<String> questionsBefore) {
+        stopOrRestart(lead, repliesBefore, ticketsBefore, questionsBefore, pendingIncidentKeysFor(lead));
+    }
+
+    private void stopOrRestart(String lead, Set<String> repliesBefore, Set<String> ticketsBefore,
+                               Set<String> questionsBefore, Set<IncidentLead> incidentsBefore) {
+        stopOrRestart(lead, repliesBefore, ticketsBefore, questionsBefore, incidentsBefore,
+                pendingUnmappedTargetKeysFor(lead));
+    }
+
+    private void stopOrRestart(String lead, Set<String> repliesBefore, Set<String> ticketsBefore,
+                               Set<String> questionsBefore, Set<IncidentLead> incidentsBefore,
+                               Set<UnmappedTargetLead> unmappedTargetsBefore) {
         activeLeads.remove(lead);
         boolean racedIn = pendingReplyTargetsFor(lead).stream().anyMatch(t -> !repliesBefore.contains(t))
                 || pendingTicketIdsFor(lead).stream().anyMatch(t -> !ticketsBefore.contains(t))
-                || pendingQuestionTurnIdsFor(lead).stream().anyMatch(t -> !questionsBefore.contains(t));
+                || pendingQuestionTurnIdsFor(lead).stream().anyMatch(t -> !questionsBefore.contains(t))
+                || pendingIncidentKeysFor(lead).stream().anyMatch(i -> !incidentsBefore.contains(i))
+                || pendingUnmappedTargetKeysFor(lead).stream().anyMatch(i -> !unmappedTargetsBefore.contains(i));
         if (racedIn && activeLeads.putIfAbsent(lead, Boolean.TRUE) == null) {
             log.debug("push: new work for lead {} raced the reminder loop's stop — restarting", lead);
             scheduleNext(lead);
@@ -495,18 +631,22 @@ public final class ReplyPushLoop {
 
     /** Send one combined nudge covering everything currently pending for {@code lead}. */
     private void injectNudge(String lead, int replyReminderCount, int ticketReminderCount,
-                             int questionReminderCount) {
+                              int questionReminderCount, int incidentReminderCount,
+                              int unmappedTargetReminderCount) {
         // Re-read rather than threading it down from decide(): a reply can drain, a ticket be
         // collected, or a question be answered (or another arrive), between the decision and the
         // injection.
         Set<String> replyTargets = pendingReplyTargetsFor(lead);
         List<PendingTicket> tickets = pendingTicketsFor(lead);
         List<PendingQuestion> questions = pendingQuestionsFor(lead);
-        if (replyTargets.isEmpty() && tickets.isEmpty() && questions.isEmpty()) {
+        List<PendingIncident> incidents = pendingIncidentsFor(lead);
+        List<PendingUnmappedTarget> unmappedTargets = pendingUnmappedTargetsFor(lead);
+        if (replyTargets.isEmpty() && tickets.isEmpty() && questions.isEmpty() && incidents.isEmpty()
+                && unmappedTargets.isEmpty()) {
             log.debug("push: pending work for lead {} drained before the nudge could be sent", lead);
             return;
         }
-        String nudge = formatNudge(replyTargets, tickets, questions);
+        String nudge = formatNudge(replyTargets, tickets, questions, incidents, unmappedTargets);
         try {
             agents.send(lead, nudge);
             log.debug("push: nudge sent to lead {} (reply {}/{}, ticket {}/{}, question {}/{}; "
@@ -515,6 +655,16 @@ public final class ReplyPushLoop {
                     questionReminderCount + 1, maxReminders,
                     replyTargets.size(), tickets.size(), questions.size());
             countNudge("delivered");
+            for (PendingIncident incident : incidents) {
+                if (pendingIncidents.remove(incident.key(), incident)) {
+                    deliveredIncidents.add(incident.key());
+                }
+            }
+            for (PendingUnmappedTarget unmappedTarget : unmappedTargets) {
+                if (pendingUnmappedTargets.remove(unmappedTarget.key(), unmappedTarget)) {
+                    deliveredUnmappedTargets.add(unmappedTarget.key());
+                }
+            }
         } catch (RuntimeException e) {
             log.warn("push: failed to nudge lead {} (reply {}/{}, ticket {}/{}, question {}/{}): {}",
                     lead, replyReminderCount + 1, maxReminders, ticketReminderCount + 1, maxReminders,
@@ -526,12 +676,13 @@ public final class ReplyPushLoop {
         // item already at or over the cap keeps riding along in the text (still pending, still
         // named) but its extra bumps here are inert: decide() already treats it as ineligible once
         // its count reaches maxReminders.
-        bumpNudgeCounts(replyTargets, tickets, questions);
+        bumpNudgeCounts(replyTargets, tickets, questions, incidents, unmappedTargets);
     }
 
     /** Record that every one of these items was just named in a sent (or attempted) nudge. */
     private void bumpNudgeCounts(Set<String> replyTargets, List<PendingTicket> tickets,
-                                 List<PendingQuestion> questions) {
+                                  List<PendingQuestion> questions, List<PendingIncident> incidents,
+                                  List<PendingUnmappedTarget> unmappedTargets) {
         for (String target : replyTargets) {
             pendingReplies.computeIfPresent(target, (t, e) -> new ReplyEntry(e.lead(), e.nudgeCount() + 1));
         }
@@ -544,6 +695,14 @@ public final class ReplyPushLoop {
                     new PendingQuestion(e.turnId(), e.ticket(), e.target(), e.lead(), e.question(),
                             e.nudgeCount() + 1));
         }
+        for (PendingIncident incident : incidents) {
+            pendingIncidents.computeIfPresent(incident.key(), (id, e) -> new PendingIncident(e.key(),
+                    e.credential(), e.profiles(), e.targets(), e.remainingCoolOffSeconds(), e.nudgeCount() + 1));
+        }
+        for (PendingUnmappedTarget unmappedTarget : unmappedTargets) {
+            pendingUnmappedTargets.computeIfPresent(unmappedTarget.key(), (id, e) ->
+                    new PendingUnmappedTarget(e.key(), e.nudgeCount() + 1));
+        }
     }
 
     /** Schedule the next tick on the scheduler thread pool. */
@@ -556,7 +715,8 @@ public final class ReplyPushLoop {
 
     /** Render everything pending for one lead as a single nudge line. */
     private static String formatNudge(Set<String> replyTargets, List<PendingTicket> tickets,
-                                      List<PendingQuestion> questions) {
+                                       List<PendingQuestion> questions, List<PendingIncident> incidents,
+                                       List<PendingUnmappedTarget> unmappedTargets) {
         List<String> parts = new ArrayList<>();
         if (!replyTargets.isEmpty()) {
             parts.add(formatRepliesNudge(replyTargets));
@@ -567,6 +727,12 @@ public final class ReplyPushLoop {
         if (!questions.isEmpty()) {
             parts.add(formatQuestionsNudge(questions));
         }
+        if (!incidents.isEmpty()) {
+            parts.addAll(incidents.stream().map(ReplyPushLoop::formatBackendIncidentNudge).toList());
+        }
+        if (!unmappedTargets.isEmpty()) {
+            parts.addAll(unmappedTargets.stream().map(ReplyPushLoop::formatUnmappedTargetNudge).toList());
+        }
         return String.join(" | ", parts);
     }
 
@@ -606,6 +772,16 @@ public final class ReplyPushLoop {
         return QUESTIONS_NUDGE_FORMAT.formatted(pending.size(), ids);
     }
 
+    private static String formatBackendIncidentNudge(PendingIncident incident) {
+        return BACKEND_INCIDENT_NUDGE_FORMAT.formatted(incident.credential(), incident.remainingCoolOffSeconds(),
+                String.join(", ", incident.profiles()), String.join(", ", incident.targets()));
+    }
+
+    private static String formatUnmappedTargetNudge(PendingUnmappedTarget unmappedTarget) {
+        return BACKEND_TARGET_UNMAPPED_NUDGE_FORMAT.formatted(unmappedTarget.key().target(),
+                unmappedTarget.key().reason());
+    }
+
     // --- lifecycle -----------------------------------------------------------------------------
 
     /**
@@ -627,6 +803,10 @@ public final class ReplyPushLoop {
         pendingReplies.clear();
         pendingTickets.clear();
         pendingQuestions.clear();
+        pendingIncidents.clear();
+        deliveredIncidents.clear();
+        pendingUnmappedTargets.clear();
+        deliveredUnmappedTargets.clear();
     }
 
     /** @see #stop() */

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

@@ -43,6 +43,7 @@ class ReplyPushLoopTest {
     private static final String PRIMARY = "term_primary";
     private static final String WORKER = "term_worker";
     private static final String WORKER2 = "term_worker2";
+    private static final String OTHER_PRIMARY = "term_other_primary";
     private static final ObjectMapper MAPPER = new ObjectMapper();
 
     private PrimaryRegistry registry;
@@ -529,6 +530,106 @@ class ReplyPushLoopTest {
         assertTrue(nudge.contains("task-1"), "the ticket must not be dropped: " + nudge);
     }
 
+    // --- CB-201: backend credential outage nudges -----------------------------------------------
+
+    @Test
+    void oneBackendIncidentForTwoWorkersOfOneLeadSendsOnceAndIsOneShot() throws Exception {
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        registry.recordDelegation(WORKER, PRIMARY);
+        registry.recordDelegation(WORKER2, PRIMARY);
+        var loop = loop(2, 100);
+
+        loop.onBackendIncident("incident-1", List.of(WORKER, WORKER2), "cred-a",
+                List.of("terra", "codex"), 47);
+
+        assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS), "one lead gets one outage injection");
+        Thread.sleep(200);
+        assertEquals(1, rec.sendCount(), "two workers for one lead must send only one outage notice");
+        String nudge = rec.sentParams().getFirst().getValue().toString();
+        assertTrue(nudge.contains("cred-a"));
+        assertTrue(nudge.contains("codex, terra"));
+        assertTrue(nudge.contains(WORKER) && nudge.contains(WORKER2));
+        assertTrue(nudge.contains("47 remaining seconds"));
+        assertTrue(nudge.contains("fleet_list"));
+
+        loop.onBackendIncident("incident-1", List.of(WORKER, WORKER2), "cred-a",
+                List.of("terra", "codex"), 47);
+        Thread.sleep(250);
+        assertEquals(1, rec.sendCount(), "a delivered incident id must never be sent again");
+    }
+
+    @Test
+    void oneBackendIncidentSendsOnceToEachAffectedLead() throws Exception {
+        var rec = recordingClient();
+        rec.sendLatch = new CountDownLatch(2);
+        agents = new AgentControl(rec);
+        registry.recordDelegation(WORKER, PRIMARY);
+        registry.recordDelegation(WORKER2, OTHER_PRIMARY);
+        var loop = loop(1, 50);
+
+        loop.onBackendIncident("incident-2", List.of(WORKER, WORKER2), "cred-a", List.of("terra"), 60);
+
+        assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS), "each affected lead gets its one notice");
+        assertEquals(2, rec.sendCount());
+    }
+
+    @Test
+    void backendIncidentWaitsForBusyLeadThenCombinesWithFailedTicket() throws Exception {
+        var rec = new BusyThenIdleHerdrClient(2);
+        agents = new AgentControl(rec);
+        var loop = loop(1, 50);
+
+        loop.onTicketTerminal("task-failed", WORKER, true);
+        loop.onBackendIncident("incident-3", List.of(WORKER), "cred-b", List.of("terra"), 31);
+
+        assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS), "busy lead should receive deferred combined nudge");
+        assertEquals(1, rec.sendCount(), "ticket and outage must use the same injection");
+        String nudge = rec.sentParams().getFirst().getValue().toString();
+        assertTrue(nudge.contains("task-failed") && nudge.contains("cred-b"),
+                "the one nudge must name both the failed ticket and outage: " + nudge);
+    }
+
+    @Test
+    void failedBackendIncidentSendRetriesWithinTheExistingReminderCap() throws Exception {
+        var rec = new ThrowOnceHerdrClient();
+        agents = new AgentControl(rec);
+        var loop = loop(2, 50);
+
+        loop.onBackendIncident("incident-4", List.of(WORKER), "cred-c", List.of("terra"), 20);
+
+        assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS), "failed first send must retry under maxReminders");
+        assertEquals(2, rec.promptAttempts.get(), "one failed send and one retry use the existing cap");
+    }
+
+    @Test
+    void unmappedBackendTargetUsesTheKnownLeadSchedule() throws Exception {
+        var rec = recordingClient();
+        agents = new AgentControl(rec);
+        var loop = loop(1, 50);
+
+        loop.onBackendTargetUnmapped(WORKER, "no credential matched");
+
+        assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS));
+        String nudge = ((Map<?, ?>) rec.sentParams().getFirst().getValue()).get("text").toString();
+        assertEquals("A backend error on worker term_worker could not be mapped to a credential "
+                        + "(no credential matched) — no cool-off was applied. "
+                        + "Run fleet_list to check that worker.",
+                nudge);
+    }
+
+    @Test
+    void stopClearsPendingBackendIncidents() {
+        agents = agentWithStatus("idle");
+        var loop = loop(1, 100_000);
+        loop.onBackendIncident("incident-stop", List.of(WORKER), "cred-stop", List.of("terra"), 60);
+
+        loop.stop();
+
+        assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0, 0, 0),
+                "stop must clear a pending backend incident as well as older sources");
+    }
+
     // --- CB-590 follow-up: per-source reminder budgets — the regression this round exists for ---
 
     @Test
@@ -928,6 +1029,31 @@ class ReplyPushLoopTest {
         return new RecordingHerdrClient();
     }
 
+    /** Fails its first prompt call, then records the retry. */
+    private static final class ThrowOnceHerdrClient implements HerdrClient {
+        private final AtomicInteger promptAttempts = new AtomicInteger();
+        private final CountDownLatch sendLatch = new CountDownLatch(1);
+
+        @Override
+        public JsonNode call(String method, Object params) {
+            if ("agent.get".equals(method)) {
+                return MAPPER.createObjectNode().set("agent", MAPPER.createObjectNode()
+                        .put("terminal_id", PRIMARY).put("agent_status", "idle"));
+            }
+            if ("agent.prompt".equals(method) && promptAttempts.getAndIncrement() == 0) {
+                throw new IllegalStateException("first prompt fails");
+            }
+            if ("agent.prompt".equals(method)) {
+                sendLatch.countDown();
+            }
+            return MAPPER.createObjectNode();
+        }
+
+        @Override
+        public void close() {
+        }
+    }
+
     /**
      * Thread-safe fake that reports {@code working} (not injectable) for its first
      * {@code busyChecks} status calls, then {@code idle} forever after — used to prove work queued