refactor(runtime): unify run attempt policy
This commit is contained in:
1 parent
58a4be7fe7
commit
473486aa61
10 files changed
+314
-217
No files matched your search
@@ -6,6 +6,7 @@ import type { ResolveWorkspaceFn } from "./run-dispatch";
|
||||
import { WorkspaceConfigurationError } from "./resolve-workspace";
|
||||
import { pickModelForAgent } from "../model-dispatch";
|
||||
import { HUMAN_SENDER_ID } from "../messaging/dispatch";
|
||||
import { decideRunAttempt } from "../run-attempt-policy";
|
||||
|
||||
/**
|
||||
* Every reason a heartbeat attempt can be rejected before a run is
|
||||
@@ -116,15 +117,19 @@ export function createHeartbeat(
|
||||
// agent runs at all.
|
||||
const heartbeatPolicy = db.agents.getHeartbeatPolicy(agent.id);
|
||||
|
||||
// maxConcurrentRuns check: count active runs for this agent
|
||||
const activeRunCount = db.agents.countActiveRunsForAgent(agent.id, companyId);
|
||||
if (activeRunCount >= heartbeatPolicy.maxConcurrentRuns) return SKIPPED("max_concurrent_runs");
|
||||
|
||||
// Cooldown check: ensure enough time has passed since last heartbeat
|
||||
if (!isMessageWake && agent.lastHeartbeatAt) {
|
||||
const elapsed = Date.now() - new Date(agent.lastHeartbeatAt).getTime();
|
||||
const cooldownMs = heartbeatPolicy.cooldownSec * 1000;
|
||||
if (elapsed < cooldownMs) return SKIPPED("cooldown");
|
||||
const runAttemptDecision = decideRunAttempt({
|
||||
kind: "heartbeat",
|
||||
trigger,
|
||||
activeRunCount,
|
||||
maxConcurrentRuns: heartbeatPolicy.maxConcurrentRuns,
|
||||
lastStartedAtMs: agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : null,
|
||||
cooldownMs: heartbeatPolicy.cooldownSec * 1000,
|
||||
nowMs: Date.now(),
|
||||
bypassCooldown: isMessageWake,
|
||||
});
|
||||
if (runAttemptDecision.action === "skip") {
|
||||
return SKIPPED(runAttemptDecision.reason);
|
||||
}
|
||||
|
||||
// wakeOnDemand gate: if disabled, only timer/bootstrap triggers allowed
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
export const DEFAULT_MAX_CONSECUTIVE_SELF_WAKES = 5;
|
||||
|
||||
export interface HeartbeatAttemptInput {
|
||||
kind: "heartbeat";
|
||||
trigger: string;
|
||||
activeRunCount: number;
|
||||
maxConcurrentRuns: number;
|
||||
lastStartedAtMs: number | null;
|
||||
cooldownMs: number;
|
||||
nowMs: number;
|
||||
bypassCooldown: boolean;
|
||||
}
|
||||
|
||||
export interface TaskFailureAttemptInput {
|
||||
kind: "task_failure";
|
||||
retryCount: number;
|
||||
maxRetryAttempts: number;
|
||||
}
|
||||
|
||||
export interface SelfWakeAttemptInput {
|
||||
kind: "self_wake";
|
||||
hasWorkToContinue: boolean;
|
||||
consecutiveSelfWakes: number;
|
||||
maxConsecutiveSelfWakes: number;
|
||||
}
|
||||
|
||||
export type HeartbeatAttemptDecision =
|
||||
| { action: "run" }
|
||||
| { action: "skip"; reason: "max_concurrent_runs" | "cooldown"; waitMs?: number };
|
||||
|
||||
export type TaskFailureAttemptDecision =
|
||||
| { action: "retry"; attemptNumber: number }
|
||||
| { action: "escalate" };
|
||||
|
||||
export type SelfWakeAttemptDecision =
|
||||
| { action: "run" }
|
||||
| { action: "skip"; reason: "no_work" | "self_wake_exhausted" };
|
||||
|
||||
type RunAttemptInput =
|
||||
| HeartbeatAttemptInput
|
||||
| TaskFailureAttemptInput
|
||||
| SelfWakeAttemptInput;
|
||||
|
||||
type RunAttemptDecision =
|
||||
| HeartbeatAttemptDecision
|
||||
| TaskFailureAttemptDecision
|
||||
| SelfWakeAttemptDecision;
|
||||
|
||||
export function runAttemptKey(agentId: string, companyId: string): string {
|
||||
return `${companyId}:${agentId}`;
|
||||
}
|
||||
|
||||
function assertNonNegativeInteger(name: string, value: number): void {
|
||||
if (!Number.isInteger(value) || value < 0) {
|
||||
throw new Error(`run-attempt policy: ${name} must be a non-negative integer`);
|
||||
}
|
||||
}
|
||||
|
||||
function assertNonNegativeNumber(name: string, value: number): void {
|
||||
if (!Number.isFinite(value) || value < 0) {
|
||||
throw new Error(`run-attempt policy: ${name} must be a non-negative number`);
|
||||
}
|
||||
}
|
||||
|
||||
export function decideRunAttempt(input: HeartbeatAttemptInput): HeartbeatAttemptDecision;
|
||||
export function decideRunAttempt(input: TaskFailureAttemptInput): TaskFailureAttemptDecision;
|
||||
export function decideRunAttempt(input: SelfWakeAttemptInput): SelfWakeAttemptDecision;
|
||||
export function decideRunAttempt(input: RunAttemptInput): RunAttemptDecision {
|
||||
switch (input.kind) {
|
||||
case "heartbeat": {
|
||||
assertNonNegativeInteger("activeRunCount", input.activeRunCount);
|
||||
assertNonNegativeInteger("maxConcurrentRuns", input.maxConcurrentRuns);
|
||||
assertNonNegativeNumber("cooldownMs", input.cooldownMs);
|
||||
assertNonNegativeNumber("nowMs", input.nowMs);
|
||||
if (input.lastStartedAtMs !== null) {
|
||||
assertNonNegativeNumber("lastStartedAtMs", input.lastStartedAtMs);
|
||||
}
|
||||
|
||||
if (input.activeRunCount >= input.maxConcurrentRuns) {
|
||||
return { action: "skip", reason: "max_concurrent_runs" };
|
||||
}
|
||||
|
||||
if (!input.bypassCooldown && input.lastStartedAtMs !== null) {
|
||||
const elapsedMs = input.nowMs - input.lastStartedAtMs;
|
||||
if (elapsedMs < input.cooldownMs) {
|
||||
return {
|
||||
action: "skip",
|
||||
reason: "cooldown",
|
||||
waitMs: input.cooldownMs - elapsedMs,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
return { action: "run" };
|
||||
}
|
||||
|
||||
case "task_failure": {
|
||||
assertNonNegativeInteger("retryCount", input.retryCount);
|
||||
assertNonNegativeInteger("maxRetryAttempts", input.maxRetryAttempts);
|
||||
if (input.retryCount < input.maxRetryAttempts) {
|
||||
return { action: "retry", attemptNumber: input.retryCount + 1 };
|
||||
}
|
||||
return { action: "escalate" };
|
||||
}
|
||||
|
||||
case "self_wake": {
|
||||
assertNonNegativeInteger("consecutiveSelfWakes", input.consecutiveSelfWakes);
|
||||
assertNonNegativeInteger("maxConsecutiveSelfWakes", input.maxConsecutiveSelfWakes);
|
||||
if (!input.hasWorkToContinue) {
|
||||
return { action: "skip", reason: "no_work" };
|
||||
}
|
||||
if (input.consecutiveSelfWakes >= input.maxConsecutiveSelfWakes) {
|
||||
return { action: "skip", reason: "self_wake_exhausted" };
|
||||
}
|
||||
return { action: "run" };
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@ import { app, utilityProcess } from "electron";
|
||||
import { workerMessageSchema, type WorkerServiceDependencies } from "./worker/entry/types";
|
||||
import { handleWorkerExit } from "./worker/entry/crash-recovery";
|
||||
import { handleWorkerMessage } from "./worker/entry/message-handler";
|
||||
import { DEFAULT_MAX_CONSECUTIVE_SELF_WAKES } from "./run-attempt-policy";
|
||||
|
||||
export { workerMessageSchema } from "./worker/entry/types";
|
||||
|
||||
@@ -25,8 +26,8 @@ export function createWorkerService({
|
||||
}: WorkerServiceDependencies) {
|
||||
const MAX_RETRY_ATTEMPTS = 2;
|
||||
const RETRY_COOLDOWN_MS = 30_000; // 30 seconds
|
||||
const consecutiveWakeMap = new Map<string, number>();
|
||||
const MAX_CONSECUTIVE_WAKES = 5;
|
||||
const selfWakeAttemptMap = new Map<string, number>();
|
||||
const MAX_CONSECUTIVE_SELF_WAKES = DEFAULT_MAX_CONSECUTIVE_SELF_WAKES;
|
||||
const leaseRenewalMap = new Map<string, number>();
|
||||
|
||||
let worker: ReturnType<typeof utilityProcess.fork> | null = null;
|
||||
@@ -131,8 +132,8 @@ export function createWorkerService({
|
||||
destroyed,
|
||||
MAX_RETRY_ATTEMPTS,
|
||||
RETRY_COOLDOWN_MS,
|
||||
consecutiveWakeMap,
|
||||
MAX_CONSECUTIVE_WAKES,
|
||||
selfWakeAttemptMap,
|
||||
MAX_CONSECUTIVE_SELF_WAKES,
|
||||
pendingRetryTimers,
|
||||
leaseRenewalMap,
|
||||
);
|
||||
|
||||
@@ -17,8 +17,8 @@ export async function handleWorkerMessage(
|
||||
destroyed: boolean,
|
||||
MAX_RETRY_ATTEMPTS: number,
|
||||
RETRY_COOLDOWN_MS: number,
|
||||
consecutiveWakeMap: Map<string, number>,
|
||||
MAX_CONSECUTIVE_WAKES: number,
|
||||
selfWakeAttemptMap: Map<string, number>,
|
||||
MAX_CONSECUTIVE_SELF_WAKES: number,
|
||||
pendingRetryTimers: Set<ReturnType<typeof setTimeout>>,
|
||||
leaseRenewalMap: Map<string, number>,
|
||||
) {
|
||||
@@ -33,8 +33,8 @@ export async function handleWorkerMessage(
|
||||
destroyed,
|
||||
MAX_RETRY_ATTEMPTS,
|
||||
RETRY_COOLDOWN_MS,
|
||||
MAX_CONSECUTIVE_WAKES,
|
||||
consecutiveWakeMap,
|
||||
MAX_CONSECUTIVE_SELF_WAKES,
|
||||
selfWakeAttemptMap,
|
||||
pendingRetryTimers,
|
||||
leaseRenewalMap,
|
||||
};
|
||||
|
||||
@@ -22,8 +22,8 @@ export interface HandlerContext {
|
||||
destroyed: boolean;
|
||||
MAX_RETRY_ATTEMPTS: number;
|
||||
RETRY_COOLDOWN_MS: number;
|
||||
MAX_CONSECUTIVE_WAKES: number;
|
||||
consecutiveWakeMap: Map<string, number>;
|
||||
MAX_CONSECUTIVE_SELF_WAKES: number;
|
||||
selfWakeAttemptMap: Map<string, number>;
|
||||
pendingRetryTimers: Set<ReturnType<typeof setTimeout>>;
|
||||
leaseRenewalMap: Map<string, number>;
|
||||
}
|
||||
@@ -1,10 +1,10 @@
|
||||
import type { z } from "zod";
|
||||
import { createCircuitBreaker } from "../observability/circuit-breaker";
|
||||
import { handleFailureEscalation, handleSuccessReset } from "../observability/failure-escalation";
|
||||
import { persistRunEvent } from "../observability/run-event-recorder";
|
||||
import type { workerMessageSchema } from "../entry/types";
|
||||
import type { HandlerContext, HandlerDeps, WorkerMessageEvent } from "./handler-context";
|
||||
import { isUsageStatusUnattributed } from "../../run-usage-status";
|
||||
import { decideRunAttempt, runAttemptKey } from "../../run-attempt-policy";
|
||||
|
||||
type RunFinishedEvent = Extract<WorkerMessageEvent, { type: "run-finished" }>;
|
||||
type RunRecord = ReturnType<HandlerDeps["db"]["runs"]["getRun"]>;
|
||||
@@ -284,28 +284,35 @@ export function triggerSelfWake(
|
||||
hadDeferredWakes: boolean,
|
||||
): void {
|
||||
const { db, logger, isQuitting, wakeAgentIfPossible } = deps;
|
||||
const { destroyed, consecutiveWakeMap, MAX_CONSECUTIVE_WAKES } = context;
|
||||
const { destroyed, selfWakeAttemptMap, MAX_CONSECUTIVE_SELF_WAKES } = context;
|
||||
const { run } = rfCtx;
|
||||
const attemptKey = runAttemptKey(run.agentId, run.companyId);
|
||||
|
||||
const cb = createCircuitBreaker(consecutiveWakeMap, MAX_CONSECUTIVE_WAKES);
|
||||
|
||||
// Reset circuit breaker when a task actually progressed (successful run)
|
||||
// Reset self-wake attempts when a task actually progressed.
|
||||
if (event.status === "succeeded") {
|
||||
cb.reset(run.agentId, run.companyId);
|
||||
selfWakeAttemptMap.delete(attemptKey);
|
||||
}
|
||||
|
||||
// Self-wake with circuit breaker to prevent infinite loops
|
||||
// Self-wake through the shared run-attempt policy.
|
||||
if (hadDeferredWakes || isQuitting() || destroyed) return;
|
||||
|
||||
try {
|
||||
const actionableTasks = db.tasks.listAgentTasks(run.agentId, run.companyId, ["todo", "in_progress"]);
|
||||
const hasWorkToContinue = actionableTasks.length > 0 || hasUnfinishedTeamBuilding(db, run.agentId, run.companyId);
|
||||
if (!hasWorkToContinue) return;
|
||||
const consecutiveSelfWakes = selfWakeAttemptMap.get(attemptKey) ?? 0;
|
||||
const decision = decideRunAttempt({
|
||||
kind: "self_wake",
|
||||
hasWorkToContinue,
|
||||
consecutiveSelfWakes,
|
||||
maxConsecutiveSelfWakes: MAX_CONSECUTIVE_SELF_WAKES,
|
||||
});
|
||||
|
||||
const allowed = cb.tryWake(run.agentId, run.companyId);
|
||||
if (!allowed) {
|
||||
if (decision.action === "skip" && decision.reason === "no_work") return;
|
||||
|
||||
if (decision.action === "skip") {
|
||||
selfWakeAttemptMap.delete(attemptKey);
|
||||
logger.warn(
|
||||
`[self-wake] Circuit breaker: agent ${run.agentId} has self-woken ${MAX_CONSECUTIVE_WAKES} times without progress. Pausing self-wake.`,
|
||||
`[self-wake] Run-attempt policy paused agent ${run.agentId} after ${MAX_CONSECUTIVE_SELF_WAKES} self-wakes without progress.`,
|
||||
);
|
||||
// Escalate to manager since agent is stuck
|
||||
try {
|
||||
@@ -319,6 +326,7 @@ export function triggerSelfWake(
|
||||
);
|
||||
}
|
||||
} else {
|
||||
selfWakeAttemptMap.set(attemptKey, consecutiveSelfWakes + 1);
|
||||
wakeAgentIfPossible(run.agentId, run.companyId, "task_continuation");
|
||||
}
|
||||
} catch (error) {
|
||||
@@ -336,9 +344,9 @@ export function triggerSelfWake(
|
||||
* waiting for the 2-minute heartbeat timer takes bootstrap from
|
||||
* 10-plus minutes down to the LLM's round-trip time.
|
||||
*
|
||||
* The circuit breaker on the caller side still applies, so a CEO that
|
||||
* The run-attempt policy on the caller side still applies, so a CEO that
|
||||
* keeps completing runs without actually creating hires gets paused
|
||||
* after MAX_CONSECUTIVE_WAKES attempts.
|
||||
* after MAX_CONSECUTIVE_SELF_WAKES attempts.
|
||||
*/
|
||||
function hasUnfinishedTeamBuilding(
|
||||
db: HandlerDeps["db"],
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
/**
|
||||
* Self-wake circuit breaker.
|
||||
*
|
||||
* Prevents infinite self-wake loops when an agent is stuck on a task.
|
||||
* After MAX_CONSECUTIVE_WAKES consecutive wakes without progress (i.e.,
|
||||
* without a successful run), self-wake is suppressed so the manager can
|
||||
* be notified instead.
|
||||
*/
|
||||
|
||||
export const MAX_CONSECUTIVE_WAKES = 5;
|
||||
|
||||
export interface CircuitBreaker {
|
||||
/** Attempt a self-wake. Returns true if allowed, false if circuit breaker trips. */
|
||||
tryWake(agentId: string, companyId: string): boolean;
|
||||
/** Reset the counter for an agent (called on task progress / successful run). */
|
||||
reset(agentId: string, companyId: string): void;
|
||||
/** Get the current consecutive-wake count for an agent. */
|
||||
getCount(agentId: string, companyId: string): number;
|
||||
}
|
||||
|
||||
export function createCircuitBreaker(
|
||||
consecutiveWakeMap: Map<string, number>,
|
||||
maxConsecutive: number = MAX_CONSECUTIVE_WAKES,
|
||||
): CircuitBreaker {
|
||||
function wakeKey(agentId: string, companyId: string): string {
|
||||
return `${agentId}:${companyId}`;
|
||||
}
|
||||
|
||||
return {
|
||||
tryWake(agentId: string, companyId: string): boolean {
|
||||
const key = wakeKey(agentId, companyId);
|
||||
const count = consecutiveWakeMap.get(key) ?? 0;
|
||||
if (count >= maxConsecutive) {
|
||||
consecutiveWakeMap.delete(key);
|
||||
return false;
|
||||
}
|
||||
consecutiveWakeMap.set(key, count + 1);
|
||||
return true;
|
||||
},
|
||||
|
||||
reset(agentId: string, companyId: string): void {
|
||||
consecutiveWakeMap.delete(wakeKey(agentId, companyId));
|
||||
},
|
||||
|
||||
getCount(agentId: string, companyId: string): number {
|
||||
return consecutiveWakeMap.get(wakeKey(agentId, companyId)) ?? 0;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -9,6 +9,8 @@
|
||||
* 5. Task succeeds -> reset retry count
|
||||
*/
|
||||
|
||||
import { decideRunAttempt } from "../../run-attempt-policy";
|
||||
|
||||
export interface EscalationDeps {
|
||||
db: {
|
||||
tasks: {
|
||||
@@ -47,16 +49,21 @@ export function handleFailureEscalation(
|
||||
): EscalationResult {
|
||||
const { db, wakeAgentIfPossible, notify, logger } = deps;
|
||||
const retryCount = db.tasks.getTaskRetryCount(run.taskId);
|
||||
const retryDecision = decideRunAttempt({
|
||||
kind: "task_failure",
|
||||
retryCount,
|
||||
maxRetryAttempts,
|
||||
});
|
||||
|
||||
if (retryCount < maxRetryAttempts) {
|
||||
if (retryDecision.action === "retry") {
|
||||
db.tasks.incrementTaskRetryCount(run.taskId);
|
||||
db.tasks.advanceTaskStatus(
|
||||
run.taskId,
|
||||
"todo",
|
||||
`Run ${eventStatus} (attempt ${retryCount + 1}/${maxRetryAttempts}). Auto-retrying after ${retryCooldownMs / 1000}s cooldown.`,
|
||||
`Run ${eventStatus} (attempt ${retryDecision.attemptNumber}/${maxRetryAttempts}). Auto-retrying after ${retryCooldownMs / 1000}s cooldown.`,
|
||||
);
|
||||
logger.info(
|
||||
`[escalation] Retrying task ${run.taskId} (attempt ${retryCount + 1}/${maxRetryAttempts}) for agent ${run.agentId}`,
|
||||
`[escalation] Retrying task ${run.taskId} (attempt ${retryDecision.attemptNumber}/${maxRetryAttempts}) for agent ${run.agentId}`,
|
||||
);
|
||||
scheduleRetry(() => {
|
||||
wakeAgentIfPossible(run.agentId, run.companyId, "retry_failed");
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
DEFAULT_MAX_CONSECUTIVE_SELF_WAKES,
|
||||
decideRunAttempt,
|
||||
runAttemptKey,
|
||||
} from "@main/run-attempt-policy";
|
||||
|
||||
describe("run-attempt policy", () => {
|
||||
describe("heartbeat dispatch", () => {
|
||||
it("allows heartbeat when concurrency and cooldown gates pass", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "heartbeat",
|
||||
trigger: "timer",
|
||||
activeRunCount: 0,
|
||||
maxConcurrentRuns: 1,
|
||||
lastStartedAtMs: 1_000,
|
||||
cooldownMs: 10_000,
|
||||
nowMs: 12_000,
|
||||
bypassCooldown: false,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "run" });
|
||||
});
|
||||
|
||||
it("blocks heartbeat when active runs reach the policy ceiling", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "heartbeat",
|
||||
trigger: "timer",
|
||||
activeRunCount: 1,
|
||||
maxConcurrentRuns: 1,
|
||||
lastStartedAtMs: null,
|
||||
cooldownMs: 10_000,
|
||||
nowMs: 12_000,
|
||||
bypassCooldown: false,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "skip", reason: "max_concurrent_runs" });
|
||||
});
|
||||
|
||||
it("blocks timer heartbeat inside cooldown and reports remaining wait", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "heartbeat",
|
||||
trigger: "timer",
|
||||
activeRunCount: 0,
|
||||
maxConcurrentRuns: 1,
|
||||
lastStartedAtMs: 10_000,
|
||||
cooldownMs: 30_000,
|
||||
nowMs: 25_000,
|
||||
bypassCooldown: false,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "skip", reason: "cooldown", waitMs: 15_000 });
|
||||
});
|
||||
|
||||
it("allows message wakes to bypass cooldown", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "heartbeat",
|
||||
trigger: "message",
|
||||
activeRunCount: 0,
|
||||
maxConcurrentRuns: 1,
|
||||
lastStartedAtMs: 10_000,
|
||||
cooldownMs: 30_000,
|
||||
nowMs: 25_000,
|
||||
bypassCooldown: true,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "run" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("task failure attempts", () => {
|
||||
it("retries while the task is under its retry budget", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "task_failure",
|
||||
retryCount: 1,
|
||||
maxRetryAttempts: 2,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "retry", attemptNumber: 2 });
|
||||
});
|
||||
|
||||
it("escalates when the retry budget is exhausted", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "task_failure",
|
||||
retryCount: 2,
|
||||
maxRetryAttempts: 2,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "escalate" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("self-wake continuation", () => {
|
||||
it("allows self-wake when work remains under the continuation budget", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "self_wake",
|
||||
hasWorkToContinue: true,
|
||||
consecutiveSelfWakes: DEFAULT_MAX_CONSECUTIVE_SELF_WAKES - 1,
|
||||
maxConsecutiveSelfWakes: DEFAULT_MAX_CONSECUTIVE_SELF_WAKES,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "run" });
|
||||
});
|
||||
|
||||
it("skips self-wake when there is no remaining work", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "self_wake",
|
||||
hasWorkToContinue: false,
|
||||
consecutiveSelfWakes: 0,
|
||||
maxConsecutiveSelfWakes: DEFAULT_MAX_CONSECUTIVE_SELF_WAKES,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "skip", reason: "no_work" });
|
||||
});
|
||||
|
||||
it("skips self-wake when the continuation budget is exhausted", () => {
|
||||
const decision = decideRunAttempt({
|
||||
kind: "self_wake",
|
||||
hasWorkToContinue: true,
|
||||
consecutiveSelfWakes: DEFAULT_MAX_CONSECUTIVE_SELF_WAKES,
|
||||
maxConsecutiveSelfWakes: DEFAULT_MAX_CONSECUTIVE_SELF_WAKES,
|
||||
});
|
||||
|
||||
expect(decision).toEqual({ action: "skip", reason: "self_wake_exhausted" });
|
||||
});
|
||||
});
|
||||
|
||||
it("keys attempt history by company and agent", () => {
|
||||
expect(runAttemptKey("agent-1", "company-1")).toBe("company-1:agent-1");
|
||||
});
|
||||
|
||||
it("rejects invalid counters loudly", () => {
|
||||
expect(() =>
|
||||
decideRunAttempt({
|
||||
kind: "task_failure",
|
||||
retryCount: -1,
|
||||
maxRetryAttempts: 2,
|
||||
}),
|
||||
).toThrow("retryCount");
|
||||
});
|
||||
});
|
||||
@@ -1,134 +0,0 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { createCircuitBreaker, MAX_CONSECUTIVE_WAKES } from "@main/worker/observability/circuit-breaker";
|
||||
|
||||
/**
|
||||
* Tests for the self-wake circuit breaker used in worker-message-handler.ts.
|
||||
*
|
||||
* The circuit breaker prevents infinite self-wake loops when an agent is stuck
|
||||
* on a task. After MAX_CONSECUTIVE_WAKES (5) consecutive wakes without progress
|
||||
* (i.e., without a successful run), self-wake is suppressed and the agent's
|
||||
* manager is notified instead.
|
||||
*/
|
||||
|
||||
describe("self-wake circuit breaker", () => {
|
||||
it("allows the first self-wake (counter=1)", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
const allowed = cb.tryWake("agent-1", "company-1");
|
||||
expect(allowed).toBe(true);
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(1);
|
||||
});
|
||||
|
||||
it("allows wakes up to MAX_CONSECUTIVE_WAKES-1 (counter=4)", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
for (let i = 0; i < MAX_CONSECUTIVE_WAKES - 1; i++) {
|
||||
expect(cb.tryWake("agent-1", "company-1")).toBe(true);
|
||||
}
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(MAX_CONSECUTIVE_WAKES - 1);
|
||||
});
|
||||
|
||||
it("trips after MAX_CONSECUTIVE_WAKES (5) consecutive wakes without progress", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
// First 5 wakes succeed (counter goes 0->1, 1->2, 2->3, 3->4, 4->5)
|
||||
for (let i = 0; i < MAX_CONSECUTIVE_WAKES; i++) {
|
||||
expect(cb.tryWake("agent-1", "company-1")).toBe(true);
|
||||
}
|
||||
// 6th wake should trip the breaker
|
||||
const tripped = cb.tryWake("agent-1", "company-1");
|
||||
expect(tripped).toBe(false);
|
||||
});
|
||||
|
||||
it("deletes the counter after circuit breaker trips (reset for next cycle)", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
for (let i = 0; i < MAX_CONSECUTIVE_WAKES; i++) {
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
}
|
||||
// Trip the breaker
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
// Counter should be deleted
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(0);
|
||||
});
|
||||
|
||||
it("resets counter to 0 on task status change", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(2);
|
||||
|
||||
// Simulate a task status change -> reset
|
||||
cb.reset("agent-1", "company-1");
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(0);
|
||||
});
|
||||
|
||||
it("maintains per-agent isolation (different agents have separate counters)", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
// Agent 1 wakes 4 times
|
||||
for (let i = 0; i < 4; i++) {
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
}
|
||||
// Agent 2 wakes 2 times
|
||||
cb.tryWake("agent-2", "company-1");
|
||||
cb.tryWake("agent-2", "company-1");
|
||||
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(4);
|
||||
expect(cb.getCount("agent-2", "company-1")).toBe(2);
|
||||
});
|
||||
|
||||
it("maintains per-company isolation for the same agent", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
cb.tryWake("agent-1", "company-2");
|
||||
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(2);
|
||||
expect(cb.getCount("agent-1", "company-2")).toBe(1);
|
||||
});
|
||||
|
||||
it("allows wakes again after reset even if previously near limit", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
// Get to 4 wakes
|
||||
for (let i = 0; i < MAX_CONSECUTIVE_WAKES - 1; i++) {
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
}
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(MAX_CONSECUTIVE_WAKES - 1);
|
||||
|
||||
// Reset
|
||||
cb.reset("agent-1", "company-1");
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(0);
|
||||
|
||||
// Should allow wakes again
|
||||
expect(cb.tryWake("agent-1", "company-1")).toBe(true);
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(1);
|
||||
});
|
||||
|
||||
it("resetting one agent does not affect another", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb = createCircuitBreaker(map);
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
cb.tryWake("agent-1", "company-1");
|
||||
cb.tryWake("agent-2", "company-1");
|
||||
cb.tryWake("agent-2", "company-1");
|
||||
cb.tryWake("agent-2", "company-1");
|
||||
|
||||
cb.reset("agent-1", "company-1");
|
||||
expect(cb.getCount("agent-1", "company-1")).toBe(0);
|
||||
expect(cb.getCount("agent-2", "company-1")).toBe(3);
|
||||
});
|
||||
|
||||
it("shares state with the underlying map (same map = same state)", () => {
|
||||
const map = new Map<string, number>();
|
||||
const cb1 = createCircuitBreaker(map);
|
||||
const cb2 = createCircuitBreaker(map);
|
||||
cb1.tryWake("agent-1", "company-1");
|
||||
cb1.tryWake("agent-1", "company-1");
|
||||
// cb2 wraps the same map, so it sees the same count
|
||||
expect(cb2.getCount("agent-1", "company-1")).toBe(2);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user