Re-arm watchdog continue decisions

This commit is contained in:
Dotta committed 2026-04-24 13:30:47 -05:00
1 parent fdb2bcec39
commit 65d410f405
4 files changed
+78 -11

No files matched your search

@@ -15,6 +15,7 @@ import {
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
import {
ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS,
ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS,
ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS,
heartbeatService,
@@ -352,6 +353,58 @@ describeEmbeddedPostgres("active-run output watchdog", () => {
});
});
it("re-arms continue decisions after the default quiet window", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId } = await seedRunningRun({
now,
ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000,
});
const heartbeat = heartbeatService(db);
const recovery = recoveryService(db, { enqueueWakeup: vi.fn() });
const scan = await heartbeat.scanSilentActiveRuns({ now, companyId });
const evaluationIssueId = scan.evaluationIssueIds[0];
expect(evaluationIssueId).toBeTruthy();
const decision = await recovery.recordWatchdogDecision({
runId,
actor: { type: "agent", agentId: managerId },
decision: "continue",
evaluationIssueId,
reason: "Current evidence is acceptable; keep watching.",
now,
});
const rearmAt = new Date(now.getTime() + ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS);
expect(decision).toMatchObject({
runId,
evaluationIssueId,
decision: "continue",
createdByAgentId: managerId,
});
expect(decision.snoozedUntil?.toISOString()).toBe(rearmAt.toISOString());
await db.update(issues).set({ status: "done" }).where(eq(issues.id, evaluationIssueId));
const beforeRearm = await heartbeat.scanSilentActiveRuns({
now: new Date(rearmAt.getTime() - 60_000),
companyId,
});
expect(beforeRearm).toMatchObject({ created: 0, snoozed: 1 });
const afterRearm = await heartbeat.scanSilentActiveRuns({
now: new Date(rearmAt.getTime() + 60_000),
companyId,
});
expect(afterRearm.created).toBe(1);
expect(afterRearm.evaluationIssueIds[0]).not.toBe(evaluationIssueId);
const evaluations = await db
.select()
.from(issues)
.where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stale_active_run_evaluation")));
expect(evaluations.filter((issue) => !["done", "cancelled"].includes(issue.status))).toHaveLength(1);
});
it("rejects agent watchdog decisions using issues not bound to the target run", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, coderId, runId, issuePrefix } = await seedRunningRun({
+1
View File
@@ -155,6 +155,7 @@ const CANCELLABLE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retr
const HEARTBEAT_RUN_TERMINAL_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"] as const;
const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const;
export {
ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS,
ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS,
ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS,
} from "./recovery/service.js";
+20 -9
View File
@@ -36,6 +36,7 @@ const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "ti
const ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_MIN_STALE_MS = 24 * 60 * 60 * 1000;
export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000;
export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000;
export const ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS = 30 * 60 * 1000;
const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024;
const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.strandedIssueRecovery;
const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation;
@@ -509,7 +510,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
return startedAt ? Math.max(0, now.getTime() - startedAt.getTime()) : null;
}
async function latestActiveOutputSnooze(companyId: string, runId: string, now = new Date()) {
async function latestActiveOutputQuietUntilDecision(companyId: string, runId: string, now = new Date()) {
const [row] = await db
.select()
.from(heartbeatRunWatchdogDecisions)
@@ -517,7 +518,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
and(
eq(heartbeatRunWatchdogDecisions.companyId, companyId),
eq(heartbeatRunWatchdogDecisions.runId, runId),
eq(heartbeatRunWatchdogDecisions.decision, "snooze"),
inArray(heartbeatRunWatchdogDecisions.decision, ["snooze", "continue"]),
gt(heartbeatRunWatchdogDecisions.snoozedUntil, now),
),
)
@@ -557,15 +558,15 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
>,
now = new Date(),
): Promise<RunOutputSilenceSummary> {
const [snooze, evaluation] = await Promise.all([
latestActiveOutputSnooze(run.companyId, run.id, now),
const [quietUntilDecision, evaluation] = await Promise.all([
latestActiveOutputQuietUntilDecision(run.companyId, run.id, now),
findOpenStaleRunEvaluation(run.companyId, run.id),
]);
const silenceStartedAt = silenceStartedAtForRun(run);
const silenceAgeMs = run.status === "running" ? silenceAgeMsForRun(run, now) : null;
const level = run.status !== "running"
? "not_applicable"
: snooze
: quietUntilDecision
? "snoozed"
: (silenceAgeMs ?? 0) >= ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS
? "critical"
@@ -583,7 +584,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
level,
suspicionThresholdMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS,
criticalThresholdMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS,
snoozedUntil: snooze?.snoozedUntil ?? null,
snoozedUntil: quietUntilDecision?.snoozedUntil ?? null,
evaluationIssueId: evaluation?.id ?? null,
evaluationIssueIdentifier: evaluation?.identifier ?? null,
};
@@ -993,7 +994,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
};
for (const run of candidates) {
if (await latestActiveOutputSnooze(run.companyId, run.id, now)) {
if (await latestActiveOutputQuietUntilDecision(run.companyId, run.id, now)) {
result.snoozed += 1;
continue;
}
@@ -1018,6 +1019,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
reason?: string | null;
snoozedUntil?: Date | null;
createdByRunId?: string | null;
now?: Date;
}) {
const [run] = await db
.select()
@@ -1095,6 +1097,15 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
}
}
const decisionNow = input.now ?? new Date();
const effectiveSnoozedUntil = input.decision === "snooze"
? input.snoozedUntil ?? null
: input.decision === "continue"
? input.snoozedUntil && input.snoozedUntil > decisionNow
? input.snoozedUntil
: new Date(decisionNow.getTime() + ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS)
: null;
const [row] = await db
.insert(heartbeatRunWatchdogDecisions)
.values({
@@ -1102,7 +1113,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
runId: run.id,
evaluationIssueId: input.evaluationIssueId ?? null,
decision: input.decision,
snoozedUntil: input.snoozedUntil ?? null,
snoozedUntil: effectiveSnoozedUntil,
reason: input.reason ?? null,
createdByAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null,
createdByUserId: input.actor.type === "board" ? input.actor.userId ?? null : null,
@@ -1127,7 +1138,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
source: "recovery.record_watchdog_decision",
decision: input.decision,
evaluationIssueId: input.evaluationIssueId ?? null,
snoozedUntil: input.snoozedUntil?.toISOString() ?? null,
snoozedUntil: effectiveSnoozedUntil?.toISOString() ?? null,
reason: input.reason ?? null,
},
});