From 7c200157dc140c81fcc343c8b8ffe9e651a629e1 Mon Sep 17 00:00:00 2001 From: Copilot Date: Sun, 22 Mar 2026 22:06:38 +0200 Subject: [PATCH 1/6] feat: cooperative rate limiting with predictive circuit breaker (#515) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .changeset/cooperative-rate-limiting.md | 11 + .squad-templates/cooperative-rate-limiting.md | 229 ++++++++++++++++++ packages/squad-sdk/src/ralph/index.ts | 2 + packages/squad-sdk/src/ralph/rate-limiting.ts | 194 +++++++++++++++ test/rate-limiting.test.ts | 209 ++++++++++++++++ 5 files changed, 645 insertions(+) create mode 100644 .changeset/cooperative-rate-limiting.md create mode 100644 .squad-templates/cooperative-rate-limiting.md create mode 100644 packages/squad-sdk/src/ralph/rate-limiting.ts create mode 100644 test/rate-limiting.test.ts diff --git a/.changeset/cooperative-rate-limiting.md b/.changeset/cooperative-rate-limiting.md new file mode 100644 index 000000000..524a58109 --- /dev/null +++ b/.changeset/cooperative-rate-limiting.md @@ -0,0 +1,11 @@ +--- +"@bradygaster/squad-sdk": minor +--- + +feat: Cooperative rate limiting with predictive circuit breaker + +Added cooperative rate limiting patterns for multi-agent deployments: +- Traffic Light, Predictive Circuit Breaker, Priority Retry Windows +- Cooperative Token Pool for shared quota management + +Closes #515 diff --git a/.squad-templates/cooperative-rate-limiting.md b/.squad-templates/cooperative-rate-limiting.md new file mode 100644 index 000000000..bf56ef122 --- /dev/null +++ b/.squad-templates/cooperative-rate-limiting.md @@ -0,0 +1,229 @@ +# Cooperative Rate Limiting for Multi-Agent Deployments + +> Coordinate API quota across multiple Ralph instances to prevent cascading failures. + +## Problem + +The [circuit breaker template](ralph-circuit-breaker.md) handles single-instance rate limiting well. But when multiple Ralphs run across machines (or pods on K8s), each instance independently hits API limits: + +- **No coordination** — 5 Ralphs each think they have full API quota +- **Thundering herd** — All Ralphs retry simultaneously after rate limit resets +- **Priority inversion** — Low-priority work exhausts quota before critical work runs +- **Reactive only** — Circuit opens AFTER 429, wasting the failed request + +## Solution: 6-Pattern Architecture + +These patterns layer on top of the existing circuit breaker. Each is independent — adopt one or all. + +### Pattern 1: Traffic Light (RAAS — Rate-Aware Agent Scheduling) + +Map GitHub API `X-RateLimit-Remaining` to traffic light states: + +| State | Remaining % | Behavior | +|-------|------------|----------| +| 🟢 GREEN | >20% | Normal operation | +| 🟡 AMBER | 5–20% | Only P0 agents proceed | +| 🔴 RED | <5% | Block all except emergency P0 | + +```typescript +type TrafficLight = 'green' | 'amber' | 'red'; + +function getTrafficLight(remaining: number, limit: number): TrafficLight { + const pct = remaining / limit; + if (pct > 0.20) return 'green'; + if (pct > 0.05) return 'amber'; + return 'red'; +} + +function shouldProceed(light: TrafficLight, agentPriority: number): boolean { + if (light === 'green') return true; + if (light === 'amber') return agentPriority === 0; // P0 only + return false; // RED — block all +} +``` + +### Pattern 2: Cooperative Token Pool (CMARP) + +A shared JSON file (`~/.squad/rate-pool.json`) distributes API quota: + +```json +{ + "totalLimit": 5000, + "resetAt": "2026-03-22T20:00:00Z", + "allocations": { + "picard": { "priority": 0, "allocated": 2000, "used": 450, "leaseExpiry": "2026-03-22T19:55:00Z" }, + "data": { "priority": 1, "allocated": 1750, "used": 200, "leaseExpiry": "2026-03-22T19:55:00Z" }, + "ralph": { "priority": 2, "allocated": 1250, "used": 100, "leaseExpiry": "2026-03-22T19:55:00Z" } + } +} +``` + +**Rules:** +- P0 agents (Lead) get 40% of quota +- P1 agents (specialists) get 35% +- P2 agents (Ralph, Scribe) get 25% +- Stale leases (>5 minutes without heartbeat) are auto-recovered +- Each agent checks their remaining allocation before making API calls + +```typescript +interface RatePoolAllocation { + priority: number; + allocated: number; + used: number; + leaseExpiry: string; +} + +interface RatePool { + totalLimit: number; + resetAt: string; + allocations: Record; +} + +function canUseQuota(pool: RatePool, agentName: string): boolean { + const alloc = pool.allocations[agentName]; + if (!alloc) return true; // Unknown agent — allow (graceful) + + // Reclaim stale leases from crashed agents + const now = new Date(); + for (const [name, a] of Object.entries(pool.allocations)) { + if (new Date(a.leaseExpiry) < now && name !== agentName) { + a.allocated = 0; // Reclaim + } + } + + return alloc.used < alloc.allocated; +} +``` + +### Pattern 3: Predictive Circuit Breaker (PCB) + +Opens the circuit BEFORE getting a 429 by predicting when quota will run out: + +```typescript +interface RateSample { + timestamp: number; // Date.now() + remaining: number; // from X-RateLimit-Remaining header +} + +class PredictiveCircuitBreaker { + private samples: RateSample[] = []; + private readonly maxSamples = 10; + private readonly warningThresholdSeconds = 120; + + addSample(remaining: number): void { + this.samples.push({ timestamp: Date.now(), remaining }); + if (this.samples.length > this.maxSamples) { + this.samples.shift(); + } + } + + /** Predict seconds until quota exhaustion using linear regression */ + predictExhaustion(): number | null { + if (this.samples.length < 3) return null; + + const n = this.samples.length; + const first = this.samples[0]; + const last = this.samples[n - 1]; + + const elapsedMs = last.timestamp - first.timestamp; + if (elapsedMs === 0) return null; + + const consumedPerMs = (first.remaining - last.remaining) / elapsedMs; + if (consumedPerMs <= 0) return null; // Not consuming — safe + + const msUntilExhausted = last.remaining / consumedPerMs; + return msUntilExhausted / 1000; + } + + shouldOpen(): boolean { + const eta = this.predictExhaustion(); + if (eta === null) return false; + return eta < this.warningThresholdSeconds; + } +} +``` + +### Pattern 4: Priority Retry Windows (PWJG) + +Non-overlapping jitter windows prevent thundering herd: + +| Priority | Retry Window | Description | +|----------|-------------|-------------| +| P0 (Lead) | 500ms–5s | Recovers first | +| P1 (Specialists) | 2s–30s | Moderate delay | +| P2 (Ralph/Scribe) | 5s–60s | Most patient | + +```typescript +function getRetryDelay(priority: number, attempt: number): number { + const windows: Record = { + 0: [500, 5000], // P0: 500ms–5s + 1: [2000, 30000], // P1: 2s–30s + 2: [5000, 60000], // P2: 5s–60s + }; + + const [min, max] = windows[priority] ?? windows[2]; + const base = Math.min(min * Math.pow(2, attempt), max); + const jitter = Math.random() * base * 0.5; + return base + jitter; +} +``` + +### Pattern 5: Resource Epoch Tracker (RET) + +Heartbeat-based lease system for multi-machine deployments: + +```typescript +interface ResourceLease { + agent: string; + machine: string; + leaseStart: string; + leaseExpiry: string; // Typically 5 minutes from now + allocated: number; +} + +// Each agent renews its lease every 2 minutes +// If lease expires (agent crashed), allocation is reclaimed +``` + +### Pattern 6: Cascade Dependency Detector (CDD) + +Track downstream failures and apply backpressure: + +``` +Agent A (rate limited) → Agent B (waiting for A) → Agent C (waiting for B) + ↑ Backpressure signal: "don't start new work" +``` + +When a dependency is rate-limited, upstream agents should pause new work rather than queuing requests that will fail. + +## Kubernetes Integration + +On K8s, cooperative rate limiting can use KEDA to scale pods based on API quota: + +```yaml +apiVersion: keda.sh/v1alpha1 +kind: ScaledObject +spec: + scaleTargetRef: + name: ralph-deployment + triggers: + - type: external + metadata: + scalerAddress: keda-copilot-scaler:6000 + # Scaler returns 0 when rate limited → pods scale to zero +``` + +See [keda-copilot-scaler](https://github.com/tamirdresher/keda-copilot-scaler) for a complete implementation. + +## Quick Start + +1. **Minimum viable:** Adopt Pattern 1 (Traffic Light) — read `X-RateLimit-Remaining` from API responses +2. **Multi-machine:** Add Pattern 2 (Cooperative Pool) — shared `rate-pool.json` +3. **Production:** Add Pattern 3 (Predictive CB) — prevent 429s entirely +4. **Kubernetes:** Add KEDA scaler for automatic pod scaling + +## References + +- [Circuit Breaker Template](ralph-circuit-breaker.md) — Foundation patterns +- [Squad on AKS](https://github.com/tamirdresher/squad-on-aks) — Production K8s deployment +- [KEDA Copilot Scaler](https://github.com/tamirdresher/keda-copilot-scaler) — Custom KEDA external scaler diff --git a/packages/squad-sdk/src/ralph/index.ts b/packages/squad-sdk/src/ralph/index.ts index f4ffa692b..cb946b201 100644 --- a/packages/squad-sdk/src/ralph/index.ts +++ b/packages/squad-sdk/src/ralph/index.ts @@ -183,3 +183,5 @@ export class RalphMonitor { this.eventBus = null; } } + +export { getTrafficLight, shouldProceed, getRetryDelay, PredictiveCircuitBreaker, canUseQuota, loadRatePool, type RatePool, type RatePoolAllocation, type RateSample, type TrafficLight, type AgentPriority } from './rate-limiting.js'; diff --git a/packages/squad-sdk/src/ralph/rate-limiting.ts b/packages/squad-sdk/src/ralph/rate-limiting.ts new file mode 100644 index 000000000..5b30eb16b --- /dev/null +++ b/packages/squad-sdk/src/ralph/rate-limiting.ts @@ -0,0 +1,194 @@ +/** + * Predictive Circuit Breaker — Rate Limit Protection + * + * Opens the circuit BEFORE getting a 429 by predicting when + * API quota will be exhausted using linear regression on + * recent rate limit header samples. + * + * @see https://github.com/bradygaster/squad/issues/515 + */ + +import { readFile, writeFile } from 'node:fs/promises'; +import { existsSync } from 'node:fs'; +import path from 'node:path'; +import os from 'node:os'; + +/** A rate limit sample from API response headers */ +export interface RateSample { + timestamp: number; // Date.now() + remaining: number; // from X-RateLimit-Remaining header + limit: number; // from X-RateLimit-Limit header +} + +/** Traffic light state for rate-aware scheduling */ +export type TrafficLight = 'green' | 'amber' | 'red'; + +/** Agent priority for quota allocation */ +export type AgentPriority = 0 | 1 | 2; + +/** Priority-based retry windows (ms) */ +const RETRY_WINDOWS: Record = { + 0: [500, 5_000], // P0 (Lead): 500ms–5s + 1: [2_000, 30_000], // P1 (Specialists): 2s–30s + 2: [5_000, 60_000], // P2 (Ralph/Scribe): 5s–60s +}; + +/** + * Determine traffic light from rate limit headers. + */ +export function getTrafficLight(remaining: number, limit: number): TrafficLight { + if (limit === 0) return 'red'; + const pct = remaining / limit; + if (pct > 0.20) return 'green'; + if (pct > 0.05) return 'amber'; + return 'red'; +} + +/** + * Check if an agent should proceed based on traffic light and priority. + * - GREEN: all agents proceed + * - AMBER: only P0 agents proceed + * - RED: no agents proceed + */ +export function shouldProceed(light: TrafficLight, priority: AgentPriority): boolean { + if (light === 'green') return true; + if (light === 'amber') return priority === 0; + return false; +} + +/** + * Get retry delay with priority-based jitter windows. + * Higher priority agents retry sooner with smaller windows. + */ +export function getRetryDelay(priority: AgentPriority, attempt: number): number { + const [min, max] = RETRY_WINDOWS[priority] ?? RETRY_WINDOWS[2]; + const base = Math.min(min * Math.pow(2, attempt), max); + const jitter = Math.random() * base * 0.5; + return Math.round(base + jitter); +} + +/** + * Predictive circuit breaker that opens BEFORE rate limit errors. + * + * Tracks the last N rate limit samples and uses linear regression + * to predict when quota will be exhausted. If predicted ETA is + * below the warning threshold, the circuit opens preemptively. + */ +export class PredictiveCircuitBreaker { + private samples: RateSample[] = []; + private readonly maxSamples: number; + private readonly warningThresholdSeconds: number; + + constructor(options?: { maxSamples?: number; warningThresholdSeconds?: number }) { + this.maxSamples = options?.maxSamples ?? 10; + this.warningThresholdSeconds = options?.warningThresholdSeconds ?? 120; + } + + /** Record a rate limit sample from API response headers */ + addSample(remaining: number, limit: number): void { + this.samples.push({ timestamp: Date.now(), remaining, limit }); + if (this.samples.length > this.maxSamples) { + this.samples.shift(); + } + } + + /** Get all recorded samples (for testing/debugging) */ + getSamples(): readonly RateSample[] { + return this.samples; + } + + /** + * Predict seconds until quota exhaustion using linear regression. + * Returns null if insufficient data or quota is not being consumed. + */ + predictExhaustion(): number | null { + if (this.samples.length < 3) return null; + + const n = this.samples.length; + const first = this.samples[0]; + const last = this.samples[n - 1]; + + const elapsedMs = last.timestamp - first.timestamp; + if (elapsedMs === 0) return null; + + const consumed = first.remaining - last.remaining; + if (consumed <= 0) return null; // Not consuming or recovering + + const consumedPerMs = consumed / elapsedMs; + const msUntilExhausted = last.remaining / consumedPerMs; + return msUntilExhausted / 1000; + } + + /** + * Should the circuit open preemptively? + * Returns true when predicted ETA to exhaustion is below threshold. + */ + shouldOpen(): boolean { + const eta = this.predictExhaustion(); + if (eta === null) return false; + return eta < this.warningThresholdSeconds; + } + + /** Reset all samples (e.g., after rate limit window resets) */ + reset(): void { + this.samples = []; + } +} + +/** Rate pool allocation for cooperative multi-agent quota management */ +export interface RatePoolAllocation { + priority: AgentPriority; + allocated: number; + used: number; + leaseExpiry: string; +} + +/** Shared rate pool state */ +export interface RatePool { + totalLimit: number; + resetAt: string; + allocations: Record; +} + +/** + * Check if an agent has remaining quota in the cooperative pool. + * Reclaims stale leases from crashed agents. + */ +export function canUseQuota(pool: RatePool, agentName: string): boolean { + const alloc = pool.allocations[agentName]; + if (!alloc) return true; // Unknown agent — allow gracefully + + // Reclaim stale leases + const now = new Date(); + for (const [name, a] of Object.entries(pool.allocations)) { + if (new Date(a.leaseExpiry) < now && name !== agentName) { + a.allocated = 0; + } + } + + return alloc.used < alloc.allocated; +} + +/** + * Load rate pool state from the shared file. + */ +export async function loadRatePool(teamRoot?: string): Promise { + const candidates: string[] = []; + + if (teamRoot) { + candidates.push(path.join(teamRoot, '.squad', 'rate-pool.json')); + } + candidates.push(path.join(os.homedir(), '.squad', 'rate-pool.json')); + + for (const candidate of candidates) { + if (existsSync(candidate)) { + try { + const raw = await readFile(candidate, 'utf8'); + return JSON.parse(raw) as RatePool; + } catch { + // Malformed — skip + } + } + } + return null; +} diff --git a/test/rate-limiting.test.ts b/test/rate-limiting.test.ts new file mode 100644 index 000000000..7013ab203 --- /dev/null +++ b/test/rate-limiting.test.ts @@ -0,0 +1,209 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import { + getTrafficLight, + shouldProceed, + getRetryDelay, + PredictiveCircuitBreaker, + canUseQuota, + type RatePool, + type AgentPriority, +} from '../packages/squad-sdk/src/ralph/rate-limiting.js'; + +describe('getTrafficLight', () => { + it('returns green when >20% remaining', () => { + expect(getTrafficLight(1500, 5000)).toBe('green'); + expect(getTrafficLight(1001, 5000)).toBe('green'); + }); + + it('returns amber when 5-20% remaining', () => { + expect(getTrafficLight(1000, 5000)).toBe('amber'); + expect(getTrafficLight(251, 5000)).toBe('amber'); + }); + + it('returns red when <5% remaining', () => { + expect(getTrafficLight(250, 5000)).toBe('red'); + expect(getTrafficLight(0, 5000)).toBe('red'); + }); + + it('returns red when limit is 0', () => { + expect(getTrafficLight(0, 0)).toBe('red'); + }); +}); + +describe('shouldProceed', () => { + it('allows all agents on green', () => { + expect(shouldProceed('green', 0)).toBe(true); + expect(shouldProceed('green', 1)).toBe(true); + expect(shouldProceed('green', 2)).toBe(true); + }); + + it('allows only P0 on amber', () => { + expect(shouldProceed('amber', 0)).toBe(true); + expect(shouldProceed('amber', 1)).toBe(false); + expect(shouldProceed('amber', 2)).toBe(false); + }); + + it('blocks all on red', () => { + expect(shouldProceed('red', 0)).toBe(false); + expect(shouldProceed('red', 1)).toBe(false); + expect(shouldProceed('red', 2)).toBe(false); + }); +}); + +describe('getRetryDelay', () => { + it('returns delay within P0 window', () => { + const delay = getRetryDelay(0, 0); + expect(delay).toBeGreaterThanOrEqual(500); + expect(delay).toBeLessThanOrEqual(7500); // 5000 + 50% jitter + }); + + it('returns delay within P2 window', () => { + const delay = getRetryDelay(2, 0); + expect(delay).toBeGreaterThanOrEqual(5000); + expect(delay).toBeLessThanOrEqual(90000); // 60000 + 50% jitter + }); + + it('increases with attempt count', () => { + const d0 = getRetryDelay(1, 0); + const d3 = getRetryDelay(1, 3); + // d3 should generally be larger (exponential backoff) + // But jitter makes this probabilistic, so we just check it runs + expect(d0).toBeGreaterThan(0); + expect(d3).toBeGreaterThan(0); + }); + + it('caps at max window', () => { + const delay = getRetryDelay(0, 100); // Very high attempt + expect(delay).toBeLessThanOrEqual(7500); // 5000 + 50% jitter max + }); +}); + +describe('PredictiveCircuitBreaker', () => { + let cb: PredictiveCircuitBreaker; + + beforeEach(() => { + cb = new PredictiveCircuitBreaker({ maxSamples: 5, warningThresholdSeconds: 120 }); + }); + + it('does not open with insufficient samples', () => { + cb.addSample(5000, 5000); + cb.addSample(4900, 5000); + expect(cb.shouldOpen()).toBe(false); // Only 2 samples + }); + + it('does not open when quota is not being consumed', () => { + cb.addSample(4000, 5000); + cb.addSample(4000, 5000); + cb.addSample(4000, 5000); + expect(cb.shouldOpen()).toBe(false); + }); + + it('opens when quota is being consumed rapidly', () => { + // Simulate rapid consumption: 1000 → 100 in ~60 seconds + const now = Date.now(); + cb = new PredictiveCircuitBreaker({ warningThresholdSeconds: 120 }); + // Manually set samples with timestamps + (cb as any).samples = [ + { timestamp: now - 60000, remaining: 1000, limit: 5000 }, + { timestamp: now - 40000, remaining: 700, limit: 5000 }, + { timestamp: now - 20000, remaining: 400, limit: 5000 }, + { timestamp: now, remaining: 100, limit: 5000 }, + ]; + + const eta = cb.predictExhaustion(); + expect(eta).not.toBeNull(); + expect(eta!).toBeLessThan(120); // Should predict exhaustion within 120s + expect(cb.shouldOpen()).toBe(true); + }); + + it('does not open when consumption rate is slow', () => { + const now = Date.now(); + (cb as any).samples = [ + { timestamp: now - 3600000, remaining: 5000, limit: 5000 }, // 1 hour ago + { timestamp: now - 1800000, remaining: 4900, limit: 5000 }, + { timestamp: now, remaining: 4800, limit: 5000 }, + ]; + + const eta = cb.predictExhaustion(); + expect(eta).not.toBeNull(); + expect(eta!).toBeGreaterThan(120); + expect(cb.shouldOpen()).toBe(false); + }); + + it('resets samples', () => { + cb.addSample(1000, 5000); + cb.addSample(500, 5000); + cb.addSample(100, 5000); + cb.reset(); + expect(cb.getSamples()).toHaveLength(0); + expect(cb.predictExhaustion()).toBeNull(); + }); + + it('limits to maxSamples', () => { + for (let i = 0; i < 10; i++) { + cb.addSample(5000 - i * 100, 5000); + } + expect(cb.getSamples()).toHaveLength(5); // maxSamples = 5 + }); +}); + +describe('canUseQuota', () => { + const futureExpiry = new Date(Date.now() + 300000).toISOString(); // 5 min from now + const pastExpiry = new Date(Date.now() - 60000).toISOString(); // 1 min ago + + it('allows unknown agents', () => { + const pool: RatePool = { + totalLimit: 5000, + resetAt: futureExpiry, + allocations: {}, + }; + expect(canUseQuota(pool, 'unknown-agent')).toBe(true); + }); + + it('allows agent with remaining quota', () => { + const pool: RatePool = { + totalLimit: 5000, + resetAt: futureExpiry, + allocations: { + picard: { priority: 0, allocated: 2000, used: 500, leaseExpiry: futureExpiry }, + }, + }; + expect(canUseQuota(pool, 'picard')).toBe(true); + }); + + it('blocks agent that exhausted quota', () => { + const pool: RatePool = { + totalLimit: 5000, + resetAt: futureExpiry, + allocations: { + ralph: { priority: 2, allocated: 1000, used: 1000, leaseExpiry: futureExpiry }, + }, + }; + expect(canUseQuota(pool, 'ralph')).toBe(false); + }); + + it('reclaims stale leases from other agents', () => { + const pool: RatePool = { + totalLimit: 5000, + resetAt: futureExpiry, + allocations: { + picard: { priority: 0, allocated: 2000, used: 500, leaseExpiry: pastExpiry }, // Stale + ralph: { priority: 2, allocated: 1000, used: 500, leaseExpiry: futureExpiry }, + }, + }; + canUseQuota(pool, 'ralph'); + expect(pool.allocations.picard.allocated).toBe(0); // Reclaimed + }); + + it('does not reclaim own stale lease', () => { + const pool: RatePool = { + totalLimit: 5000, + resetAt: futureExpiry, + allocations: { + ralph: { priority: 2, allocated: 1000, used: 500, leaseExpiry: pastExpiry }, + }, + }; + expect(canUseQuota(pool, 'ralph')).toBe(true); + expect(pool.allocations.ralph.allocated).toBe(1000); // Not reclaimed + }); +}); From 4138c2a174801f67467eeb0841254a128c846303 Mon Sep 17 00:00:00 2001 From: Copilot Date: Mon, 23 Mar 2026 00:21:31 +0200 Subject: [PATCH 2/6] fix: TypeScript strict null checks and add missing SDK export - Add non-null assertions for array indexing in PredictiveCircuitBreaker - Add ./ralph/rate-limiting export entry to SDK package.json Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- packages/squad-sdk/package.json | 4 ++++ packages/squad-sdk/src/ralph/rate-limiting.ts | 4 ++-- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/packages/squad-sdk/package.json b/packages/squad-sdk/package.json index 5fc8902e3..bb50a2aa2 100644 --- a/packages/squad-sdk/package.json +++ b/packages/squad-sdk/package.json @@ -78,6 +78,10 @@ "types": "./dist/ralph/triage.d.ts", "import": "./dist/ralph/triage.js" }, + "./ralph/rate-limiting": { + "types": "./dist/ralph/rate-limiting.d.ts", + "import": "./dist/ralph/rate-limiting.js" + }, "./casting": { "types": "./dist/casting/index.d.ts", "import": "./dist/casting/index.js" diff --git a/packages/squad-sdk/src/ralph/rate-limiting.ts b/packages/squad-sdk/src/ralph/rate-limiting.ts index 5b30eb16b..bcff9711b 100644 --- a/packages/squad-sdk/src/ralph/rate-limiting.ts +++ b/packages/squad-sdk/src/ralph/rate-limiting.ts @@ -105,8 +105,8 @@ export class PredictiveCircuitBreaker { if (this.samples.length < 3) return null; const n = this.samples.length; - const first = this.samples[0]; - const last = this.samples[n - 1]; + const first = this.samples[0]!; + const last = this.samples[n - 1]!; const elapsedMs = last.timestamp - first.timestamp; if (elapsedMs === 0) return null; From f00dd39283de309965736e678f4dc99ddf3b59cb Mon Sep 17 00:00:00 2001 From: Copilot Date: Sun, 22 Mar 2026 23:35:18 +0200 Subject: [PATCH 3/6] feat: wire rate limiting & circuit breaker into watch command (#515) Integrates PredictiveCircuitBreaker from squad-sdk into Ralph's watch loop: - Pre-round rate limit check via gh api rate_limit - Traffic light gating (GREEN/AMBER/RED) - Predictive circuit opening before 429 - Exponential backoff on rate limit errors - Half-open recovery with 2-success threshold - State persistence to .squad/ralph-circuit-breaker.json - Adds ghRateLimitCheck() and isRateLimitError() to gh-cli.ts Depends on #518 for the SDK module. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .changeset/rate-limit-watch-integration.md | 20 + packages/squad-cli/src/cli/commands/watch.ts | 889 +++++++++++-------- packages/squad-cli/src/cli/core/gh-cli.ts | 30 + test/cli/watch-rate-limiting.test.ts | 76 ++ 4 files changed, 660 insertions(+), 355 deletions(-) create mode 100644 .changeset/rate-limit-watch-integration.md create mode 100644 test/cli/watch-rate-limiting.test.ts diff --git a/.changeset/rate-limit-watch-integration.md b/.changeset/rate-limit-watch-integration.md new file mode 100644 index 000000000..620fb6515 --- /dev/null +++ b/.changeset/rate-limit-watch-integration.md @@ -0,0 +1,20 @@ +--- +'@bradygaster/squad-cli': minor +--- + +Wire rate limiting & circuit breaker into watch command + +Integrates the Predictive Circuit Breaker from squad-sdk/ralph/rate-limiting +into Ralph's watch polling loop: + +- Pre-round rate limit check via gh api rate_limit +- Traffic light gating - skips rounds when API quota is RED +- Predictive circuit opening - opens BEFORE hitting 429 +- 429 error handling with exponential backoff cooldown +- Half-open recovery - tests API after cooldown, 2 successes to close +- State persistence to .squad/ralph-circuit-breaker.json +- Board display shows traffic light indicator in round header + +Also adds ghRateLimitCheck() and isRateLimitError() to gh-cli.ts. + +Depends on #518 for the SDK rate-limiting module. diff --git a/packages/squad-cli/src/cli/commands/watch.ts b/packages/squad-cli/src/cli/commands/watch.ts index 8cdb176a8..446d740c3 100644 --- a/packages/squad-cli/src/cli/commands/watch.ts +++ b/packages/squad-cli/src/cli/commands/watch.ts @@ -1,355 +1,534 @@ -/** - * Watch command — Ralph's standalone polling process - */ - -import fs from 'node:fs'; -import path from 'node:path'; -import { detectSquadDir } from '../core/detect-squad-dir.js'; -import { fatal } from '../core/errors.js'; -import { GREEN, RED, DIM, BOLD, RESET, YELLOW } from '../core/output.js'; -import { - parseRoutingRules, - parseModuleOwnership, - parseRoster, - triageIssue, - type TriageIssue, -} from '@bradygaster/squad-sdk/ralph/triage'; -import { RalphMonitor } from '@bradygaster/squad-sdk/ralph'; -import { EventBus } from '@bradygaster/squad-sdk/runtime/event-bus'; -import { ghAvailable, ghAuthenticated, ghIssueList, ghIssueEdit, ghPrList, type GhIssue, type GhPullRequest } from '../core/gh-cli.js'; - -export interface BoardState { - untriaged: number; - assigned: number; - drafts: number; - needsReview: number; - changesRequested: number; - ciFailures: number; - readyToMerge: number; -} - -export function reportBoard(state: BoardState, round: number): void { - const total = Object.values(state).reduce((a, b) => a + b, 0); - - if (total === 0) { - console.log(`${DIM}📋 Board is clear — Ralph is idling${RESET}`); - return; - } - - console.log(`\n${BOLD}🔄 Ralph — Round ${round}${RESET}`); - console.log('━'.repeat(30)); - if (state.untriaged > 0) console.log(` 🔴 Untriaged: ${state.untriaged}`); - if (state.assigned > 0) console.log(` 🟡 Assigned: ${state.assigned}`); - if (state.drafts > 0) console.log(` 🟡 Draft PRs: ${state.drafts}`); - if (state.changesRequested > 0) console.log(` ⚠️ Changes requested: ${state.changesRequested}`); - if (state.ciFailures > 0) console.log(` ❌ CI failures: ${state.ciFailures}`); - if (state.needsReview > 0) console.log(` 🔵 Needs review: ${state.needsReview}`); - if (state.readyToMerge > 0) console.log(` 🟢 Ready to merge: ${state.readyToMerge}`); - console.log(); -} - -function emptyBoardState(): BoardState { - return { - untriaged: 0, - assigned: 0, - drafts: 0, - needsReview: 0, - changesRequested: 0, - ciFailures: 0, - readyToMerge: 0, - }; -} - -type PRBoardState = Pick & { - totalOpen: number; -}; - -async function checkPRs(roster: ReturnType): Promise { - const timestamp = new Date().toLocaleTimeString(); - const prs = await ghPrList({ state: 'open', limit: 20 }); - - // Filter to squad-related PRs (has squad label or branch starts with squad/) - const squadPRs: GhPullRequest[] = prs.filter(pr => - pr.labels.some(l => l.name.startsWith('squad')) || - pr.headRefName.startsWith('squad/') - ); - - if (squadPRs.length === 0) { - return { - drafts: 0, - needsReview: 0, - changesRequested: 0, - ciFailures: 0, - readyToMerge: 0, - totalOpen: 0, - }; - } - - const drafts = squadPRs.filter(pr => pr.isDraft); - const changesRequested = squadPRs.filter(pr => pr.reviewDecision === 'CHANGES_REQUESTED'); - const approved = squadPRs.filter(pr => pr.reviewDecision === 'APPROVED' && !pr.isDraft); - const ciFailures = squadPRs.filter(pr => - pr.statusCheckRollup?.some(check => check.state === 'FAILURE' || check.state === 'ERROR') - ); - const readyToMerge = approved.filter(pr => - !pr.statusCheckRollup?.some(c => c.state === 'FAILURE' || c.state === 'ERROR' || c.state === 'PENDING') - ); - const changesRequestedSet = new Set(changesRequested.map(pr => pr.number)); - const ciFailureSet = new Set(ciFailures.map(pr => pr.number)); - const readyToMergeSet = new Set(readyToMerge.map(pr => pr.number)); - const needsReview = squadPRs.filter(pr => - !pr.isDraft && - !changesRequestedSet.has(pr.number) && - !ciFailureSet.has(pr.number) && - !readyToMergeSet.has(pr.number) - ); - - const memberNames = new Set(roster.map(m => m.name.toLowerCase())); - - // Report each category - if (drafts.length > 0) { - console.log(`${DIM}[${timestamp}]${RESET} 🟡 ${drafts.length} draft PR(s) in progress`); - for (const pr of drafts) { - console.log(` ${DIM}PR #${pr.number}: ${pr.title} (${pr.author.login})${RESET}`); - } - } - if (changesRequested.length > 0) { - console.log(`${YELLOW}[${timestamp}]${RESET} ⚠️ ${changesRequested.length} PR(s) need revision`); - for (const pr of changesRequested) { - const owner = memberNames.has(pr.author.login.toLowerCase()) ? ` — ${pr.author.login}` : ''; - console.log(` PR #${pr.number}: ${pr.title} — changes requested${owner}`); - } - } - if (ciFailures.length > 0) { - console.log(`${RED}[${timestamp}]${RESET} ❌ ${ciFailures.length} PR(s) with CI failures`); - for (const pr of ciFailures) { - const failedChecks = pr.statusCheckRollup?.filter(c => c.state === 'FAILURE' || c.state === 'ERROR') || []; - const owner = memberNames.has(pr.author.login.toLowerCase()) ? ` — ${pr.author.login}` : ''; - console.log(` PR #${pr.number}: ${pr.title}${owner} — ${failedChecks.map(c => c.name).join(', ')}`); - } - } - if (approved.length > 0) { - if (readyToMerge.length > 0) { - console.log(`${GREEN}[${timestamp}]${RESET} 🟢 ${readyToMerge.length} PR(s) ready to merge`); - for (const pr of readyToMerge) { - console.log(` PR #${pr.number}: ${pr.title} — approved, CI green`); - } - } - } - - return { - drafts: drafts.length, - needsReview: needsReview.length, - changesRequested: changesRequestedSet.size, - ciFailures: ciFailureSet.size, - readyToMerge: readyToMergeSet.size, - totalOpen: squadPRs.length, - }; -} - -/** - * Run a single check cycle - */ -async function runCheck( - rules: ReturnType, - modules: ReturnType, - roster: ReturnType, - hasCopilot: boolean, - autoAssign: boolean -): Promise { - const timestamp = new Date().toLocaleTimeString(); - - try { - // Fetch open issues with squad label - const issues = await ghIssueList({ label: 'squad', state: 'open', limit: 20 }); - - // Find untriaged issues (no squad:{member} label) - const memberLabels = roster.map(m => m.label); - const untriaged = issues.filter(issue => { - const issueLabels = issue.labels.map(l => l.name); - return !memberLabels.some(ml => issueLabels.includes(ml)); - }); - const assignedIssues = issues.filter(issue => { - const issueLabels = issue.labels.map(l => l.name); - return memberLabels.some(ml => issueLabels.includes(ml)); - }); - - // Find unassigned squad:copilot issues - let unassignedCopilot: GhIssue[] = []; - if (hasCopilot && autoAssign) { - try { - const copilotIssues = await ghIssueList({ label: 'squad:copilot', state: 'open', limit: 10 }); - unassignedCopilot = copilotIssues.filter(i => !i.assignees || i.assignees.length === 0); - } catch { - // Label may not exist yet - } - } - - // Triage untriaged issues - for (const issue of untriaged) { - const triageInput: TriageIssue = { - number: issue.number, - title: issue.title, - body: issue.body, - labels: issue.labels.map((l) => l.name), - }; - const triage = triageIssue(triageInput, rules, modules, roster); - - if (triage) { - try { - await ghIssueEdit(issue.number, { addLabel: triage.agent.label }); - console.log( - `${GREEN}✓${RESET} [${timestamp}] Triaged #${issue.number} "${issue.title}" → ${triage.agent.name} (${triage.reason})` - ); - } catch (e) { - const err = e as Error; - console.error(`${RED}✗${RESET} [${timestamp}] Failed to label #${issue.number}: ${err.message}`); - } - } - } - - // Assign @copilot to unassigned copilot issues - for (const issue of unassignedCopilot) { - try { - await ghIssueEdit(issue.number, { addAssignee: 'copilot-swe-agent' }); - console.log(`${GREEN}✓${RESET} [${timestamp}] Assigned @copilot to #${issue.number} "${issue.title}"`); - } catch (e) { - const err = e as Error; - console.error(`${RED}✗${RESET} [${timestamp}] Failed to assign @copilot to #${issue.number}: ${err.message}`); - } - } - - const prState = await checkPRs(roster); - - return { - untriaged: untriaged.length, - assigned: assignedIssues.length, - ...prState, - }; - } catch (e) { - const err = e as Error; - console.error(`${RED}✗${RESET} [${timestamp}] Check failed: ${err.message}`); - return emptyBoardState(); - } -} - -/** - * Run watch command — Ralph's local polling process - */ -export async function runWatch(dest: string, intervalMinutes: number): Promise { - // Validate interval - if (isNaN(intervalMinutes) || intervalMinutes < 1) { - fatal('--interval must be a positive number of minutes'); - } - - // Detect squad directory - const squadDirInfo = detectSquadDir(dest); - const teamMd = path.join(squadDirInfo.path, 'team.md'); - const routingMdPath = path.join(squadDirInfo.path, 'routing.md'); - - if (!fs.existsSync(teamMd)) { - fatal('No squad found — run init first.'); - } - - // Verify gh CLI - if (!(await ghAvailable())) { - fatal('gh CLI not found — install from https://cli.github.com'); - } - - if (!(await ghAuthenticated())) { - console.error(`${YELLOW}⚠️${RESET} gh CLI not authenticated`); - console.error(` Run: ${BOLD}gh auth login${RESET}\n`); - fatal('gh authentication required'); - } - - // Parse team.md - const content = fs.readFileSync(teamMd, 'utf8'); - const roster = parseRoster(content); - const routingContent = fs.existsSync(routingMdPath) ? fs.readFileSync(routingMdPath, 'utf8') : ''; - const rules = parseRoutingRules(routingContent); - const modules = parseModuleOwnership(routingContent); - - if (roster.length === 0) { - fatal('No squad members found in team.md'); - } - - const hasCopilot = content.includes('🤖 Coding Agent') || content.includes('@copilot'); - const autoAssign = content.includes(''); - const monitorSessionId = 'ralph-watch'; - const eventBus = new EventBus(); - const monitor = new RalphMonitor({ - teamRoot: path.dirname(squadDirInfo.path), - healthCheckInterval: intervalMinutes * 60 * 1000, - staleSessionThreshold: intervalMinutes * 60 * 1000 * 3, - statePath: path.join(squadDirInfo.path, '.ralph-state.json'), - }); - await monitor.start(eventBus); - await eventBus.emit({ - type: 'session:created', - sessionId: monitorSessionId, - agentName: 'Ralph', - payload: { intervalMinutes }, - timestamp: new Date(), - }); - - // Print startup banner - console.log(`\n${BOLD}🔄 Ralph — Watch Mode${RESET}`); - console.log(`${DIM}Polling every ${intervalMinutes} minute(s) for squad work. Ctrl+C to stop.${RESET}\n`); - - let round = 0; - - // Run immediately, then on interval - round++; - const state = await runCheck(rules, modules, roster, hasCopilot, autoAssign); - await eventBus.emit({ - type: 'agent:milestone', - sessionId: monitorSessionId, - agentName: 'Ralph', - payload: { milestone: `Completed watch round ${round}`, task: 'watch cycle' }, - timestamp: new Date(), - }); - await monitor.healthCheck(); - reportBoard(state, round); - - return new Promise((resolve) => { - const intervalId = setInterval( - async () => { - round++; - const roundState = await runCheck(rules, modules, roster, hasCopilot, autoAssign); - await eventBus.emit({ - type: 'agent:milestone', - sessionId: monitorSessionId, - agentName: 'Ralph', - payload: { milestone: `Completed watch round ${round}`, task: 'watch cycle' }, - timestamp: new Date(), - }); - await monitor.healthCheck(); - reportBoard(roundState, round); - }, - intervalMinutes * 60 * 1000 - ); - - // Graceful shutdown - let isShuttingDown = false; - const shutdown = async () => { - if (isShuttingDown) return; - isShuttingDown = true; - clearInterval(intervalId); - process.off('SIGINT', shutdown); - process.off('SIGTERM', shutdown); - await eventBus.emit({ - type: 'session:destroyed', - sessionId: monitorSessionId, - agentName: 'Ralph', - payload: null, - timestamp: new Date(), - }); - await monitor.stop(); - console.log(`\n${DIM}🔄 Ralph — Watch stopped${RESET}`); - resolve(); - }; - - process.on('SIGINT', shutdown); - process.on('SIGTERM', shutdown); - }); -} +/** + * Watch command — Ralph's standalone polling process + * + * Integrates Predictive Circuit Breaker for rate limit protection. + * When GitHub API quota runs low, Ralph backs off automatically + * and resumes when quota recovers. + * + * @see https://github.com/bradygaster/squad/issues/515 + */ + +import fs from 'node:fs'; +import path from 'node:path'; +import { detectSquadDir } from '../core/detect-squad-dir.js'; +import { fatal } from '../core/errors.js'; +import { GREEN, RED, DIM, BOLD, RESET, YELLOW } from '../core/output.js'; +import { + parseRoutingRules, + parseModuleOwnership, + parseRoster, + triageIssue, + type TriageIssue, +} from '@bradygaster/squad-sdk/ralph/triage'; +import { RalphMonitor } from '@bradygaster/squad-sdk/ralph'; +import { EventBus } from '@bradygaster/squad-sdk/runtime/event-bus'; +import { + ghAvailable, + ghAuthenticated, + ghIssueList, + ghIssueEdit, + ghPrList, + ghRateLimitCheck, + isRateLimitError, + type GhIssue, + type GhPullRequest, + type GhRateLimitStatus, +} from '../core/gh-cli.js'; +import { + PredictiveCircuitBreaker, + getTrafficLight, + shouldProceed, + getRetryDelay, + type TrafficLight, +} from '@bradygaster/squad-sdk/ralph/rate-limiting'; + +export interface BoardState { + untriaged: number; + assigned: number; + drafts: number; + needsReview: number; + changesRequested: number; + ciFailures: number; + readyToMerge: number; +} + +/** Circuit breaker state persisted between rounds */ +interface CircuitBreakerState { + state: 'closed' | 'open' | 'half-open'; + consecutiveFailures: number; + consecutiveSuccesses: number; + lastRateLimitHit: string | null; + cooldownMinutes: number; + backoffRound: number; +} + +function defaultCBState(): CircuitBreakerState { + return { + state: 'closed', + consecutiveFailures: 0, + consecutiveSuccesses: 0, + lastRateLimitHit: null, + cooldownMinutes: 10, + backoffRound: 0, + }; +} + +function loadCBState(squadDir: string): CircuitBreakerState { + const filePath = path.join(squadDir, 'ralph-circuit-breaker.json'); + try { + if (fs.existsSync(filePath)) { + return JSON.parse(fs.readFileSync(filePath, 'utf8')); + } + } catch { /* corrupted — use defaults */ } + return defaultCBState(); +} + +function saveCBState(squadDir: string, cbState: CircuitBreakerState): void { + const filePath = path.join(squadDir, 'ralph-circuit-breaker.json'); + fs.writeFileSync(filePath, JSON.stringify(cbState, null, 2)); +} + +function trafficLightIcon(light: TrafficLight): string { + if (light === 'green') return `${GREEN}\u{1F7E2}${RESET}`; + if (light === 'amber') return `${YELLOW}\u{1F7E1}${RESET}`; + return `${RED}\u{1F534}${RESET}`; +} + +export function reportBoard(state: BoardState, round: number, light?: TrafficLight): void { + const total = Object.values(state).reduce((a, b) => a + b, 0); + + if (total === 0) { + console.log(`${DIM}\u{1F4CB} Board is clear \u2014 Ralph is idling${RESET}`); + return; + } + + const lightStr = light ? ` ${trafficLightIcon(light)}` : ''; + console.log(`\n${BOLD}\u{1F504} Ralph \u2014 Round ${round}${lightStr}${RESET}`); + console.log('\u2501'.repeat(30)); + if (state.untriaged > 0) console.log(` \u{1F534} Untriaged: ${state.untriaged}`); + if (state.assigned > 0) console.log(` \u{1F7E1} Assigned: ${state.assigned}`); + if (state.drafts > 0) console.log(` \u{1F7E1} Draft PRs: ${state.drafts}`); + if (state.changesRequested > 0) console.log(` \u26A0\uFE0F Changes requested: ${state.changesRequested}`); + if (state.ciFailures > 0) console.log(` \u274C CI failures: ${state.ciFailures}`); + if (state.needsReview > 0) console.log(` \u{1F535} Needs review: ${state.needsReview}`); + if (state.readyToMerge > 0) console.log(` \u{1F7E2} Ready to merge: ${state.readyToMerge}`); + console.log(); +} + +function emptyBoardState(): BoardState { + return { + untriaged: 0, + assigned: 0, + drafts: 0, + needsReview: 0, + changesRequested: 0, + ciFailures: 0, + readyToMerge: 0, + }; +} + +type PRBoardState = Pick & { + totalOpen: number; +}; + +async function checkPRs(roster: ReturnType): Promise { + const prs = await ghPrList({ state: 'open', limit: 20 }); + + const squadPRs: GhPullRequest[] = prs.filter(pr => + pr.labels.some(l => l.name.startsWith('squad')) || + pr.headRefName.startsWith('squad/') + ); + + if (squadPRs.length === 0) { + return { drafts: 0, needsReview: 0, changesRequested: 0, ciFailures: 0, readyToMerge: 0, totalOpen: 0 }; + } + + const drafts = squadPRs.filter(pr => pr.isDraft); + const changesRequested = squadPRs.filter(pr => pr.reviewDecision === 'CHANGES_REQUESTED'); + const approved = squadPRs.filter(pr => pr.reviewDecision === 'APPROVED' && !pr.isDraft); + const ciFailures = squadPRs.filter(pr => + pr.statusCheckRollup?.some(check => check.state === 'FAILURE' || check.state === 'ERROR') + ); + const readyToMerge = approved.filter(pr => + !pr.statusCheckRollup?.some(c => c.state === 'FAILURE' || c.state === 'ERROR' || c.state === 'PENDING') + ); + const changesRequestedSet = new Set(changesRequested.map(pr => pr.number)); + const ciFailureSet = new Set(ciFailures.map(pr => pr.number)); + const readyToMergeSet = new Set(readyToMerge.map(pr => pr.number)); + const needsReview = squadPRs.filter(pr => + !pr.isDraft && + !changesRequestedSet.has(pr.number) && + !ciFailureSet.has(pr.number) && + !readyToMergeSet.has(pr.number) + ); + + const memberNames = new Set(roster.map(m => m.name.toLowerCase())); + const timestamp = new Date().toLocaleTimeString(); + + if (drafts.length > 0) { + console.log(`${DIM}[${timestamp}]${RESET} \u{1F7E1} ${drafts.length} draft PR(s) in progress`); + for (const pr of drafts) { + console.log(` ${DIM}PR #${pr.number}: ${pr.title} (${pr.author.login})${RESET}`); + } + } + if (changesRequested.length > 0) { + console.log(`${YELLOW}[${timestamp}]${RESET} \u26A0\uFE0F ${changesRequested.length} PR(s) need revision`); + for (const pr of changesRequested) { + const owner = memberNames.has(pr.author.login.toLowerCase()) ? ` \u2014 ${pr.author.login}` : ''; + console.log(` PR #${pr.number}: ${pr.title} \u2014 changes requested${owner}`); + } + } + if (ciFailures.length > 0) { + console.log(`${RED}[${timestamp}]${RESET} \u274C ${ciFailures.length} PR(s) with CI failures`); + for (const pr of ciFailures) { + const failedChecks = pr.statusCheckRollup?.filter(c => c.state === 'FAILURE' || c.state === 'ERROR') || []; + const owner = memberNames.has(pr.author.login.toLowerCase()) ? ` \u2014 ${pr.author.login}` : ''; + console.log(` PR #${pr.number}: ${pr.title}${owner} \u2014 ${failedChecks.map(c => c.name).join(', ')}`); + } + } + if (readyToMerge.length > 0) { + console.log(`${GREEN}[${timestamp}]${RESET} \u{1F7E2} ${readyToMerge.length} PR(s) ready to merge`); + for (const pr of readyToMerge) { + console.log(` PR #${pr.number}: ${pr.title} \u2014 approved, CI green`); + } + } + + return { + drafts: drafts.length, + needsReview: needsReview.length, + changesRequested: changesRequestedSet.size, + ciFailures: ciFailureSet.size, + readyToMerge: readyToMergeSet.size, + totalOpen: squadPRs.length, + }; +} + +/** + * Run a single check cycle. + * Throws rate limit errors so the caller can update the circuit breaker. + */ +async function runCheck( + rules: ReturnType, + modules: ReturnType, + roster: ReturnType, + hasCopilot: boolean, + autoAssign: boolean +): Promise { + const timestamp = new Date().toLocaleTimeString(); + + // Fetch open issues with squad label (may throw on rate limit) + const issues = await ghIssueList({ label: 'squad', state: 'open', limit: 20 }); + + const memberLabels = roster.map(m => m.label); + const untriaged = issues.filter(issue => { + const issueLabels = issue.labels.map(l => l.name); + return !memberLabels.some(ml => issueLabels.includes(ml)); + }); + const assignedIssues = issues.filter(issue => { + const issueLabels = issue.labels.map(l => l.name); + return memberLabels.some(ml => issueLabels.includes(ml)); + }); + + let unassignedCopilot: GhIssue[] = []; + if (hasCopilot && autoAssign) { + try { + const copilotIssues = await ghIssueList({ label: 'squad:copilot', state: 'open', limit: 10 }); + unassignedCopilot = copilotIssues.filter(i => !i.assignees || i.assignees.length === 0); + } catch { + // Label may not exist yet + } + } + + for (const issue of untriaged) { + const triageInput: TriageIssue = { + number: issue.number, + title: issue.title, + body: issue.body, + labels: issue.labels.map((l) => l.name), + }; + const triage = triageIssue(triageInput, rules, modules, roster); + + if (triage) { + try { + await ghIssueEdit(issue.number, { addLabel: triage.agent.label }); + console.log( + `${GREEN}\u2713${RESET} [${timestamp}] Triaged #${issue.number} "${issue.title}" \u2192 ${triage.agent.name} (${triage.reason})` + ); + } catch (e) { + const err = e as Error; + if (isRateLimitError(err)) throw err; // bubble up for circuit breaker + console.error(`${RED}\u2717${RESET} [${timestamp}] Failed to label #${issue.number}: ${err.message}`); + } + } + } + + for (const issue of unassignedCopilot) { + try { + await ghIssueEdit(issue.number, { addAssignee: 'copilot-swe-agent' }); + console.log(`${GREEN}\u2713${RESET} [${timestamp}] Assigned @copilot to #${issue.number} "${issue.title}"`); + } catch (e) { + const err = e as Error; + if (isRateLimitError(err)) throw err; + console.error(`${RED}\u2717${RESET} [${timestamp}] Failed to assign @copilot to #${issue.number}: ${err.message}`); + } + } + + const prState = await checkPRs(roster); + + return { + untriaged: untriaged.length, + assigned: assignedIssues.length, + ...prState, + }; +} + +/** + * Execute a rate-limit-aware watch round. + * Checks API quota before calling runCheck. On 429, opens the circuit + * breaker and backs off. On recovery, closes it and resumes. + */ +async function executeRound( + round: number, + pcb: PredictiveCircuitBreaker, + cbState: CircuitBreakerState, + squadDir: string, + rules: ReturnType, + modules: ReturnType, + roster: ReturnType, + hasCopilot: boolean, + autoAssign: boolean, +): Promise<{ boardState: BoardState; light: TrafficLight }> { + const timestamp = new Date().toLocaleTimeString(); + + // 1. Check rate limit before doing work + let light: TrafficLight = 'green'; + try { + const rl = await ghRateLimitCheck(); + pcb.addSample(rl.remaining, rl.limit); + light = getTrafficLight(rl.remaining, rl.limit); + + if (light !== 'green') { + console.log( + `${YELLOW}[${timestamp}]${RESET} Rate limit: ${rl.remaining}/${rl.limit} remaining ` + + `(resets ${rl.resetAt})` + ); + } + } catch { + // gh api rate_limit failed — proceed cautiously + } + + // 2. Check circuit breaker state + if (cbState.state === 'open') { + const hitTime = cbState.lastRateLimitHit ? new Date(cbState.lastRateLimitHit).getTime() : 0; + const elapsed = (Date.now() - hitTime) / 1000 / 60; + if (elapsed < cbState.cooldownMinutes) { + const remaining = Math.ceil(cbState.cooldownMinutes - elapsed); + console.log( + `${RED}[${timestamp}]${RESET} \u{26A1} Circuit OPEN \u2014 ` + + `backing off (${remaining}min cooldown remaining)` + ); + return { boardState: emptyBoardState(), light: 'red' }; + } + // Cooldown expired — try half-open + cbState.state = 'half-open'; + console.log(`${YELLOW}[${timestamp}]${RESET} \u{1F50C} Circuit HALF-OPEN \u2014 testing API...`); + } + + // 3. Predictive check — open before hitting 429 + if (pcb.shouldOpen() && cbState.state === 'closed') { + const eta = pcb.predictExhaustion(); + console.log( + `${YELLOW}[${timestamp}]${RESET} \u{26A0}\uFE0F Predictive circuit breaker: ` + + `quota exhaustion in ~${Math.round(eta ?? 0)}s \u2014 opening circuit` + ); + cbState.state = 'open'; + cbState.lastRateLimitHit = new Date().toISOString(); + cbState.backoffRound++; + await saveCBState(squadDir, cbState); + return { boardState: emptyBoardState(), light: 'red' }; + } + + // 4. If traffic light is red and we're not in half-open test, skip + if (light === 'red' && cbState.state !== 'half-open') { + console.log(`${RED}[${timestamp}]${RESET} \u{1F6D1} Traffic light RED \u2014 skipping round`); + return { boardState: emptyBoardState(), light }; + } + + // 5. Execute the actual check + try { + const boardState = await runCheck(rules, modules, roster, hasCopilot, autoAssign); + + // Success — update circuit breaker + if (cbState.state === 'half-open') { + cbState.consecutiveSuccesses++; + if (cbState.consecutiveSuccesses >= 2) { + cbState.state = 'closed'; + cbState.consecutiveFailures = 0; + cbState.consecutiveSuccesses = 0; + cbState.backoffRound = 0; + console.log(`${GREEN}[${timestamp}]${RESET} \u{2705} Circuit CLOSED \u2014 API recovered`); + } + } else { + cbState.consecutiveSuccesses++; + cbState.consecutiveFailures = 0; + } + await saveCBState(squadDir, cbState); + + return { boardState, light }; + } catch (e) { + const err = e as Error; + if (isRateLimitError(err)) { + // Rate limited — open circuit breaker + cbState.state = 'open'; + cbState.consecutiveFailures++; + cbState.consecutiveSuccesses = 0; + cbState.lastRateLimitHit = new Date().toISOString(); + cbState.backoffRound++; + // Exponential cooldown: 10min, 20min, 40min (capped at 60) + cbState.cooldownMinutes = Math.min(10 * Math.pow(2, cbState.backoffRound - 1), 60); + await saveCBState(squadDir, cbState); + + const delay = getRetryDelay(2, cbState.consecutiveFailures); + console.error( + `${RED}[${timestamp}]${RESET} \u{26A1} Rate limited! Circuit OPEN \u2014 ` + + `cooldown ${cbState.cooldownMinutes}min (failure #${cbState.consecutiveFailures})` + ); + return { boardState: emptyBoardState(), light: 'red' }; + } + + // Non-rate-limit error — log and continue + console.error(`${RED}\u2717${RESET} [${timestamp}] Check failed: ${err.message}`); + return { boardState: emptyBoardState(), light }; + } +} + +/** + * Run watch command — Ralph's local polling process + */ +export async function runWatch(dest: string, intervalMinutes: number): Promise { + if (isNaN(intervalMinutes) || intervalMinutes < 1) { + fatal('--interval must be a positive number of minutes'); + } + + const squadDirInfo = detectSquadDir(dest); + const teamMd = path.join(squadDirInfo.path, 'team.md'); + const routingMdPath = path.join(squadDirInfo.path, 'routing.md'); + + if (!fs.existsSync(teamMd)) { + fatal('No squad found \u2014 run init first.'); + } + + if (!(await ghAvailable())) { + fatal('gh CLI not found \u2014 install from https://cli.github.com'); + } + + if (!(await ghAuthenticated())) { + console.error(`${YELLOW}\u26A0\uFE0F${RESET} gh CLI not authenticated`); + console.error(` Run: ${BOLD}gh auth login${RESET}\n`); + fatal('gh authentication required'); + } + + const content = fs.readFileSync(teamMd, 'utf8'); + const roster = parseRoster(content); + const routingContent = fs.existsSync(routingMdPath) ? fs.readFileSync(routingMdPath, 'utf8') : ''; + const rules = parseRoutingRules(routingContent); + const modules = parseModuleOwnership(routingContent); + + if (roster.length === 0) { + fatal('No squad members found in team.md'); + } + + const hasCopilot = content.includes('\u{1F916} Coding Agent') || content.includes('@copilot'); + const autoAssign = content.includes(''); + const monitorSessionId = 'ralph-watch'; + const eventBus = new EventBus(); + const monitor = new RalphMonitor({ + teamRoot: path.dirname(squadDirInfo.path), + healthCheckInterval: intervalMinutes * 60 * 1000, + staleSessionThreshold: intervalMinutes * 60 * 1000 * 3, + statePath: path.join(squadDirInfo.path, '.ralph-state.json'), + }); + await monitor.start(eventBus); + await eventBus.emit({ + type: 'session:created', + sessionId: monitorSessionId, + agentName: 'Ralph', + payload: { intervalMinutes }, + timestamp: new Date(), + }); + + // Initialize circuit breaker + const pcb = new PredictiveCircuitBreaker({ maxSamples: 10, warningThresholdSeconds: 120 }); + const cbState = loadCBState(squadDirInfo.path); + + console.log(`\n${BOLD}\u{1F504} Ralph \u2014 Watch Mode${RESET}`); + console.log(`${DIM}Polling every ${intervalMinutes} minute(s) for squad work. Ctrl+C to stop.${RESET}`); + if (cbState.state !== 'closed') { + console.log(`${YELLOW}\u26A0\uFE0F Resuming with circuit breaker ${cbState.state}${RESET}`); + } + console.log(); + + let round = 0; + + // Run immediately, then on interval + round++; + const { boardState: state, light } = await executeRound( + round, pcb, cbState, squadDirInfo.path, + rules, modules, roster, hasCopilot, autoAssign + ); + await eventBus.emit({ + type: 'agent:milestone', + sessionId: monitorSessionId, + agentName: 'Ralph', + payload: { milestone: `Completed watch round ${round}`, task: 'watch cycle' }, + timestamp: new Date(), + }); + await monitor.healthCheck(); + reportBoard(state, round, light); + + return new Promise((resolve) => { + const intervalId = setInterval( + async () => { + round++; + const { boardState: roundState, light: roundLight } = await executeRound( + round, pcb, cbState, squadDirInfo.path, + rules, modules, roster, hasCopilot, autoAssign + ); + await eventBus.emit({ + type: 'agent:milestone', + sessionId: monitorSessionId, + agentName: 'Ralph', + payload: { milestone: `Completed watch round ${round}`, task: 'watch cycle' }, + timestamp: new Date(), + }); + await monitor.healthCheck(); + reportBoard(roundState, round, roundLight); + }, + intervalMinutes * 60 * 1000 + ); + + // Graceful shutdown — persist circuit breaker state + let isShuttingDown = false; + const shutdown = async () => { + if (isShuttingDown) return; + isShuttingDown = true; + clearInterval(intervalId); + process.off('SIGINT', shutdown); + process.off('SIGTERM', shutdown); + saveCBState(squadDirInfo.path, cbState); + await eventBus.emit({ + type: 'session:destroyed', + sessionId: monitorSessionId, + agentName: 'Ralph', + payload: null, + timestamp: new Date(), + }); + await monitor.stop(); + console.log(`\n${DIM}\u{1F504} Ralph \u2014 Watch stopped (circuit: ${cbState.state})${RESET}`); + resolve(); + }; + + process.on('SIGINT', shutdown); + process.on('SIGTERM', shutdown); + }); +} \ No newline at end of file diff --git a/packages/squad-cli/src/cli/core/gh-cli.ts b/packages/squad-cli/src/cli/core/gh-cli.ts index 9f68d848e..0a1833f51 100644 --- a/packages/squad-cli/src/cli/core/gh-cli.ts +++ b/packages/squad-cli/src/cli/core/gh-cli.ts @@ -128,3 +128,33 @@ export async function ghIssueEdit(issueNumber: number, options: GhEditOptions): await execFileAsync('gh', args); } + +/** Rate limit status from GitHub API */ +export interface GhRateLimitStatus { + remaining: number; + limit: number; + resetAt: string; +} + +/** + * Check current GitHub API rate limit status. + * Returns remaining/limit so the circuit breaker can make decisions. + */ +export async function ghRateLimitCheck(): Promise { + const { stdout } = await execFileAsync('gh', [ + 'api', 'rate_limit', '--jq', + '{ remaining: .rate.remaining, limit: .rate.limit, resetAt: .rate.reset | todate }' + ]); + return JSON.parse(stdout); +} + +/** + * Detect if an error is a rate limit error (HTTP 429 or secondary rate limit). + */ +export function isRateLimitError(err: Error): boolean { + const msg = err.message.toLowerCase(); + return msg.includes('rate limit') || + msg.includes('429') || + msg.includes('secondary rate limit') || + msg.includes('api rate limit exceeded'); +} diff --git a/test/cli/watch-rate-limiting.test.ts b/test/cli/watch-rate-limiting.test.ts new file mode 100644 index 000000000..78624bebc --- /dev/null +++ b/test/cli/watch-rate-limiting.test.ts @@ -0,0 +1,76 @@ +/** + * Watch Command Rate Limiting Integration Tests + * + * Tests the circuit breaker integration in watch.ts and + * the rate limit utilities in gh-cli.ts. + */ + +import { describe, it, expect } from 'vitest'; + +describe('CLI: watch command rate limiting', () => { + it('module exports reportBoard with optional light parameter', async () => { + const mod = await import('@bradygaster/squad-cli/commands/watch'); + expect(typeof mod.reportBoard).toBe('function'); + // reportBoard should accept 2 or 3 args (state, round, light?) + expect(mod.reportBoard.length).toBeGreaterThanOrEqual(2); + }); + + it('reportBoard handles all board states without crashing', async () => { + const { reportBoard } = await import('@bradygaster/squad-cli/commands/watch'); + const state = { + untriaged: 3, + assigned: 2, + drafts: 1, + needsReview: 1, + changesRequested: 0, + ciFailures: 0, + readyToMerge: 1, + }; + // Should not throw + expect(() => reportBoard(state, 1)).not.toThrow(); + expect(() => reportBoard(state, 2, 'green')).not.toThrow(); + expect(() => reportBoard(state, 3, 'amber')).not.toThrow(); + expect(() => reportBoard(state, 4, 'red')).not.toThrow(); + }); + + it('reportBoard handles empty state', async () => { + const { reportBoard } = await import('@bradygaster/squad-cli/commands/watch'); + const empty = { + untriaged: 0, + assigned: 0, + drafts: 0, + needsReview: 0, + changesRequested: 0, + ciFailures: 0, + readyToMerge: 0, + }; + expect(() => reportBoard(empty, 1, 'green')).not.toThrow(); + }); +}); + +describe('gh-cli: rate limit utilities', () => { + it('exports isRateLimitError function', async () => { + const mod = await import('@bradygaster/squad-cli/core/gh-cli'); + expect(typeof mod.isRateLimitError).toBe('function'); + }); + + it('isRateLimitError detects 429 errors', async () => { + const { isRateLimitError } = await import('@bradygaster/squad-cli/core/gh-cli'); + expect(isRateLimitError(new Error('HTTP 429: rate limit exceeded'))).toBe(true); + expect(isRateLimitError(new Error('API rate limit exceeded'))).toBe(true); + expect(isRateLimitError(new Error('secondary rate limit hit'))).toBe(true); + expect(isRateLimitError(new Error('You have exceeded a secondary rate limit'))).toBe(true); + }); + + it('isRateLimitError rejects non-rate-limit errors', async () => { + const { isRateLimitError } = await import('@bradygaster/squad-cli/core/gh-cli'); + expect(isRateLimitError(new Error('Not found'))).toBe(false); + expect(isRateLimitError(new Error('Network timeout'))).toBe(false); + expect(isRateLimitError(new Error('Permission denied'))).toBe(false); + }); + + it('exports ghRateLimitCheck function', async () => { + const mod = await import('@bradygaster/squad-cli/core/gh-cli'); + expect(typeof mod.ghRateLimitCheck).toBe('function'); + }); +}); \ No newline at end of file From d97cd7cd71498d2c16ad6b92d7f24e938a1d8973 Mon Sep 17 00:00:00 2001 From: Copilot Date: Mon, 23 Mar 2026 00:23:55 +0200 Subject: [PATCH 4/6] fix: add core/gh-cli export for rate limit utilities Adds ./core/gh-cli subpath export to CLI package.json so the rate limit utilities (isRateLimitError, ghRateLimitCheck) are accessible from tests and external consumers. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- packages/squad-cli/package.json | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/squad-cli/package.json b/packages/squad-cli/package.json index a67125734..d865c1ffe 100644 --- a/packages/squad-cli/package.json +++ b/packages/squad-cli/package.json @@ -1,6 +1,6 @@ { "name": "@bradygaster/squad-cli", - "version": "0.8.25-build.10", + "version": "0.8.25-build.13", "description": "Squad CLI — Command-line interface for the Squad multi-agent runtime", "type": "module", "bin": { @@ -88,6 +88,10 @@ "types": "./dist/cli/core/version.d.ts", "import": "./dist/cli/core/version.js" }, + "./core/gh-cli": { + "types": "./dist/cli/core/gh-cli.d.ts", + "import": "./dist/cli/core/gh-cli.js" + }, "./commands/export": { "types": "./dist/cli/commands/export.d.ts", "import": "./dist/cli/commands/export.js" From fc63d24d851a3fa58bf83dea8adff888f04ad9fe Mon Sep 17 00:00:00 2001 From: Copilot Date: Mon, 23 Mar 2026 05:30:24 +0200 Subject: [PATCH 5/6] fix: race condition guard + remove dead code in watch command - Add roundInProgress flag to prevent overlapping setInterval rounds - Remove unused getRetryDelay import and dead variable (line 391) - Remove unused shouldProceed import Fixes Q review findings: race condition when executeRound() takes longer than interval causes double-triaging. Now skips round if previous is still running. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- packages/squad-cli/src/cli/commands/watch.ts | 41 ++++++++++++-------- 1 file changed, 24 insertions(+), 17 deletions(-) diff --git a/packages/squad-cli/src/cli/commands/watch.ts b/packages/squad-cli/src/cli/commands/watch.ts index 446d740c3..e1fabae3e 100644 --- a/packages/squad-cli/src/cli/commands/watch.ts +++ b/packages/squad-cli/src/cli/commands/watch.ts @@ -37,8 +37,6 @@ import { import { PredictiveCircuitBreaker, getTrafficLight, - shouldProceed, - getRetryDelay, type TrafficLight, } from '@bradygaster/squad-sdk/ralph/rate-limiting'; @@ -388,7 +386,6 @@ async function executeRound( cbState.cooldownMinutes = Math.min(10 * Math.pow(2, cbState.backoffRound - 1), 60); await saveCBState(squadDir, cbState); - const delay = getRetryDelay(2, cbState.consecutiveFailures); console.error( `${RED}[${timestamp}]${RESET} \u{26A1} Rate limited! Circuit OPEN \u2014 ` + `cooldown ${cbState.cooldownMinutes}min (failure #${cbState.consecutiveFailures})` @@ -487,22 +484,32 @@ export async function runWatch(dest: string, intervalMinutes: number): Promise((resolve) => { + let roundInProgress = false; const intervalId = setInterval( async () => { - round++; - const { boardState: roundState, light: roundLight } = await executeRound( - round, pcb, cbState, squadDirInfo.path, - rules, modules, roster, hasCopilot, autoAssign - ); - await eventBus.emit({ - type: 'agent:milestone', - sessionId: monitorSessionId, - agentName: 'Ralph', - payload: { milestone: `Completed watch round ${round}`, task: 'watch cycle' }, - timestamp: new Date(), - }); - await monitor.healthCheck(); - reportBoard(roundState, round, roundLight); + if (roundInProgress) { + console.log(`${DIM}[${new Date().toLocaleTimeString()}] Previous round still running — skipping${RESET}`); + return; + } + roundInProgress = true; + try { + round++; + const { boardState: roundState, light: roundLight } = await executeRound( + round, pcb, cbState, squadDirInfo.path, + rules, modules, roster, hasCopilot, autoAssign + ); + await eventBus.emit({ + type: 'agent:milestone', + sessionId: monitorSessionId, + agentName: 'Ralph', + payload: { milestone: `Completed watch round ${round}`, task: 'watch cycle' }, + timestamp: new Date(), + }); + await monitor.healthCheck(); + reportBoard(roundState, round, roundLight); + } finally { + roundInProgress = false; + } }, intervalMinutes * 60 * 1000 ); From a560bf8230e117ab292a4dfee9563111422fff3d Mon Sep 17 00:00:00 2001 From: Tamir Dresher Date: Mon, 23 Mar 2026 05:57:52 +0200 Subject: [PATCH 6/6] test: add 5 more tests for rate limiting & watch command - roundInProgress prevents overlapping concurrent rounds - roundInProgress resets to false after a round throws (finally block) - Rate limit amber traffic light blocks P1/P2 agents - Rate limit red traffic light blocks all agents - PredictiveCircuitBreaker stays closed with no samples Fixes Q review finding: only 7 tests for 534 lines of new code --- test/cli/watch-rate-limiting.test.ts | 82 ++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/test/cli/watch-rate-limiting.test.ts b/test/cli/watch-rate-limiting.test.ts index 78624bebc..bdad9e7a1 100644 --- a/test/cli/watch-rate-limiting.test.ts +++ b/test/cli/watch-rate-limiting.test.ts @@ -46,6 +46,88 @@ describe('CLI: watch command rate limiting', () => { }; expect(() => reportBoard(empty, 1, 'green')).not.toThrow(); }); + + it('roundInProgress flag prevents overlapping rounds', async () => { + // Simulate the guard logic used in runWatch's setInterval callback. + // Verifies that a second "tick" while the first is still in flight is + // a no-op and does not increment the counter a second time. + let callCount = 0; + let roundInProgress = false; + + const tick = async () => { + if (roundInProgress) return; + roundInProgress = true; + try { + callCount++; + // Simulate slow async work + await new Promise(r => setTimeout(r, 20)); + } finally { + roundInProgress = false; + } + }; + + // Fire two ticks concurrently — only the first should execute the body + await Promise.all([tick(), tick()]); + expect(callCount).toBe(1); + }); + + it('roundInProgress resets to false after a round throws', async () => { + // Verifies the finally block in the setInterval callback properly + // releases the lock even when executeRound throws. + let roundInProgress = false; + + const tick = async () => { + if (roundInProgress) return; + roundInProgress = true; + try { + throw new Error('Simulated round failure'); + } finally { + roundInProgress = false; + } + }; + + await expect(tick()).rejects.toThrow('Simulated round failure'); + // Lock must be released so subsequent rounds can run + expect(roundInProgress).toBe(false); + + // Confirm next round proceeds normally + let nextRanCount = 0; + const tick2 = async () => { + if (roundInProgress) return; + roundInProgress = true; + try { nextRanCount++; } finally { roundInProgress = false; } + }; + await tick2(); + expect(nextRanCount).toBe(1); + }); + + it('rate limit interaction: getTrafficLight + shouldProceed block lower-priority agents on amber', async () => { + const { getTrafficLight, shouldProceed } = await import('@bradygaster/squad-sdk/ralph/rate-limiting'); + // 10% remaining → amber + const light = getTrafficLight(100, 1000); + expect(light).toBe('amber'); + // P0 (Lead) allowed, P1+ blocked + expect(shouldProceed(light, 0)).toBe(true); + expect(shouldProceed(light, 1)).toBe(false); + expect(shouldProceed(light, 2)).toBe(false); + }); + + it('rate limit interaction: getTrafficLight returns red at 0 remaining', async () => { + const { getTrafficLight, shouldProceed } = await import('@bradygaster/squad-sdk/ralph/rate-limiting'); + const light = getTrafficLight(0, 5000); + expect(light).toBe('red'); + // All agents blocked + expect(shouldProceed(light, 0)).toBe(false); + expect(shouldProceed(light, 1)).toBe(false); + expect(shouldProceed(light, 2)).toBe(false); + }); + + it('PredictiveCircuitBreaker stays closed with no samples', async () => { + const { PredictiveCircuitBreaker } = await import('@bradygaster/squad-sdk/ralph/rate-limiting'); + const pcb = new PredictiveCircuitBreaker({ maxSamples: 5, warningThresholdSeconds: 120 }); + expect(pcb.shouldOpen()).toBe(false); + expect(pcb.predictExhaustion()).toBeNull(); + }); }); describe('gh-cli: rate limit utilities', () => {