diff --git a/src/main/orchestration/heartbeat.ts b/src/main/orchestration/heartbeat.ts index a581215b..18fd1677 100644 --- a/src/main/orchestration/heartbeat.ts +++ b/src/main/orchestration/heartbeat.ts @@ -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 diff --git a/src/main/run-attempt-policy.ts b/src/main/run-attempt-policy.ts new file mode 100644 index 00000000..10853fe2 --- /dev/null +++ b/src/main/run-attempt-policy.ts @@ -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" }; + } + } +} diff --git a/src/main/worker-service.ts b/src/main/worker-service.ts index 5198879c..1027bf2b 100644 --- a/src/main/worker-service.ts +++ b/src/main/worker-service.ts @@ -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(); - const MAX_CONSECUTIVE_WAKES = 5; + const selfWakeAttemptMap = new Map(); + const MAX_CONSECUTIVE_SELF_WAKES = DEFAULT_MAX_CONSECUTIVE_SELF_WAKES; const leaseRenewalMap = new Map(); let worker: ReturnType | 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, ); diff --git a/src/main/worker/entry/message-handler.ts b/src/main/worker/entry/message-handler.ts index bd225fea..42059428 100644 --- a/src/main/worker/entry/message-handler.ts +++ b/src/main/worker/entry/message-handler.ts @@ -17,8 +17,8 @@ export async function handleWorkerMessage( destroyed: boolean, MAX_RETRY_ATTEMPTS: number, RETRY_COOLDOWN_MS: number, - consecutiveWakeMap: Map, - MAX_CONSECUTIVE_WAKES: number, + selfWakeAttemptMap: Map, + MAX_CONSECUTIVE_SELF_WAKES: number, pendingRetryTimers: Set>, leaseRenewalMap: Map, ) { @@ -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, }; diff --git a/src/main/worker/handlers/handler-context.ts b/src/main/worker/handlers/handler-context.ts index 5668fcda..838c261d 100644 --- a/src/main/worker/handlers/handler-context.ts +++ b/src/main/worker/handlers/handler-context.ts @@ -22,8 +22,8 @@ export interface HandlerContext { destroyed: boolean; MAX_RETRY_ATTEMPTS: number; RETRY_COOLDOWN_MS: number; - MAX_CONSECUTIVE_WAKES: number; - consecutiveWakeMap: Map; + MAX_CONSECUTIVE_SELF_WAKES: number; + selfWakeAttemptMap: Map; pendingRetryTimers: Set>; leaseRenewalMap: Map; } diff --git a/src/main/worker/handlers/run-finished-helpers.ts b/src/main/worker/handlers/run-finished-helpers.ts index 0094f2fc..80013495 100644 --- a/src/main/worker/handlers/run-finished-helpers.ts +++ b/src/main/worker/handlers/run-finished-helpers.ts @@ -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; type RunRecord = ReturnType; @@ -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"], diff --git a/src/main/worker/observability/circuit-breaker.ts b/src/main/worker/observability/circuit-breaker.ts deleted file mode 100644 index dcf0f00a..00000000 --- a/src/main/worker/observability/circuit-breaker.ts +++ /dev/null @@ -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, - 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; - }, - }; -} diff --git a/src/main/worker/observability/failure-escalation.ts b/src/main/worker/observability/failure-escalation.ts index 8d24b9dd..da2bfbfc 100644 --- a/src/main/worker/observability/failure-escalation.ts +++ b/src/main/worker/observability/failure-escalation.ts @@ -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"); diff --git a/tests/unit/run-attempt-policy.test.ts b/tests/unit/run-attempt-policy.test.ts new file mode 100644 index 00000000..123e8022 --- /dev/null +++ b/tests/unit/run-attempt-policy.test.ts @@ -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"); + }); +}); diff --git a/tests/unit/self-wake.test.ts b/tests/unit/self-wake.test.ts deleted file mode 100644 index 47d33346..00000000 --- a/tests/unit/self-wake.test.ts +++ /dev/null @@ -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(); - 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(); - 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(); - 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(); - 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(); - 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(); - 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(); - 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(); - 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(); - 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(); - 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); - }); -});