fleet/fleetd
3
0
Fork 0
Code Issues 73 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fleetd #201/#227 unit 3: lead outage nudge

Browse Source 
Backend incidents become a fourth source inside ReplyPushLoop, not a new scheduler — a
second injector would race the one control that already owns lead-pane delivery. One
notice per incident per affected lead (not per member), one-shot, waiting while the lead
pane is not injectable, and combined into the same nudge as any pending failed ticket.

A classified target that cannot be mapped to a credential gets its own truthful notice
and its own pending/delivered records. It previously reused the incident message, which
told the lead a credential was 'cooling for 0 remaining seconds' when nothing was cooling,
and smuggled the free-text reason into the profiles field.

Verified by the lead: 1129 tests green; routing the unmapped path back through
onBackendIncident turns unmappedBackendTargetUsesTheKnownLeadSchedule red with 0 compile
errors, and that test now asserts the whole rendered message rather than two substrings.
This commit is contained in:
Dai Ha
2026-09-03 10:49:25 +07:00
parent 959c83534f 7840e9adf6
commit e5eb3534c7
2 changed files with 320 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fleetd/src/main/java/dev/ltms/fleet/msg/ReplyPushLoop.java
+194 -14
View File
@@ -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
+126
View File
@@ -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