diff --git a/docs/code/worker.md b/docs/code/worker.md index cb80558c..93143707 100644 --- a/docs/code/worker.md +++ b/docs/code/worker.md @@ -41,6 +41,8 @@ devintern webhook serve Set `AGENT_HARNESS=codex,grok` (comma-separated, priority first) in the workspace `.env` so the worker keeps going when one agent hits a usage limit. Failover applies to every worker job: tracker tasks, PR review addressing, `@mention` runs, conflict resolution, scheduled automations, estimations, dashboard retries, and relay-driven work — except automations, estimations, or conflict resolution that pin their own `harness` (see [Choosing the agent per schedule](#choosing-the-agent-per-schedule)). Details: [Failover across multiple harnesses](./configuration.md#failover-across-multiple-harnesses). +When every harness in the chain is usage-limited, automation and estimation occurrences that come due are skipped (logged as `occurrence skipped: no agent harness is available until …`) and the schedule moves on to its next occurrence. They are not queued for catch-up. Dashboard **Run now** is rejected with the same reason. An entry that pins its own `harness` is skipped only while that harness is limited. + ## Error-monitor auto-fixes `[[error_monitors]]` entries let the worker turn unresolved production errors diff --git a/packages/code/src/lib/automation/acquirer.ts b/packages/code/src/lib/automation/acquirer.ts index fdd911cf..451c425f 100644 --- a/packages/code/src/lib/automation/acquirer.ts +++ b/packages/code/src/lib/automation/acquirer.ts @@ -116,6 +116,12 @@ export interface AutomationAcquirerOptions { jobKind?: string; /** Acquirer name override (worker banner / analytics worker mode). */ name?: string; + /** + * Epoch ms until which no agent harness can run this entry, or `null` when + * one is available. Defaults to the worker failover state: the whole chain + * for unpinned entries, the pinned harness's own window otherwise. + */ + harnessBlockedUntil?: (automation: AutomationConfig) => number | null; } interface ActiveAutomationRun { @@ -123,6 +129,24 @@ interface ActiveAutomationRun { lifecycle: Promise; } +/** + * Default harness gate: while every chain entry is usage-limited or cooling + * down, a spawned run would only fail within seconds and re-report the limit. + */ +export function defaultHarnessBlockedUntil(automation: AutomationConfig): number | null { + const failover = getWorkerFailover(); + if (!failover) return null; + const windows = failover.windows(); + if (automation.harness) { + const until = windows[automation.harness]; + return until !== undefined && until > Date.now() ? until : null; + } + if (!failover.allExhausted()) return null; + const now = Date.now(); + const until = Math.min(...Object.values(windows).filter((value) => value > now)); + return Number.isFinite(until) ? until : now; +} + /** Calculate the first future occurrence after `afterMs` (cron uses host timezone). */ export function nextAutomationDue(automation: AutomationConfig, afterMs: number): number { if (!automation.intervalMs && !automation.cron) { @@ -285,6 +309,17 @@ export class AutomationAcquirer implements Acquirer { } if (this.paused || state.nextDueAt > this.now()) continue; + const blockedUntil = this.harnessBlockedUntil(automation); + if (blockedUntil !== null) { + const skipNow = this.now(); + if (this.store.skipDue(stateId, skipNow, nextAutomationDue(automation, skipNow))) { + console.warn( + `⏭️ [${kind}:${automation.id}] occurrence skipped: ${harnessBlockedReason(blockedUntil)}`, + ); + } + continue; + } + const overlapNow = this.now(); if (state.leaseOwner && (state.leaseExpiresAt ?? 0) > overlapNow) { const skipNow = this.now(); @@ -428,6 +463,10 @@ export class AutomationAcquirer implements Acquirer { if (this.active.has(automationId)) { return { ok: false, reason: `automation "${automationId}" is already running` }; } + const blockedUntil = this.harnessBlockedUntil(automation); + if (blockedUntil !== null) { + return { ok: false, reason: harnessBlockedReason(blockedUntil) }; + } const stateId = this.stateId(automation); const state = this.store.get(stateId); @@ -541,6 +580,10 @@ export class AutomationAcquirer implements Acquirer { return resolveAutomationRunArgs(automation, this.options); } + private harnessBlockedUntil(automation: AutomationConfig): number | null { + return (this.options.harnessBlockedUntil ?? defaultHarnessBlockedUntil)(automation); + } + private jobKind(): string { return this.options.jobKind ?? "automation"; } @@ -573,6 +616,10 @@ export class AutomationAcquirer implements Acquirer { } } +function harnessBlockedReason(untilMs: number): string { + return `no agent harness is available until ${new Date(untilMs).toISOString()} (usage limit)`; +} + /** * Directory that receives one markdown task file per occurrence. * diff --git a/packages/code/src/lib/automation/state.ts b/packages/code/src/lib/automation/state.ts index 0d770465..1ea493ea 100644 --- a/packages/code/src/lib/automation/state.ts +++ b/packages/code/src/lib/automation/state.ts @@ -117,6 +117,19 @@ export class AutomationStateStore { return result.changes === 1; } + /** + * Skip a due occurrence without running it (no lease taken), advancing the + * cursor. Used when no agent harness can start a run right now. + */ + skipDue(automationId: string, now: number, nextDueAt: number): boolean { + const result = this.db.run( + `UPDATE automation_schedules SET last_scheduled_at = next_due_at, next_due_at = ? + WHERE automation_id = ? AND next_due_at <= ?`, + [nextDueAt, automationId, now], + ); + return result.changes === 1; + } + /** Coalesce a due occurrence skipped because the previous run is still active. */ skipOverlap(automationId: string, now: number, nextDueAt: number): boolean { const result = this.db.run( diff --git a/packages/code/tests/automation-acquirer.test.ts b/packages/code/tests/automation-acquirer.test.ts index 36c9e25c..800c05a7 100644 --- a/packages/code/tests/automation-acquirer.test.ts +++ b/packages/code/tests/automation-acquirer.test.ts @@ -100,6 +100,92 @@ describe("AutomationAcquirer", () => { expect(nextAutomationDue(cron, after)).toBeGreaterThan(after); }); + describe("harness usage-limit gate", () => { + const hourly = (extra: Partial = {}): AutomationConfig => ({ + id: "hourly", + enabled: true, + prompt: "p", + interval: "1h", + intervalMs: 3_600_000, + ...extra, + }); + + function build(blockedUntil: number | null, now: { value: number }) { + const dbPath = join(tmpdir(), `acquirer-gate-${Date.now()}-${Math.random()}.db`); + dbPaths.push(dbPath); + let spawned = 0; + let resolved = 0; + const acquirer = new AutomationAcquirer({ + automations: [hourly()], + dbPath, + now: () => now.value, + setTimer: () => 1 as unknown as ReturnType, + clearTimer: () => {}, + harnessBlockedUntil: () => blockedUntil, + resolveContext: async () => { + resolved++; + return { cwd: tmpdir(), env: {}, release: () => {} }; + }, + spawnRun: () => { + spawned++; + return { completion: Promise.resolve(true), terminate: () => {} }; + }, + spawnManualRun: () => { + spawned++; + return { completion: Promise.resolve(true), terminate: () => {} }; + }, + }); + return { acquirer, dbPath, counts: () => ({ spawned, resolved }) }; + } + + test("skips due occurrences and advances the cursor while the chain is exhausted", async () => { + const now = { value: 0 }; + const { acquirer, dbPath, counts } = build(10 * 3_600_000, now); + await acquirer.start(); + now.value = 3_600_000; + await acquirer.tick(); + expect(counts()).toEqual({ spawned: 0, resolved: 0 }); + const store = new AutomationStateStore(dbPath); + expect(store.get("hourly")?.nextDueAt).toBe(2 * 3_600_000); + expect(store.get("hourly")?.leaseOwner).toBeUndefined(); + store.close(); + await acquirer.stop(); + }); + + test("rejects manual runs while the chain is exhausted", async () => { + const now = { value: 0 }; + const { acquirer, counts } = build(10 * 3_600_000, now); + await acquirer.start(); + const outcome = await acquirer.triggerManual("hourly"); + expect(outcome.ok).toBe(false); + expect(!outcome.ok && outcome.reason).toContain("no agent harness is available"); + expect(counts().spawned).toBe(0); + await acquirer.stop(); + }); + + test("runs normally when a harness is available", async () => { + const now = { value: 0 }; + const { acquirer, counts } = build(null, now); + await acquirer.start(); + now.value = 3_600_000; + await acquirer.tick(); + expect(counts()).toEqual({ spawned: 1, resolved: 1 }); + await acquirer.stop(); + }); + + test("default gate reads the worker failover chain and pinned-harness windows", async () => { + const { defaultHarnessBlockedUntil } = await import("../src/lib/automation/acquirer"); + await startWorkerFailover({ raw: "codex", checkInstalled: false, log: () => {} }); + expect(defaultHarnessBlockedUntil(hourly())).toBeNull(); + const { getWorkerFailover } = await import("../src/lib/worker/failover"); + const until = Date.now() + 3_600_000; + getWorkerFailover()?.reportFromHint({ untilMs: until }); + expect(defaultHarnessBlockedUntil(hourly())).toBe(until); + expect(defaultHarnessBlockedUntil(hourly({ harness: "codex" }))).toBe(until); + expect(defaultHarnessBlockedUntil(hourly({ harness: "grok" }))).toBeNull(); + }); + }); + test.each([ { label: "interval",