From 0faadad383cb799c972549c99a76baec03e360f8 Mon Sep 17 00:00:00 2001 From: WEB3NOVA Date: Tue, 1 Sep 2026 10:03:09 +0100 Subject: [PATCH 1/2] feat(agent): add circuit breaker (#345) Add an agent circuit breaker that halts agent-initiated rebalancing when the market, a protocol, or a user shows risk. Withdrawals are untouched. - Pure trip rules (breakerRules.ts): abnormal-loss drawdown, stablecoin de-peg (fails safe on a missing feed), rebalance oscillation, stale APY. - Pure CLOSED/OPEN/HALF_OPEN state machine (breakerState.ts) with cooldown backoff and manual trip/reset. - Integration (breakerService.ts): measurement loads, per-tick evaluation, fail-closed halting, operator alerts, user events, status surfaces. - Loop wiring: GLOBAL > PROTOCOL > USER blocking, BLOCKED decisions, blockedProtocols target exclusion, agent status + per-user status route. - Admin endpoints: list/trip/reset breakers with audit logging. - Prometheus gauge + counters and alert rules; RUNBOOK + OBSERVABILITY docs. - Tests: 5 new suites (rules, state machine, service status/privacy, loop blocking, target exclusion) + harness update. De-peg is a pure rule consuming a price input; no live feed exists yet so getStablecoinPrice() returns null (fails safe) and the rule stays off by default (BREAKER_DEPEG_ENABLED=false). --- deploy/monitoring/prometheus/alert-rules.yaml | 18 + docs/OBSERVABILITY.md | 20 + docs/RUNBOOK.md | 82 ++ .../migration.sql | 27 + .../rollback.sql | 13 + prisma/schema.prisma | 31 + src/agent/breakerRules.ts | 345 ++++++ src/agent/breakerService.ts | 1030 +++++++++++++++++ src/agent/breakerState.ts | 259 +++++ src/agent/loop.ts | 122 ++ src/agent/router.ts | 31 +- src/config/env.ts | 62 + src/events/types.ts | 6 + src/routes/admin.ts | 151 +++ src/routes/agent.ts | 32 + src/utils/metrics.ts | 35 + .../agent/breakerLoop.integration.test.ts | 252 ++++ .../agent/strategy-follow.integration.test.ts | 12 + tests/unit/agent/breakerRules.test.ts | 371 ++++++ tests/unit/agent/breakerServiceStatus.test.ts | 148 +++ tests/unit/agent/breakerState.test.ts | 252 ++++ 21 files changed, 3294 insertions(+), 5 deletions(-) create mode 100644 prisma/migrations/20260901000000_add_agent_circuit_breaker/migration.sql create mode 100644 prisma/migrations/20260901000000_add_agent_circuit_breaker/rollback.sql create mode 100644 src/agent/breakerRules.ts create mode 100644 src/agent/breakerService.ts create mode 100644 src/agent/breakerState.ts create mode 100644 tests/integration/agent/breakerLoop.integration.test.ts create mode 100644 tests/unit/agent/breakerRules.test.ts create mode 100644 tests/unit/agent/breakerServiceStatus.test.ts create mode 100644 tests/unit/agent/breakerState.test.ts diff --git a/deploy/monitoring/prometheus/alert-rules.yaml b/deploy/monitoring/prometheus/alert-rules.yaml index a48f60d..363aca2 100644 --- a/deploy/monitoring/prometheus/alert-rules.yaml +++ b/deploy/monitoring/prometheus/alert-rules.yaml @@ -74,6 +74,15 @@ groups: summary: "Severe network congestion" description: "Congestion level severe for 5 minutes — LOW ops deferring" + - alert: AgentBreakerGlobalOpen + expr: agent_breaker_state{scope="GLOBAL"} == 2 + for: 1m + labels: + severity: critical + annotations: + summary: "Agent circuit breaker is OPEN globally" + description: "All agent rebalancing is halted" + - name: neurowealth_warning interval: 30s rules: @@ -139,3 +148,12 @@ groups: annotations: summary: "HTTP requests slow" description: "P95 HTTP request duration is {{ $value }}s (> 5s)" + + - alert: AgentBreakerOpen + expr: agent_breaker_state{scope!="GLOBAL"} == 2 + for: 2m + labels: + severity: warning + annotations: + summary: "Agent circuit breaker OPEN for a protocol or user" + description: "{{ $labels.scope }} {{ $labels.scopeKey }} is halted (state={{ $value }})" diff --git a/docs/OBSERVABILITY.md b/docs/OBSERVABILITY.md index 68cd077..b32d0aa 100644 --- a/docs/OBSERVABILITY.md +++ b/docs/OBSERVABILITY.md @@ -48,6 +48,24 @@ The backend exposes Prometheus-compatible metrics through the `/metrics` endpoin - `agent_rebalances_triggered_total` - Counter - `agent_snapshot_duration_seconds` - Histogram +### Agent Circuit Breaker Metrics (#345) + +- `agent_breaker_state` - Gauge, labels `scope` (`GLOBAL`|`PROTOCOL`|`USER`) and `scopeKey`; 0 = CLOSED, 1 = HALF_OPEN, 2 = OPEN. Written every agent breaker evaluation tick and on every transition. +- `agent_breaker_trips_total` - Counter, labels `scope`, `rule` (`abnormal_loss`|`depeg`|`oscillation`|`stale_data`|`manual`). Increments whenever a breaker trips or re-trips. + +Query examples: + +``` +# Any circuit breaker open right now +agent_breaker_state == 2 + +# Global halt (stops all agent rebalancing) +agent_breaker_state{scope="GLOBAL"} == 2 + +# Breaker trip rate by rule +sum(rate(agent_breaker_trips_total[15m])) by (rule) +``` + ### Database Metrics - `db_operation_duration_seconds` - Histogram with label: `operation` @@ -74,6 +92,7 @@ The backend exposes Prometheus-compatible metrics through the `/metrics` endpoin | `cursor_lag_ledgers` | `> 100` | Critical | Event processing lagging significantly | | `dlq_size` | `> 50` | Critical | Dead Letter Queue critically large | | `failures_total` (rate) | `> 10 per minute` for 5m | Critical | High failure rate | +| `agent_breaker_state{scope="GLOBAL"}` | `== 2` for 1m | Critical | Global circuit breaker OPEN — all rebalancing halted | ### Warning Alerts (Investigate Within 1 Hour) @@ -85,6 +104,7 @@ The backend exposes Prometheus-compatible metrics through the `/metrics` endpoin | `events_processing_duration_seconds` (p95) | `> 2 seconds` | Warning | Event processing slow | | `db_operation_duration_seconds` (p95) | `> 1 second` | Warning | Database operations slow | | `http_request_duration_seconds` (p95) | `> 5 seconds` | Warning | HTTP requests slow | +| `agent_breaker_state{scope!="GLOBAL"}` | `== 2` for 2m | Warning | Protocol or user breaker OPEN — affected rebalancing halted | ### Info Alerts (Monitor Trend) diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index 5d541a2..ea86430 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -482,3 +482,85 @@ psql "$DATABASE_URL" -c "SELECT \"sponsorAccount\", count(*), sum(\"xlmReserved\ ``` No auto top-up — operational runbook only. Reconciliation job (`reserveReconciliation` hourly) flags drift where on-chain sponsor ≠ ledger. + +## 9. Agent Circuit Breaker (#345) + +The agent circuit breaker halts agent-initiated rebalancing when the market, +a protocol, or a user's account shows risk. It never touches withdrawals. +Scopes: `GLOBAL` (halt everything), `PROTOCOL` (halt a target protocol + +batches leaving it), `USER` (halt that user's batches). Breakers are +`CLOSED → OPEN → HALF_OPEN → CLOSED`; an `OPEN` breaker auto-probes after its +cooldown and needs `BREAKER_DEPEG_SUSTAINED_CHECKS` clean evaluations before +recovering to `HALF_OPEN`, then one clean probe to close. + +### Rules + +| Rule | Env | Default | Trips when | +|---|---|---|---| +| abnormal_loss | `BREAKER_LOSS_PCT`, `BREAKER_LOSS_WINDOW_HOURS` | 5% / 24h | mark-to-market drawdown over the window exceeds the pct | +| depeg | `BREAKER_DEPEG_ENABLED` (=`false`) | off | reported stablecoin price deviates > `BREAKER_DEPEG_BPS` (150) from $1 | +| oscillation | `BREAKER_MAX_FLIPS`, `BREAKER_FLIP_WINDOW_HOURS` | 3 / 24h | same batch rebalances ≥ N times in the window | +| stale_data | `BREAKER_STALE_MINUTES`, `BREAKER_STALE_CONSECUTIVE_FAILURES` | 120m / 3 | APY table older than limit, never scanned, or ≥ N consecutive failures | + +`BREAKER_COOLDOWN_MS` (1h) is the base cooldown; a repeated HALF_OPEN trip +doubles it up to `BREAKER_MAX_COOLDOWN_MS` (24h). + +### Known limitation — de-peg price feed + +The de-peg rule is a pure consumer of a stablecoin spot price. As of this +change no live price feed exists in this codebase: the fee oracle +(`src/stellar/feeOracle.ts`) publishes only fees, and `src/stellar/routing.ts` +is a stub. `getStablecoinPrice()` (`src/agent/breakerService.ts`) is the single +integration point and currently returns `null` (fails safe — the rule never +trips); the rule is disabled by default. Wire the oracle there, keep the pure +rule unchanged, flip `BREAKER_DEPEG_ENABLED=true`, and re-run the de-peg unit +tests. + +### Inspector + +```bash +# All breakers with current state +curl -H "Authorization: Bearer $ADMIN_API_TOKEN" http://localhost:3001/api/v1/admin/agent/breakers | jq + +# Agent status (incl. cached global breaker summary) +curl -H "X-Internal-Token: $INTERNAL_SERVICE_TOKEN" http://localhost:3001/api/v1/agent/status | jq +``` + +### Manual trip + +`POST /api/v1/admin/agent/breakers` — body `{"scope":"PROTOCOL","scopeKey":"blend","reason":"..."}`. +`GLOBAL` needs no `scopeKey`; `USER`/`PROTOCOL` require it. `reason` is always +required. Written to the admin audit log (`TRIP_AGENT_BREAKER`). + +```bash +curl -X POST http://localhost:3001/api/v1/admin/agent/breakers \ + -H "Authorization: Bearer $ADMIN_API_TOKEN" -H "Content-Type: application/json" \ + -d '{"scope":"PROTOCOL","scopeKey":"blend","reason":"incident-4821 protocol outage"}' +``` + +A manual trip can only be cleared manually (`rule=manual` breakers never +auto-reset). Manual resets are audit-logged too: + +```bash +curl -X POST http://localhost:3001/api/v1/admin/agent/breakers//reset \ + -H "Authorization: Bearer $ADMIN_API_TOKEN" -H "Content-Type: application/json" \ + -d '{"reason":"incident-4821 resolved, APY table verified fresh"}' +``` + +A breaker skips its tick when evaluation itself fails: the agent alerts +(critical, `agent-breaker:eval-failed`) and halts **all** rebalancing for that +tick rather than trading blind. + +### Response + +1. **Global halt**: investigate the trip rule (`agent_breaker_trips_total{scope="GLOBAL"}`), check the `[Breaker]` logs for the `lastEvaluation` detail, fix root cause, then either wait for auto-recovery or reset manually. +2. **Protocol halt**: verify the protocol's APY/status independently before resetting; `compareProtocols` already refuses it as a target while OPEN. +3. **User halt**: confirm with the user before resetting. +4. **Evaluation-failed halt**: the breaker could not decide — check DB connectivity and the logged error before the next tick. + +### Verify + +```bash +curl -s http://localhost:3001/metrics | grep -E "agent_breaker_(state|trips_total)" +psql "$DATABASE_URL" -c "SELECT scope, \"scopeKey\", state, \"trippedRule\" FROM agent_circuit_breakers ORDER BY \"updatedAt\" DESC;" +``` diff --git a/prisma/migrations/20260901000000_add_agent_circuit_breaker/migration.sql b/prisma/migrations/20260901000000_add_agent_circuit_breaker/migration.sql new file mode 100644 index 0000000..ad7a260 --- /dev/null +++ b/prisma/migrations/20260901000000_add_agent_circuit_breaker/migration.sql @@ -0,0 +1,27 @@ +-- Migration: add_agent_circuit_breaker (#345) +-- Agent circuit breaker: a pre-trade kill switch that halts agent-initiated +-- rebalancing at GLOBAL / PROTOCOL / USER scope when abnormal loss, de-peg, +-- oscillation or stale data is detected. Requires an explicit, audited reset. +-- User-initiated withdrawals are never blocked by these rows (enforced in code). + +CREATE TYPE "BreakerScope" AS ENUM ('GLOBAL', 'PROTOCOL', 'USER'); +CREATE TYPE "BreakerState" AS ENUM ('CLOSED', 'OPEN', 'HALF_OPEN'); + +CREATE TABLE "agent_circuit_breakers" ( + "id" TEXT NOT NULL, + "scope" "BreakerScope" NOT NULL, + "scopeKey" TEXT NOT NULL, -- "" for GLOBAL, protocolName, or userId + "state" "BreakerState" NOT NULL DEFAULT 'CLOSED', + "trippedRule" TEXT, -- abnormal_loss | depeg | oscillation | stale_data | manual + "trippedAt" TIMESTAMP(3), + "detail" JSONB, + "resetBy" TEXT, -- admin identity on manual reset + "resetAt" TIMESTAMP(3), + "autoResetAt" TIMESTAMP(3), -- earliest time HALF_OPEN is allowed + "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "agent_circuit_breakers_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "agent_circuit_breakers_scope_scopeKey_key" ON "agent_circuit_breakers"("scope", "scopeKey"); +CREATE INDEX "agent_circuit_breakers_state_idx" ON "agent_circuit_breakers"("state"); diff --git a/prisma/migrations/20260901000000_add_agent_circuit_breaker/rollback.sql b/prisma/migrations/20260901000000_add_agent_circuit_breaker/rollback.sql new file mode 100644 index 0000000..4ae0781 --- /dev/null +++ b/prisma/migrations/20260901000000_add_agent_circuit_breaker/rollback.sql @@ -0,0 +1,13 @@ +-- rollback.sql — reverse of 20260901000000_add_agent_circuit_breaker/migration.sql +-- Drops the agent circuit breaker table and its enum types (#345). +-- WARNING: DATA LOSS — all breaker state (OPEN/HALF_OPEN), trip details and +-- admin reset audit rows are lost. Revert app code BEFORE running. +-- Indexes are dropped with the table (explicit drops for idempotency). +-- Safe to run multiple times. +-- Run with: psql $DATABASE_URL -f prisma/migrations/20260901000000_add_agent_circuit_breaker/rollback.sql + +DROP INDEX IF EXISTS "agent_circuit_breakers_state_idx"; +DROP INDEX IF EXISTS "agent_circuit_breakers_scope_scopeKey_key"; +DROP TABLE IF EXISTS "agent_circuit_breakers" CASCADE; +DROP TYPE IF EXISTS "BreakerState"; +DROP TYPE IF EXISTS "BreakerScope"; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index edfa1d0..bbb8e64 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1906,3 +1906,34 @@ model ProtocolLiquiditySnapshot { @@index([fetchedAt]) @@map("protocol_liquidity_snapshots") } + +// --- Issue #345: Agent Circuit Breaker --- +enum BreakerScope { + GLOBAL + PROTOCOL + USER +} + +enum BreakerState { + CLOSED + OPEN + HALF_OPEN +} + +model AgentCircuitBreaker { + id String @id @default(uuid()) + scope BreakerScope + scopeKey String // "" for GLOBAL, protocolName, or userId + state BreakerState @default(CLOSED) + trippedRule String? // "abnormal_loss" | "depeg" | "oscillation" | "stale_data" | "manual" + trippedAt DateTime? + detail Json? // the measurements that tripped it + resetBy String? // admin identity on manual reset + resetAt DateTime? + autoResetAt DateTime? // earliest time HALF_OPEN is allowed + updatedAt DateTime @updatedAt + + @@unique([scope, scopeKey]) + @@index([state]) + @@map("agent_circuit_breakers") +} diff --git a/src/agent/breakerRules.ts b/src/agent/breakerRules.ts new file mode 100644 index 0000000..6a36f79 --- /dev/null +++ b/src/agent/breakerRules.ts @@ -0,0 +1,345 @@ +/** + * src/agent/breakerRules.ts + * + * Pure trip rules for the agent circuit breaker (#345). Each function is + * deterministic: it takes measured inputs and configuration numbers, never + * touches the database, the clock, or the environment. The integration layer + * in the agent loop supplies the measurements; these rules only decide. + */ + +export type BreakerTripReason = + 'abnormal_loss' | 'depeg' | 'oscillation' | 'stale_data' | 'manual' + +export interface RuleResult { + tripped: boolean + rule: BreakerTripReason + detail: Record +} + +/** A mark-to-market value point for a position or portfolio series. */ +export interface ValuePoint { + at: Date + value: number +} + +interface AbnormalLossInput { + /** Mark-to-market series, newest-first or oldest-first (sorted internally). */ + series: ValuePoint[] + lossPct: number + windowHours: number + now: Date +} + +interface DepegInput { + /** Current USD price of the stablecoin (1 === $1). null = no feed = fail-safe. */ + price: number | null + depegBps: number +} + +interface OscillationInput { + /** Rebalance count for the same batchKey within the flip window. */ + flips: number + maxFlips: number +} + +interface StaleDataInput { + /** Timestamp of the latest successful APY scan. null = never scanned. */ + latestFetchedAt: Date | null + maxStaleMinutes: number + maxConsecutiveFailures: number + consecutiveFailures: number + now: Date +} + +/** + * Aggregate mark-to-market drawdown rule. + * + * Drawdown is measured as (currentValue - peakInWindow) / peakInWindow using + * the raw point series — the same period-return convention the portfolio-risk + * analytics stack uses, not a smoothed cumulative column. A short or empty + * series cannot trip (need at least two points to establish a drawdown). + * + * Trips when the window drawdown is worse than `lossPct` (e.g. down >5%). + */ +export function evaluateAbnormalLossRule(input: AbnormalLossInput): RuleResult { + const { series, lossPct, windowHours, now } = input + + const windowStartMs = now.getTime() - windowHours * 60 * 60 * 1000 + const inWindow = series + .filter((p) => p.at.getTime() >= windowStartMs) + .slice() + .sort((a, b) => a.at.getTime() - b.at.getTime()) + + const detail: Record = { lossPct, windowHours } + + if (inWindow.length < 2) { + return { + tripped: false, + rule: 'abnormal_loss', + detail: { + ...detail, + reason: 'insufficient_history', + points: inWindow.length, + }, + } + } + + const current = inWindow[inWindow.length - 1].value + let peak = Number.NEGATIVE_INFINITY + for (const p of inWindow) { + if (p.value > peak) peak = p.value + } + + if (peak <= 0 || current <= 0) { + return { + tripped: false, + rule: 'abnormal_loss', + detail: { ...detail, reason: 'non_positive_peak', peak, current }, + } + } + + const drawdownPct = ((current - peak) / peak) * 100 + const thresholdPct = -Math.abs(lossPct) + const tripped = drawdownPct <= thresholdPct + + return { + tripped, + rule: 'abnormal_loss', + detail: { + ...detail, + current: round4(current), + peak: round4(peak), + drawdownPct: round4(drawdownPct), + thresholdPct, + }, + } +} + +/** + * Stablecoin de-peg rule. + * + * Trips when the reported USD price deviates more than `depegBps` from $1. + * A null price (no feed) never trips — the rule fails safe rather than + * halting rebalancing on missing data; missing *fresh* data is + * stale_data's job. + */ +export function evaluateDepegRule(input: DepegInput): RuleResult { + const { price, depegBps } = input + + if (price === null || !Number.isFinite(price) || price <= 0) { + return { + tripped: false, + rule: 'depeg', + detail: { depegBps, reason: 'no_price_feed' }, + } + } + + const deviationBps = Math.abs(price - 1) * 10000 + const tripped = deviationBps > depegBps + + return { + tripped, + rule: 'depeg', + detail: { + depegBps, + price: round4(price), + deviationBps: round4(deviationBps), + }, + } +} + +/** + * Rebalance oscillation rule. + * + * Trips when the same batchKey has rebalanced at least `maxFlips` times + * within the flip window. Detects A→B→A→B fee-burning. Each counted flip + * individually passed the net-improvement gate, so this is a fee-protection + * heuristic, not a correctness claim about any single flip. + */ +export function evaluateOscillationRule(input: OscillationInput): RuleResult { + const { flips, maxFlips } = input + const tripped = flips >= maxFlips + + return { + tripped, + rule: 'oscillation', + detail: { flips, maxFlips }, + } +} + +/** + * Stale data rule. + * + * Trips when the APY table is older than `maxStaleMinutes`, or when scanning + * has failed at least `maxConsecutiveFailures` times in a row. Never having + * scanned also trips — trading on a missing APY table is strictly worse than + * halting. A single failure by itself does not trip (transient blips happen); + * the consecutive-failure threshold is what makes it a circuit breaker. + */ +export function evaluateStaleDataRule(input: StaleDataInput): RuleResult { + const { + latestFetchedAt, + maxStaleMinutes, + maxConsecutiveFailures, + consecutiveFailures, + now, + } = input + + const detail: Record = { + maxStaleMinutes, + maxConsecutiveFailures, + } + + if (consecutiveFailures >= maxConsecutiveFailures) { + return { + tripped: true, + rule: 'stale_data', + detail: { + ...detail, + reason: 'consecutive_scan_failures', + consecutiveFailures, + }, + } + } + + if (latestFetchedAt === null) { + return { + tripped: true, + rule: 'stale_data', + detail: { ...detail, reason: 'never_scanned' }, + } + } + + const ageMinutes = (now.getTime() - latestFetchedAt.getTime()) / 60000 + const tripped = ageMinutes > maxStaleMinutes + + return { + tripped, + rule: 'stale_data', + detail: { ...detail, ageMinutes: round4(ageMinutes) }, + } +} + +export interface BreakerRuleConfig { + abnormalLoss: { enabled: boolean; lossPct: number; windowHours: number } + depeg: { enabled: boolean; depegBps: number } + oscillation: { enabled: boolean; maxFlips: number } + staleData: { + enabled: boolean + maxStaleMinutes: number + maxConsecutiveFailures: number + } +} + +export interface BreakerEvalInput { + /** Optional mark-to-market series for abnormal_loss (skip rule if absent). */ + abnormalLossSeries?: ValuePoint[] + /** Current stablecoin USD price; null = no feed. */ + depegPrice: number | null + /** Rebalance count for the batch within the flip window. */ + oscillationFlips: number + /** Latest successful APY scan time; null = never scanned. */ + latestFetchedAt: Date | null + /** Consecutive scan failures so far. */ + consecutiveFailures: number + /** Authoritative evaluation time (the command/fetch time, not Date.now()). */ + now: Date +} + +/** + * Run all enabled rules in evaluation order (abnormal_loss → depeg → + * oscillation → stale_data) and return the first trip, or a no-trip result. + * Manual trips are applied through the admin API directly, not here. + */ +export function evaluateBreakerRules( + config: BreakerRuleConfig, + input: BreakerEvalInput +): RuleResult { + const noTrip: RuleResult = { + tripped: false, + rule: 'stale_data', + detail: { reason: 'none' }, + } + + if (config.abnormalLoss.enabled && input.abnormalLossSeries) { + const r = evaluateAbnormalLossRule({ + series: input.abnormalLossSeries, + lossPct: config.abnormalLoss.lossPct, + windowHours: config.abnormalLoss.windowHours, + now: input.now, + }) + if (r.tripped) return r + } + + if (config.depeg.enabled) { + const r = evaluateDepegRule({ + price: input.depegPrice, + depegBps: config.depeg.depegBps, + }) + if (r.tripped) return r + } + + if (config.oscillation.enabled) { + const r = evaluateOscillationRule({ + flips: input.oscillationFlips, + maxFlips: config.oscillation.maxFlips, + }) + if (r.tripped) return r + } + + if (config.staleData.enabled) { + const r = evaluateStaleDataRule({ + latestFetchedAt: input.latestFetchedAt, + maxStaleMinutes: config.staleData.maxStaleMinutes, + maxConsecutiveFailures: config.staleData.maxConsecutiveFailures, + consecutiveFailures: input.consecutiveFailures, + now: input.now, + }) + if (r.tripped) return r + } + + return noTrip +} + +/** + * Validate a breaker config at boot. Throws on impossible values so a + * misconfigured deployment stops loudly instead of silently guarding nothing. + */ +export function validateBreakerConfig(config: BreakerRuleConfig): void { + const { abnormalLoss, depeg, oscillation, staleData } = config + + if (abnormalLoss.enabled && abnormalLoss.lossPct <= 0) { + throw new Error( + `Invalid BREAKER_LOSS_PCT: ${abnormalLoss.lossPct} (must be > 0)` + ) + } + if (abnormalLoss.enabled && abnormalLoss.windowHours <= 0) { + throw new Error( + `Invalid BREAKER_LOSS_WINDOW_HOURS: ${abnormalLoss.windowHours} (must be > 0)` + ) + } + if (depeg.enabled && depeg.depegBps < 0) { + throw new Error( + `Invalid BREAKER_DEPEG_BPS: ${depeg.depegBps} (must be >= 0)` + ) + } + if (oscillation.enabled && oscillation.maxFlips < 2) { + throw new Error( + `Invalid BREAKER_MAX_FLIPS: ${oscillation.maxFlips} (must be >= 2)` + ) + } + if (staleData.enabled && staleData.maxStaleMinutes <= 0) { + throw new Error( + `Invalid BREAKER_STALE_MINUTES: ${staleData.maxStaleMinutes} (must be > 0)` + ) + } + if (staleData.enabled && staleData.maxConsecutiveFailures < 1) { + throw new Error( + `Invalid BREAKER_STALE_CONSECUTIVE_FAILURES: ${staleData.maxConsecutiveFailures} (must be >= 1)` + ) + } +} + +function round4(n: number): number { + return Math.round(n * 10000) / 10000 +} diff --git a/src/agent/breakerService.ts b/src/agent/breakerService.ts new file mode 100644 index 0000000..3c7330c --- /dev/null +++ b/src/agent/breakerService.ts @@ -0,0 +1,1030 @@ +/** + * src/agent/breakerService.ts + * + * Integration layer for the agent circuit breaker (#345). The heavy lifting + * (trip rules, state machine) lives in pure modules (breakerRules.ts, + * breakerState.ts); this service: + * + * 1. loads the measurements the rules need (position value series, batch flip + * counts, APY-table freshness, stablecoin price), + * 2. evaluates the enabled rules per scope (GLOBAL -> PROTOCOL -> USER), + * 3. applies the CLOSED/OPEN/HALF_OPEN state machine and persists transitions, + * 4. emits operator alerts + user events on trip/reset, + * 5. exposes a blocking context the rebalance loop consumes before batches. + * + * Evaluations FAIL CLOSED: if measurement loading or persistence throws, the + * caller is told to halt all rebalancing for the tick and an operator alert + * is emitted — the agent never trades blind when the breaker cannot decide. + * + * Withdrawals are deliberately untouched here: this module only gates + * agent-initiated rebalances. The outbox dispatcher's `isUserHalted` gate + * (#321) is independent and out of scope. + */ + +import { config } from '../config/env' +import db from '../db' +import { logger } from '../utils/logger' +import { alertingService } from '../services/alerting' +import { publishUserEvent } from '../events/publisher' +import { EVENT_TYPE_TOPIC } from '../events/types' +import { recordAgentBreakerTrip, setAgentBreakerState } from '../utils/metrics' +import { + BreakerRuleConfig, + BreakerEvalInput, + evaluateBreakerRules, + RuleResult, + ValuePoint, +} from './breakerRules' +import { + applyBreakerEvaluation, + applyManualReset, + applyManualTrip, + BreakerRecord, + BreakerScope, + BreakerState, + BreakerTransitionConfig, + describeBreaker, +} from './breakerState' + +// ── Types ────────────────────────────────────────────────────────────────────── + +export interface PositionLike { + id: string + userId: string + protocolName: string +} + +export interface BatchLike { + batchKey: string + protocol: string +} + +/** What the rebalance loop needs to decide which batches may run. */ +export interface BreakerBlockContext { + /** True when a GLOBAL breaker is OPEN (halt everything this tick). */ + globalOpen: boolean + /** Protocols whose breaker is OPEN (halt batches whose FROM is here). */ + openProtocols: Set + /** Users whose breaker is OPEN (halt batches containing any of these). */ + openUsers: Set + /** OPEN-protocol names to exclude as rebalance targets (FROM-AND-TO guard). */ + blockedTargetProtocols: string[] + /** Breakers that newly tripped this tick. */ + trips: TripRecord[] + /** Breakers that closed this tick (auto-recovered). */ + closes: TripRecord[] + /** True when evaluation itself failed — everything blocked (fail-closed). */ + evalFailed: boolean +} + +export interface TripRecord { + id: string + scope: BreakerScope + scopeKey: string + rule: string + state: BreakerState + autoResetAt: string | null +} + +interface BreakerRowLike { + id: string + scope: string + scopeKey: string + state: string + trippedRule: string | null + detail: unknown + trippedAt: Date | null + autoResetAt: Date | null + resetBy: string | null + resetAt: Date | null +} + +// ── Config adapters ──────────────────────────────────────────────────────────── + +export function breakerRuleConfig(): BreakerRuleConfig { + const b = config.breaker + return { + abnormalLoss: { + enabled: b.abnormalLoss.enabled, + lossPct: b.abnormalLoss.lossPct, + windowHours: b.abnormalLoss.windowHours, + }, + depeg: { + enabled: b.depeg.enabled, + depegBps: b.depeg.depegBps, + }, + oscillation: { + enabled: b.oscillation.enabled, + maxFlips: b.oscillation.maxFlips, + }, + staleData: { + enabled: b.staleData.enabled, + maxStaleMinutes: b.staleData.staleMinutes, + maxConsecutiveFailures: b.staleData.consecutiveFailures, + }, + } +} + +export function breakerTransitionConfig(): BreakerTransitionConfig { + return { + cooldownMs: config.breaker.cooldownMs, + maxCooldownMs: config.breaker.maxCooldownMs, + sustainedClearChecks: config.breaker.depeg.sustainedClearChecks, + } +} + +// ── Stablecoin price provider ────────────────────────────────────────────────── + +/** + * Current USD spot price of the stablecoin the platform prices at $1, or null + * when no feed is available. + * + * #345 explicitly names the fee-oracle/routing path as the de-peg price + * source. As of this change the fee oracle publishes base-fee/congestion data + * only — it carries no stablecoin price — and the routing module's path + * finding is a stub, so there is no real spot price to read. The provider + * therefore returns null (the rule fails safe: null never trips), and the + * de-peg rule stays OFF by default (BREAKER_DEPEG_ENABLED=false) until an + * oracle feed exists. This function is the single integration point for that + * feed. + */ +export function getStablecoinPrice(): number | null { + return null +} + +// ── Row <-> record adapters ──────────────────────────────────────────────────── + +function normalizeState(s: string): BreakerState { + if (s === 'OPEN' || s === 'HALF_OPEN' || s === 'CLOSED') return s + return 'CLOSED' +} + +function normalizeScope(s: string): BreakerScope { + if (s === 'PROTOCOL' || s === 'USER' || s === 'GLOBAL') return s + return 'GLOBAL' +} + +function rowToRecord(row: BreakerRowLike): BreakerRecord { + return { + state: normalizeState(row.state), + trippedRule: (row.trippedRule as BreakerRecord['trippedRule']) ?? null, + detail: (row.detail as Record) ?? null, + trippedAt: row.trippedAt ?? null, + autoResetAt: row.autoResetAt ?? null, + } +} + +function keyFor(scope: BreakerScope, scopeKey: string): string { + return scope === 'GLOBAL' ? 'GLOBAL' : `${scope}:${scopeKey}` +} + +function emptyRecord(): BreakerRecord { + return { + state: 'CLOSED', + trippedRule: null, + detail: null, + trippedAt: null, + autoResetAt: null, + } +} + +const SNAPSHOT_ALERT_SEVERITY = 'critical' +const RESET_ALERT_SEVERITY = 'info' + +// ── Measurement loads ────────────────────────────────────────────────────────── + +function toNumber(v: unknown): number { + return typeof v === 'number' ? v : parseFloat(String(v ?? '0')) +} + +function toSeries(entries: Array<[number, number]>): ValuePoint[] { + return entries + .map(([t, value]) => ({ at: new Date(t), value })) + .sort((a, b) => a.at.getTime() - b.at.getTime()) +} + +interface ValueSeriesByKey { + global: ValuePoint[] + byProtocol: Map + byUser: Map +} + +/** + * Aggregate mark-to-market value series per scope from YieldSnapshot rows. + * Each snapshot's value is principal + yield at that time (the same rows the + * position-history API reads). Series are summed across positions sharing a + * scope key, oldest-first. + */ +async function loadValueSeries( + positions: PositionLike[], + windowHours: number, + now: Date +): Promise { + const ids = positions.map((p) => p.id).filter(Boolean) + const byPosition = new Map(positions.map((p) => [p.id, p])) + + const empty: ValueSeriesByKey = { + global: [], + byProtocol: new Map(), + byUser: new Map(), + } + if (ids.length === 0) return empty + + const windowStart = new Date(now.getTime() - windowHours * 60 * 60 * 1000) + + const snapshots = await db.yieldSnapshot.findMany({ + where: { positionId: { in: ids }, snapshotAt: { gte: windowStart } }, + select: { + positionId: true, + snapshotAt: true, + principalAmount: true, + yieldAmount: true, + }, + }) + + const buckets = new Map>() + const record = (key: string, t: number, value: number) => { + let m = buckets.get(key) + if (!m) { + m = new Map() + buckets.set(key, m) + } + m.set(t, (m.get(t) ?? 0) + value) + } + + for (const s of snapshots as Array<{ + positionId: string + snapshotAt: Date + principalAmount: unknown + yieldAmount: unknown + }>) { + const pos = byPosition.get(s.positionId) + if (!pos) continue + const value = toNumber(s.principalAmount) + toNumber(s.yieldAmount) + if (!Number.isFinite(value)) continue + const t = s.snapshotAt.getTime() + record('global', t, value) + record(`proto:${pos.protocolName}`, t, value) + record(`user:${pos.userId}`, t, value) + } + + const global = toSeries( + buckets.get('global') ? Array.from(buckets.get('global')!) : [] + ) + + const byProtocol = new Map() + const byUser = new Map() + for (const [k, m] of buckets) { + const series = toSeries(Array.from(m.entries())) + if (k.startsWith('proto:')) byProtocol.set(k.slice('proto:'.length), series) + else if (k.startsWith('user:')) byUser.set(k.slice('user:'.length), series) + } + + return { global, byProtocol, byUser } +} + +/** REBALANCED decision count per batchKey within the flip window. */ +async function loadFlipCounts( + batchKeys: string[], + flipWindowHours: number, + now: Date +): Promise> { + const keys = Array.from(new Set(batchKeys.filter(Boolean))) + const counts = new Map(keys.map((k) => [k, 0])) + if (keys.length === 0) return counts + + const windowStart = new Date(now.getTime() - flipWindowHours * 60 * 60 * 1000) + const rows = await db.rebalanceDecision.groupBy({ + by: ['batchKey'], + where: { + batchKey: { in: keys }, + outcome: 'REBALANCED', + createdAt: { gte: windowStart }, + }, + _count: { _all: true }, + }) + + for (const r of rows as Array<{ + batchKey: string + _count: { _all: number } + }>) { + counts.set(r.batchKey, r._count._all) + } + return counts +} + +async function loadLatestProtocolRate(): Promise<{ + latestFetchedAt: Date | null + fresh: boolean +}> { + const row = await db.protocolRate.findFirst({ + orderBy: { fetchedAt: 'desc' }, + select: { fetchedAt: true }, + }) + const latestFetchedAt = row?.fetchedAt ?? null + const fresh = + latestFetchedAt !== null && + Date.now() - latestFetchedAt.getTime() < + config.breaker.staleData.staleMinutes * 60 * 1000 + return { latestFetchedAt, fresh } +} + +/** Consecutive failed scans — in-memory, like the scanner's own counters. */ +let staleDataFailureCount = 0 + +function trackScanHealth(fresh: boolean): void { + staleDataFailureCount = fresh ? 0 : staleDataFailureCount + 1 +} + +// ── Sync status cache (for the sync getAgentStatus surface) ─────────────────── + +let cachedGlobalSummary: { + state: string + trippedRule: string | null + autoResetAt: string | null +} | null = null + +function updateGlobalCache(wk: WorkingBreaker | undefined, now: Date): void { + void now + if (!wk) { + cachedGlobalSummary = null + return + } + cachedGlobalSummary = { + state: wk.record.state, + trippedRule: wk.record.trippedRule, + autoResetAt: wk.record.autoResetAt?.toISOString() ?? null, + } +} + +/** Same shape exposed by listBreakers but DB-free, for the sync status route. */ +export function getBreakerStatusSummary(): { + global: { + state: string + trippedRule: string | null + autoResetAt: string | null + } | null +} { + return { global: cachedGlobalSummary } +} + +/** + * User-facing breaker status: what applies to one player. Only plain-language + * fields — never other users' loss figures or thresholds. + */ +export async function getBreakerStatusForUser( + userId: string +): Promise<{ global: string | null; affectingYou: string[] }> { + const rows = await db.agentCircuitBreaker.findMany({ + where: { + OR: [ + { scope: 'GLOBAL', scopeKey: '' }, + { scope: 'USER', scopeKey: userId }, + ], + }, + }) + + let global: string | null = null + const affectingYou: string[] = [] + + for (const r of rows) { + // Defense in depth: never infer "applies to me" from the query alone — + // a USER row only ever describes the calling user. + if (r.scope === 'USER' && r.scopeKey !== userId) continue + const state = normalizeState(r.state) + if (state === 'CLOSED') continue + if (r.scope === 'GLOBAL') { + global = r.trippedRule ?? 'open' + } else if (r.scope === 'USER') { + affectingYou.push( + `rebalancing paused for your account: ${r.trippedRule ?? 'manual'}` + ) + } + } + + return { global, affectingYou } +} + +// ── Alerts + user events ─────────────────────────────────────────────────────── + +function safeAlert( + payload: Parameters[0], + alertKey: string +): void { + try { + void alertingService.emit(payload, alertKey).catch(() => {}) + } catch { + // alerting must never break the money path + } +} + +function plainTripReason(rule: string, scope: BreakerScope): string { + switch (rule) { + case 'abnormal_loss': + return 'Your portfolio has taken an abnormal loss; the agent paused automatic rebalancing to avoid churning fees while positions fall.' + case 'depeg': + return 'A stablecoin the platform prices at $1 moved outside its band; the agent paused automatic rebalancing around that asset.' + case 'oscillation': + return 'Rebalancing flipped between protocols repeatedly; the agent paused to stop burning fees.' + case 'stale_data': + return 'Yield data is stale or unavailable; the agent paused rebalancing rather than trade on outdated rates.' + case 'manual': + return scope === 'USER' + ? 'An operator paused automatic rebalancing for your account.' + : 'An operator paused automatic rebalancing.' + default: + return 'Automatic rebalancing has been paused.' + } +} + +function emitTripEvents( + scope: BreakerScope, + scopeKey: string, + rule: string, + users: string[] +): void { + const described = scope === 'GLOBAL' ? 'globally' : `for ${scope} ${scopeKey}` + safeAlert( + { + title: `Agent circuit breaker tripped (${rule})`, + description: `Rebalancing is paused ${described}. Rule: ${rule}. Auto-recovery: when the cooldown elapses and the condition clears.`, + severity: SNAPSHOT_ALERT_SEVERITY as 'critical', + component: 'agent-breaker', + metadata: { scope, scopeKey, rule }, + }, + `agent-breaker:trip:${scope}:${scopeKey}` + ) + + if (scope === 'GLOBAL' || scope === 'USER') { + publishUserEvent( + users, + EVENT_TYPE_TOPIC['agent.circuit_breaker_tripped'], + 'agent.circuit_breaker_tripped', + { + scope, + scopeKey, + rule, + reason: plainTripReason(rule, scope), + } + ).catch(() => {}) + } +} + +function emitResetEvents( + scope: BreakerScope, + scopeKey: string, + rule: string, + users: string[] +): void { + const described = scope === 'GLOBAL' ? 'globally' : `for ${scope} ${scopeKey}` + safeAlert( + { + title: `Agent circuit breaker reset (${rule})`, + description: `Rebalancing may resume ${described}.`, + severity: RESET_ALERT_SEVERITY as 'info', + component: 'agent-breaker', + metadata: { scope, scopeKey, rule }, + }, + `agent-breaker:reset:${scope}:${scopeKey}` + ) + + if (scope === 'GLOBAL' || scope === 'USER') { + publishUserEvent( + users, + EVENT_TYPE_TOPIC['agent.circuit_breaker_reset'], + 'agent.circuit_breaker_reset', + { + scope, + scopeKey, + rule, + reason: 'Rebalancing has resumed.', + } + ).catch(() => {}) + } +} + +/** Fail-closed alert when the breaker evaluation itself errors. */ +export function emitBreakerEvalFailureAlert(error: string): void { + safeAlert( + { + title: 'Agent circuit breaker evaluation failed', + description: `Breaker evaluation threw (${error}); all rebalancing is halted for the tick (fail-closed).`, + severity: SNAPSHOT_ALERT_SEVERITY as 'critical', + component: 'agent-breaker', + metadata: { error }, + }, + 'agent-breaker:eval-failed' + ) +} + +// ── Tick evaluation ───────────────────────────────────────────────────────────── + +interface WorkingBreaker { + key: string + id: string + scope: BreakerScope + scopeKey: string + row: BreakerRowLike + record: BreakerRecord + persisted: boolean +} + +async function loadWorkingBreakers( + now: Date +): Promise> { + const rows = await db.agentCircuitBreaker.findMany() + const map = new Map() + for (const r of rows) { + const scope = normalizeScope(r.scope as string) + const wk: WorkingBreaker = { + key: keyFor(scope, r.scopeKey), + id: r.id, + scope, + scopeKey: r.scopeKey, + row: { + id: r.id, + scope: r.scope, + scopeKey: r.scopeKey, + state: r.state, + trippedRule: r.trippedRule, + detail: r.detail, + trippedAt: r.trippedAt, + autoResetAt: r.autoResetAt, + resetBy: r.resetBy, + resetAt: r.resetAt, + }, + record: rowToRecord({ ...r } as unknown as BreakerRowLike), + persisted: true, + } + map.set(wk.key, wk) + } + return map +} + +function recordsEqual(a: BreakerRecord, b: BreakerRecord): boolean { + return ( + a.state === b.state && + a.trippedRule === b.trippedRule && + a.trippedAt?.getTime() === b.trippedAt?.getTime() && + a.autoResetAt?.getTime() === b.autoResetAt?.getTime() && + JSON.stringify(a.detail ?? null) === JSON.stringify(b.detail ?? null) + ) +} + +function getOrCreate( + map: Map, + scope: BreakerScope, + scopeKey: string +): WorkingBreaker { + const key = keyFor(scope, scopeKey) + const existing = map.get(key) + if (existing) return existing + const created: WorkingBreaker = { + key, + id: key + ':unpersisted', + scope, + scopeKey, + row: { + id: key + ':unpersisted', + scope, + scopeKey, + state: 'CLOSED', + trippedRule: null, + detail: null, + trippedAt: null, + autoResetAt: null, + resetBy: null, + resetAt: null, + }, + record: emptyRecord(), + persisted: false, + } + map.set(key, created) + return created +} + +async function persistIfChanged( + wk: WorkingBreaker, + prev: BreakerRecord, + next: BreakerRecord +): Promise { + if (recordsEqual(prev, next)) return + if (!wk.persisted && next.state === 'CLOSED' && next.trippedRule === null) + return + + const data = { + state: next.state, + trippedRule: next.trippedRule, + trippedAt: next.trippedAt, + detail: (next.detail as any) ?? undefined, + autoResetAt: next.autoResetAt, + } + + if (wk.persisted) { + await db.agentCircuitBreaker.update({ where: { id: wk.id }, data }) + } else { + const created = await db.agentCircuitBreaker.create({ + data: { scope: wk.scope, scopeKey: wk.scopeKey, ...data }, + }) + wk.persisted = true + wk.id = created.id + } +} + +/** + * Apply one rule outcome to a working breaker and record trip/close events. + */ +async function applyAndRecord( + wk: WorkingBreaker, + outcome: RuleResult, + now: Date, + usersAffected: string[] +): Promise { + const prev = wk.record + const next = applyBreakerEvaluation( + prev, + outcome, + now, + breakerTransitionConfig() + ) + + const stateChanged = prev.state !== next.state + + if (!stateChanged) { + // Still CLOSED or still OPEN. An OPEN breaker accrues lastEvaluation + // detail — persist that so operators can inspect failed probes. + if (next.state === 'OPEN' && !recordsEqual(prev, next)) { + wk.record = next + await persistIfChanged(wk, prev, next) + } + return + } + + // CLOSED -> OPEN or HALF_OPEN -> OPEN: a trip. + if (next.state === 'OPEN') { + wk.record = next + await persistIfChanged(wk, prev, next) + recordAgentBreakerTrip(wk.scope, next.trippedRule ?? outcome.rule) + setAgentBreakerState(wk.scope, wk.scopeKey, 'OPEN') + emitTripEvents( + wk.scope, + wk.scopeKey, + next.trippedRule ?? outcome.rule, + usersAffected + ) + return + } + + // OPEN -> HALF_OPEN: cooldown elapsed + condition cleared; probed. + if (next.state === 'HALF_OPEN') { + wk.record = next + await persistIfChanged(wk, prev, next) + return + } + + // HALF_OPEN -> CLOSED: recovered. + if (next.state === 'CLOSED' && prev.state === 'HALF_OPEN') { + wk.record = next + await persistIfChanged(wk, prev, next) + emitResetEvents( + wk.scope, + wk.scopeKey, + prev.trippedRule ?? 'auto', + usersAffected + ) + return + } + + wk.record = next + await persistIfChanged(wk, prev, next) +} + +/** + * Evaluate breakers for one agent tick and return the blocking context for the + * rebalance loop. Throws on DB failure — the loop catches and fails closed. + */ +export async function evaluateBreakerTick( + positions: PositionLike[], + batches: BatchLike[], + now: Date +): Promise { + const ruleCfg = breakerRuleConfig() + const transCfg = breakerTransitionConfig() + + // Load measurements once per tick. + const series = await loadValueSeries( + positions, + ruleCfg.abnormalLoss.windowHours, + now + ) + const flipCounts = await loadFlipCounts( + batches.map((b) => b.batchKey), + config.breaker.oscillation.windowHours, + now + ) + const { latestFetchedAt, fresh } = await loadLatestProtocolRate() + trackScanHealth(fresh) + + const depegPrice = config.breaker.depeg.enabled ? getStablecoinPrice() : null + + const byKey = await loadWorkingBreakers(now) + const trips: TripRecord[] = [] + const closes: TripRecord[] = [] + const usersByProtocol = new Map() + + const protocols = Array.from(new Set(positions.map((p) => p.protocolName))) + const userIds = Array.from(new Set(positions.map((p) => p.userId))) + for (const p of protocols) { + usersByProtocol.set( + p, + positions.filter((x) => x.protocolName === p).map((x) => x.userId) + ) + } + + const evalStaleness: Pick< + BreakerEvalInput, + 'latestFetchedAt' | 'consecutiveFailures' | 'now' + > = { + latestFetchedAt, + consecutiveFailures: staleDataFailureCount, + now, + } + + // GLOBAL: stale_data + abnormal_loss. + { + const wk = getOrCreate(byKey, 'GLOBAL', '') + const outcome = evaluateBreakerRules(ruleCfg, { + ...evalStaleness, + abnormalLossSeries: ruleCfg.abnormalLoss.enabled + ? series.global + : undefined, + depegPrice, + oscillationFlips: 0, + }) + const before = wk.record + await applyAndRecord(wk, outcome, now, userIds) + if (before.state !== 'OPEN' && wk.record.state === 'OPEN') { + trips.push(toTripRecord(wk)) + } else if (before.state === 'HALF_OPEN' && wk.record.state === 'CLOSED') { + closes.push(toTripRecord(wk)) + } + } + + // PROTOCOL: abnormal_loss (per protocol) + oscillation (batch flips) + depeg. + for (const protocol of protocols) { + const wk = getOrCreate(byKey, 'PROTOCOL', protocol) + const batchKeys = batches + .filter((b) => b.protocol === protocol) + .map((b) => b.batchKey) + const maxFlips = Math.max( + 0, + ...batchKeys.map((k) => flipCounts.get(k) ?? 0) + ) + const outcome = evaluateBreakerRules(ruleCfg, { + ...evalStaleness, + abnormalLossSeries: ruleCfg.abnormalLoss.enabled + ? (series.byProtocol.get(protocol) ?? []) + : undefined, + depegPrice, + oscillationFlips: maxFlips, + }) + const before = wk.record + await applyAndRecord(wk, outcome, now, usersByProtocol.get(protocol) ?? []) + if (before.state !== 'OPEN' && wk.record.state === 'OPEN') { + trips.push(toTripRecord(wk)) + } else if (before.state === 'HALF_OPEN' && wk.record.state === 'CLOSED') { + closes.push(toTripRecord(wk)) + } + } + + // USER: abnormal_loss per user. + for (const userId of userIds) { + const wk = getOrCreate(byKey, 'USER', userId) + const outcome = evaluateBreakerRules(ruleCfg, { + ...evalStaleness, + abnormalLossSeries: ruleCfg.abnormalLoss.enabled + ? (series.byUser.get(userId) ?? []) + : undefined, + depegPrice, + oscillationFlips: 0, + }) + const before = wk.record + await applyAndRecord(wk, outcome, now, [userId]) + if (before.state !== 'OPEN' && wk.record.state === 'OPEN') { + trips.push(toTripRecord(wk)) + } else if (before.state === 'HALF_OPEN' && wk.record.state === 'CLOSED') { + closes.push(toTripRecord(wk)) + } + } + + // Build the blocking context from the (now final) states. + const openProtocols = new Set() + const openUsers = new Set() + const blockedTargetProtocols = new Set() + + let globalOpen = false + let globalWk: WorkingBreaker | undefined + for (const wk of byKey.values()) { + if (wk.scope === 'GLOBAL') globalWk = wk + const open = wk.record.state === 'OPEN' || wk.record.state === 'HALF_OPEN' + if (!open) continue + if (wk.scope === 'GLOBAL') globalOpen = true + if (wk.scope === 'PROTOCOL') { + openProtocols.add(wk.scopeKey) + blockedTargetProtocols.add(wk.scopeKey) + } + if (wk.scope === 'USER') openUsers.add(wk.scopeKey) + } + + updateGlobalCache(globalWk, now) + + logger.info('[Breaker] tick complete', { + globalOpen, + openProtocols: Array.from(openProtocols), + openUsers: Array.from(openUsers).length, + trips: trips.map((t) => t.rule), + closes: closes.map((c) => c.rule), + }) + + for (const wk of byKey.values()) { + setAgentBreakerState(wk.scope, wk.scopeKey, wk.record.state) + } + + return { + globalOpen, + openProtocols, + openUsers, + blockedTargetProtocols: Array.from(blockedTargetProtocols), + trips, + closes, + evalFailed: false, + } +} + +function toTripRecord(wk: WorkingBreaker): TripRecord { + return { + id: wk.id, + scope: wk.scope, + scopeKey: wk.scopeKey, + rule: wk.record.trippedRule ?? 'auto', + state: wk.record.state, + autoResetAt: wk.record.autoResetAt?.toISOString() ?? null, + } +} + +// ── Manual admin operations (used by admin routes) ──────────────────────────── + +export async function listBreakers(): Promise>> { + const rows = await db.agentCircuitBreaker.findMany({ + orderBy: { updatedAt: 'desc' }, + }) + return rows.map((r) => ({ + id: r.id, + scope: r.scope, + scopeKey: r.scopeKey, + state: r.state, + trippedRule: r.trippedRule, + trippedAt: r.trippedAt?.toISOString() ?? null, + autoResetAt: r.autoResetAt?.toISOString() ?? null, + resetBy: r.resetBy, + resetAt: r.resetAt?.toISOString() ?? null, + detail: r.detail, + updatedAt: r.updatedAt.toISOString(), + })) +} + +async function activeUserIds(): Promise { + const rows = await db.user.findMany({ + where: { isActive: true }, + select: { id: true }, + }) + return rows.map((r) => r.id) +} + +export async function manualTripBreaker( + scope: BreakerScope, + scopeKey: string, + reason: string, + adminIdentity: string +): Promise> { + const now = new Date() + const existing = await db.agentCircuitBreaker.findUnique({ + where: { scope_scopeKey: { scope, scopeKey } }, + }) + + const record = existing + ? rowToRecord(existing as unknown as BreakerRowLike) + : emptyRecord() + + if (record.state === 'OPEN' && record.trippedRule === 'manual' && existing) { + // Already manually open — no-op, return current state. + return { + id: existing.id, + scope, + scopeKey, + state: record.state, + trippedRule: record.trippedRule, + trippedAt: record.trippedAt?.toISOString() ?? null, + autoResetAt: record.autoResetAt?.toISOString() ?? null, + } + } + + const next = applyManualTrip(record, now, breakerTransitionConfig()) + + let id: string + if (existing) { + id = existing.id + await db.agentCircuitBreaker.update({ + where: { id }, + data: { + state: next.state, + trippedRule: next.trippedRule, + trippedAt: next.trippedAt, + detail: (next.detail as any) ?? undefined, + autoResetAt: next.autoResetAt, + resetBy: null, + resetAt: null, + }, + }) + } else { + const created = await db.agentCircuitBreaker.create({ + data: { + scope, + scopeKey, + state: next.state, + trippedRule: next.trippedRule, + trippedAt: next.trippedAt, + detail: (next.detail as any) ?? undefined, + autoResetAt: next.autoResetAt, + }, + }) + id = created.id + } + + const users = + scope === 'GLOBAL' + ? await activeUserIds() + : scope === 'USER' + ? [scopeKey] + : [] + emitTripEvents(scope, scopeKey, 'manual', users) + + const persisted = await db.agentCircuitBreaker.findUniqueOrThrow({ + where: { id }, + }) + return { + id, + scope, + scopeKey, + reason, + resetBy: adminIdentity, + ...describeBreaker(rowToRecord(persisted as unknown as BreakerRowLike)), + } +} + +export async function manualResetBreaker( + id: string, + reason: string, + adminIdentity: string +): Promise | null> { + const existing = await db.agentCircuitBreaker.findUnique({ where: { id } }) + if (!existing) return null + const now = new Date() + const next = applyManualReset( + rowToRecord(existing as unknown as BreakerRowLike), + now, + adminIdentity, + reason + ) + + await db.agentCircuitBreaker.update({ + where: { id: existing.id }, + data: { + state: next.state, + trippedRule: next.trippedRule, + trippedAt: next.trippedAt, + detail: (next.detail as any) ?? undefined, + autoResetAt: next.autoResetAt, + resetBy: adminIdentity, + resetAt: now, + }, + }) + + const scope = normalizeScope(existing.scope) + const users = + scope === 'GLOBAL' + ? await activeUserIds() + : scope === 'USER' + ? [existing.scopeKey] + : [] + emitResetEvents(scope, existing.scopeKey, 'manual', users) + + return { + id, + scope: scope, + scopeKey: existing.scopeKey, + resetBy: adminIdentity, + reason, + state: next.state, + } +} diff --git a/src/agent/breakerState.ts b/src/agent/breakerState.ts new file mode 100644 index 0000000..3caaceb --- /dev/null +++ b/src/agent/breakerState.ts @@ -0,0 +1,259 @@ +/** + * src/agent/breakerState.ts + * + * Pure state machine for the agent circuit breaker (#345). + * + * CLOSED --trip--> OPEN --(cooldown elapsed + sustained clean)--> HALF_OPEN + * HALF_OPEN --clean--> CLOSED + * HALF_OPEN --trip--> OPEN (cooldown doubles, capped) + * + * Machine state that must survive restarts (current cooldown, consecutive + * clean evaluations) is kept under the reserved `detail._machine` key so the + * schema's `detail` JSONB column carries it alongside the trip measurements. + * Nothing here touches the DB or the clock — `now` is always passed in. + */ + +import type { BreakerTripReason, RuleResult } from './breakerRules' + +export type BreakerState = 'CLOSED' | 'OPEN' | 'HALF_OPEN' +export type BreakerScope = 'GLOBAL' | 'PROTOCOL' | 'USER' + +export const BREAKER_SCOPES: readonly BreakerScope[] = [ + 'GLOBAL', + 'PROTOCOL', + 'USER', +] + +export interface BreakerTransitionConfig { + /** Base cooldown (ms) before an OPEN breaker may auto-reset. */ + cooldownMs: number + /** Hard cap (ms) on cooldown after repeated re-tripping. */ + maxCooldownMs: number + /** Consecutive clean evaluations required before OPEN -> HALF_OPEN. */ + sustainedClearChecks: number +} + +export interface BreakerRecord { + state: BreakerState + trippedRule: BreakerTripReason | null + detail: Record | null + trippedAt: Date | null + autoResetAt: Date | null +} + +export interface MachineState { + /** Consecutive clean evaluations since the last trip. */ + clearCount: number + /** Current cooldown (ms) applicable to the OPEN state. */ + cooldownMs: number +} + +const MACHINE_KEY = '_machine' + +function readMachine(detail: Record | null): MachineState { + const m = detail?.[MACHINE_KEY] + if (!m || typeof m !== 'object') { + return { clearCount: 0, cooldownMs: 0 } + } + return { + clearCount: + typeof m.clearCount === 'number' && m.clearCount >= 0 ? m.clearCount : 0, + cooldownMs: + typeof m.cooldownMs === 'number' && m.cooldownMs > 0 ? m.cooldownMs : 0, + } +} + +function writeMachine( + detail: Record | null, + machine: MachineState +): Record { + return { ...(detail ?? {}), [MACHINE_KEY]: machine } +} + +/** + * Apply a rule evaluation outcome to the current breaker record and return the + * resulting record (pure). Only transitions; does not emit alerts/events. + */ +export function applyBreakerEvaluation( + current: BreakerRecord, + outcome: RuleResult, + now: Date, + config: BreakerTransitionConfig +): BreakerRecord { + return outcome.tripped + ? handleTrip(current, outcome, now, config) + : handleClean(current, now, config) +} + +function handleTrip( + current: BreakerRecord, + outcome: RuleResult, + now: Date, + config: BreakerTransitionConfig +): BreakerRecord { + const machine = readMachine(current.detail) + + if (current.state === 'OPEN') { + // Already open; stay open, record the failed evaluation. autoResetAt was + // set on the original trip and governs the earliest recovery attempt. + return { + ...current, + detail: { + ...(current.detail ?? {}), + lastEvaluation: { + rule: outcome.rule, + detail: outcome.detail, + at: now.toISOString(), + }, + }, + } + } + + if (current.state === 'HALF_OPEN') { + // Probe failed: back to OPEN with doubled cooldown (floored at the base). + const nextCooldown = Math.min( + Math.max(machine.cooldownMs || config.cooldownMs, config.cooldownMs) * 2, + config.maxCooldownMs + ) + return toOpen(current, outcome, now, nextCooldown) + } + + // CLOSED -> OPEN with the base cooldown. + return toOpen(current, outcome, now, config.cooldownMs) +} + +function toOpen( + current: BreakerRecord, + outcome: RuleResult, + now: Date, + cooldownMs: number +): BreakerRecord { + return { + state: 'OPEN', + trippedRule: outcome.rule, + trippedAt: now, + autoResetAt: new Date(now.getTime() + cooldownMs), + detail: writeMachine( + { + rule: outcome.rule, + detail: outcome.detail, + at: now.toISOString(), + }, + { clearCount: 0, cooldownMs } + ), + } +} + +function handleClean( + current: BreakerRecord, + now: Date, + config: BreakerTransitionConfig +): BreakerRecord { + const machine = readMachine(current.detail) + const clearCount = machine.clearCount + 1 + + if (current.state === 'OPEN') { + const cooldownReady = + current.autoResetAt !== null && now >= current.autoResetAt + const sustained = clearCount >= config.sustainedClearChecks + if (cooldownReady && sustained) { + return { + ...current, + state: 'HALF_OPEN', + detail: writeMachine(current.detail, { ...machine, clearCount }), + } + } + return { + ...current, + detail: writeMachine(current.detail, { ...machine, clearCount }), + } + } + + if (current.state === 'HALF_OPEN') { + // One clean full evaluation cycle in HALF_OPEN closes the breaker. + return { + state: 'CLOSED', + trippedRule: null, + trippedAt: null, + autoResetAt: null, + detail: writeMachine(current.detail, { + clearCount: 0, + cooldownMs: config.cooldownMs, + }), + } + } + + // CLOSED stays CLOSED. + return current +} + +/** + * Manual trip: any state -> OPEN with rule 'manual'. Always audited by the + * caller. If the breaker is already OPEN manual, returns it unchanged. + */ +export function applyManualTrip( + current: BreakerRecord, + now: Date, + config: BreakerTransitionConfig +): BreakerRecord { + if (current.state === 'OPEN' && current.trippedRule === 'manual') { + return current + } + return { + state: 'OPEN', + trippedRule: 'manual', + trippedAt: now, + autoResetAt: new Date(now.getTime() + config.cooldownMs), + detail: writeMachine( + { rule: 'manual', at: now.toISOString() }, + { clearCount: 0, cooldownMs: config.cooldownMs } + ), + } +} + +/** + * Manual reset: any state -> CLOSED. Clears the trip and the machine state so + * the next automatic evaluation starts clean. Carries the admin identity plus + * the mandatory audit `reason` (recorded under detail._lastReset). + */ +export function applyManualReset( + current: BreakerRecord, + now: Date, + resetBy: string, + reason: string +): BreakerRecord { + return { + state: 'CLOSED', + trippedRule: null, + trippedAt: null, + autoResetAt: null, + detail: { + ...(current.detail ?? {}), + _lastReset: { + resetBy, + reason, + at: now.toISOString(), + }, + }, + } +} + +/** Readable summary for admin inspect + getAgentStatus surfaces. */ +export function describeBreaker(record: BreakerRecord): { + state: string + trippedRule: BreakerTripReason | null + trippedAt: string | null + autoResetAt: string | null + cooldownMs: number + clearCount: number +} { + const m = readMachine(record.detail) + return { + state: record.state, + trippedRule: record.trippedRule, + trippedAt: record.trippedAt?.toISOString() ?? null, + autoResetAt: record.autoResetAt?.toISOString() ?? null, + cooldownMs: m.cooldownMs, + clearCount: m.clearCount, + } +} diff --git a/src/agent/loop.ts b/src/agent/loop.ts index da20efd..19dbf4d 100644 --- a/src/agent/loop.ts +++ b/src/agent/loop.ts @@ -17,6 +17,14 @@ import { import { publishUserEvent } from '../events/publisher' import { EVENT_TYPE_TOPIC } from '../events/types' import { captureAllUserBalances, cleanupOldSnapshots } from './snapshotter' +import { + evaluateBreakerTick, + emitBreakerEvalFailureAlert, + getBreakerStatusSummary, + type BreakerBlockContext, +} from './breakerService' +import { persistRebalanceDecision } from './rebalanceDecision' +import type { DecisionTrace } from './types' import { resolveEffectiveConfig } from './effectiveStrategy' import { loadActiveFollowsForUsers, @@ -62,6 +70,7 @@ export function getAgentStatus() { lastError, healthStatus: determineHealthStatus(), lastTickAt, + breakers: getBreakerStatusSummary(), } } @@ -193,10 +202,114 @@ async function rebalanceCheckJob(): Promise { let rebalancesTriggered = 0 const thresholds = getThresholds() + // #345 Agent circuit breaker — evaluated BEFORE any batch executes. + // Order: GLOBAL -> PROTOCOL(from) -> USER. An OPEN breaker at any scope + // skips the affected batches and a BLOCKED decision is recorded. On + // failure the tick FAILS CLOSED: nothing rebalances and the operator is + // alerted — the agent never trades blind when the breaker cannot decide. + let breakerCtx: BreakerBlockContext + try { + breakerCtx = await evaluateBreakerTick( + positions.map((p: PositionWithUser) => ({ + id: p.id, + userId: p.userId, + protocolName: p.protocolName, + })), + Array.from(byProtocolAndStrategy.values()).map((b) => ({ + batchKey: b.batchKey, + protocol: b.protocol, + })), + new Date() + ) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + emitBreakerEvalFailureAlert(message) + logger.error( + '[Breaker] evaluation failed; halting all rebalancing for the tick', + { + correlationId, + error: message, + } + ) + breakerCtx = { + globalOpen: true, + openProtocols: new Set(), + openUsers: new Set(), + blockedTargetProtocols: [], + trips: [], + closes: [], + evalFailed: true, + } + } + + const recordBreakerBlockedDecision = async ( + batch: { + protocol: string + batchKey: string + positions: PositionWithUser[] + }, + blocking: string[], + detail: string + ): Promise => { + const affectedUserIds = Array.from( + new Set(batch.positions.map((p: PositionWithUser) => p.userId)) + ) + const trace: DecisionTrace = { + currentApy: null, + chosenProtocol: null, + chosenApy: null, + rawImprovement: null, + netImprovement: null, + estCostPercent: null, + costBreakdown: null, + thresholds, + candidates: [], + } + await persistRebalanceDecision({ + batchKey: batch.batchKey, + fromProtocol: batch.protocol, + outcome: 'BLOCKED', + blockedReason: 'circuit_breaker_open', + thresholds, + trace, + rationale: `Skipped by agent circuit breaker (${detail}): ${blocking.join(', ')}`, + affectedUserIds, + affectedPositions: batch.positions.length, + }) + } + for (const batch of byProtocolAndStrategy.values()) { const { protocol, positions: protocolPositions } = batch const lead = effectiveByUser.get(protocolPositions[0].userId)! + // #345 — skip batches affected by an OPEN breaker. Precedence: + // GLOBAL beats PROTOCOL beats USER; a user's own breaker never helps + // them if the whole market is halted. Recorded as a BLOCKED decision + // so the explainable-rebalance ledger shows WHY nothing moved. + const blocking = breakerCtx.globalOpen + ? ['GLOBAL'] + : [ + ...(breakerCtx.openProtocols.has(protocol) + ? [`PROTOCOL:${protocol}`] + : []), + ...protocolPositions + .filter((p: PositionWithUser) => + breakerCtx.openUsers.has(p.userId) + ) + .map((p: PositionWithUser) => `USER:${p.userId}`), + ] + + if (blocking.length > 0) { + await recordBreakerBlockedDecision( + batch, + blocking, + breakerCtx.evalFailed + ? 'breaker evaluation failed' + : 'open circuit breaker' + ) + continue + } + // HAZARD: this guard used to be `strategyName ? … : undefined`, and // executeRebalanceIfNeeded skips the strategy engine entirely when // preferences are undefined. A follower whose OWN rebalanceStrategy is @@ -243,6 +356,7 @@ async function rebalanceCheckJob(): Promise { strategyName: lead.config.strategyName ?? null, strategyIsFollowed: Boolean(lead.follow), followedStrategyId: lead.follow?.followedStrategyId ?? null, + blockedProtocols: breakerCtx.blockedTargetProtocols, } ) @@ -301,6 +415,14 @@ async function rebalanceCheckJob(): Promise { positionsChecked: positions.length, rebalancesTriggered, duration, + breakers: { + globalOpen: breakerCtx.globalOpen, + evalFailed: breakerCtx.evalFailed, + trips: breakerCtx.trips.map( + (t) => `${t.scope}:${t.scopeKey}:${t.rule}` + ), + closes: breakerCtx.closes.map((c) => `${c.scope}:${c.scopeKey}`), + }, }, }) diff --git a/src/agent/router.ts b/src/agent/router.ts index 1295303..daeb15c 100644 --- a/src/agent/router.ts +++ b/src/agent/router.ts @@ -206,7 +206,8 @@ function planFromStrategyDecision( export async function compareProtocols( currentProtocol: string, amount: string = '0', - thresholds: RebalanceThresholds = DEFAULT_THRESHOLDS + thresholds: RebalanceThresholds = DEFAULT_THRESHOLDS, + excludedProtocols: string[] = [] ): Promise { try { // Get current on-chain APY @@ -216,8 +217,13 @@ export async function compareProtocols( return null } - // Get best available protocol from latest scan - const allProtocols = await scanAllProtocols() + // Get best available protocol from latest scan (#345: exclude protocols + // whose circuit breaker is OPEN — never move INTO a broken protocol). + let allProtocols = await scanAllProtocols() + if (excludedProtocols.length > 0) { + const excluded = new Set(excludedProtocols) + allProtocols = allProtocols.filter((p) => !excluded.has(p.name)) + } if (allProtocols.length === 0) { logger.warn('No protocols available for comparison') return null @@ -499,6 +505,11 @@ export interface RebalanceBatchContext { strategyName?: string | null strategyIsFollowed?: boolean followedStrategyId?: string | null + /** + * #345 — protocol names with an OPEN circuit breaker. Excluded as rebalance + * targets so a broken protocol is never moved INTO. + */ + blockedProtocols?: string[] } /** @@ -542,6 +553,11 @@ export async function executeRebalanceIfNeeded( batchContext?.batchKey ?? `${currentProtocol}:${ctxStrategyName ?? 'DEFAULT'}:${ctxFollowedStrategyId ?? 'none'}` + // #345 — protocols with an OPEN circuit breaker are excluded as rebalance + // targets (their existing positions may still be rebalanced OUT, but the + // agent never moves money INTO a broken protocol). + const blockedProtocols = batchContext?.blockedProtocols ?? [] + const recordDecision = async (args: { outcome: 'REBALANCED' | 'HELD' | 'BLOCKED' blockedReason?: string | null @@ -637,7 +653,11 @@ export async function executeRebalanceIfNeeded( return null } - const allProtocols = await scanAllProtocols() + let allProtocols = await scanAllProtocols() + if (blockedProtocols.length > 0) { + const excluded = new Set(blockedProtocols) + allProtocols = allProtocols.filter((p) => !excluded.has(p.name)) + } if (allProtocols.length === 0) { logger.warn('No protocols available for comparison') const trace: DecisionTrace = { @@ -890,7 +910,8 @@ export async function executeRebalanceIfNeeded( const comparison = await compareProtocols( currentProtocol, totalAmount, - effectiveThresholds + effectiveThresholds, + blockedProtocols ) if (!comparison || !comparison.shouldRebalance) { diff --git a/src/config/env.ts b/src/config/env.ts index 708905e..8a2afcf 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -677,6 +677,68 @@ export const config = { lowDeferMs: parseInt(process.env.OUTBOX_LOW_DEFER_MS || '15000'), lowMaxDeferMs: parseInt(process.env.OUTBOX_LOW_MAX_DEFER_MS || '300000'), }, + /** + * Agent circuit breaker (#345) — pre-trade guards that halt agent-initiated + * rebalancing at GLOBAL / PROTOCOL / USER scope on abnormal loss, de-peg, + * oscillation or stale data. Each rule is independently toggleable and is + * evaluated by pure functions in src/agent/breakerRules.ts. The de-peg rule + * is disabled by default: it needs a live stablecoin price feed, and the + * platform currently has none wired (the fee oracle exposes fees, not + * prices), so enabling it without a real feed would never trip — leave off + * until an oracle path exists. + */ + breaker: { + /** Master switch for the whole circuit breaker. */ + enabled: (process.env.BREAKER_ENABLED ?? 'true').toLowerCase() === 'true', + abnormalLoss: { + enabled: + (process.env.BREAKER_ABNORMAL_LOSS_ENABLED ?? 'true').toLowerCase() === + 'true', + /** Portfolio down more than this percentage over the window trips. */ + lossPct: parseFloat(process.env.BREAKER_LOSS_PCT || '5'), + /** Trailing window (hours) over which drawdown is measured. */ + windowHours: parseInt(process.env.BREAKER_LOSS_WINDOW_HOURS || '24'), + }, + depeg: { + enabled: + (process.env.BREAKER_DEPEG_ENABLED ?? 'false').toLowerCase() === 'true', + /** Stablecoin price deviation from $1 (basis points) that trips. */ + depegBps: parseInt(process.env.BREAKER_DEPEG_BPS || '150'), + /** + * Consecutive clean evaluations required before an OPEN de-peg breaker + * may transition to HALF_OPEN — a single recovered tick is not enough. + */ + sustainedClearChecks: parseInt( + process.env.BREAKER_DEPEG_SUSTAINED_CHECKS || '3' + ), + /** Where the integration layer looks for the stablecoin spot price. */ + priceSource: process.env.BREAKER_DEPEG_PRICE_SOURCE || 'fee-oracle', + }, + oscillation: { + enabled: + (process.env.BREAKER_OSCILLATION_ENABLED ?? 'true').toLowerCase() === + 'true', + /** Rebalances of the same batch within the window that trip. */ + maxFlips: parseInt(process.env.BREAKER_MAX_FLIPS || '3'), + /** Sliding window (hours) for the flip count. */ + windowHours: parseInt(process.env.BREAKER_FLIP_WINDOW_HOURS || '24'), + }, + staleData: { + enabled: + (process.env.BREAKER_STALE_DATA_ENABLED ?? 'true').toLowerCase() === + 'true', + /** APY table older than this (minutes) must not be traded on. */ + staleMinutes: parseInt(process.env.BREAKER_STALE_MINUTES || '120'), + /** Consecutive scan failures before stale-data trips. */ + consecutiveFailures: parseInt( + process.env.BREAKER_STALE_CONSECUTIVE_FAILURES || '3' + ), + }, + /** Base cooldown before an OPEN breaker may auto-transition to HALF_OPEN. */ + cooldownMs: parseInt(process.env.BREAKER_COOLDOWN_MS || '3600000'), + /** Hard cap on cooldown after repeated re-tripping (backoff doubling). */ + maxCooldownMs: parseInt(process.env.BREAKER_MAX_COOLDOWN_MS || '86400000'), + }, feeOracle: { pollMs: parseInt(process.env.FEE_ORACLE_POLL_MS || '10000'), ttlMs: parseInt(process.env.FEE_ORACLE_TTL_MS || '30000'), diff --git a/src/events/types.ts b/src/events/types.ts index 05939d7..669b21f 100644 --- a/src/events/types.ts +++ b/src/events/types.ts @@ -39,6 +39,10 @@ export const SOCKET_ONLY_EVENT_TYPES = [ 'portfolio.updated', /** #343 — a rebalance decision was recorded; deep-linkable explanation. */ 'agent.decision_recorded', + /** #345 — the agent circuit breaker tripped; rebalancing paused (reason: plain text). */ + 'agent.circuit_breaker_tripped', + /** #345 — the agent circuit breaker reset; rebalancing may resume. */ + 'agent.circuit_breaker_reset', /** #374 — API key lifecycle notifications. */ 'security.api_key_changed', /** #376 — new session sign-in alert. */ @@ -77,6 +81,8 @@ export const EVENT_TYPE_TOPIC: Record = { 'approval.cancelled': 'transactions', 'agent.rebalanced': 'agent', 'agent.decision_recorded': 'agent', + 'agent.circuit_breaker_tripped': 'agent', + 'agent.circuit_breaker_reset': 'agent', 'alert_rule.triggered': 'alerts', 'strategy.updated': 'strategies', 'strategy.unpublished': 'strategies', diff --git a/src/routes/admin.ts b/src/routes/admin.ts index 1783872..4805a61 100644 --- a/src/routes/admin.ts +++ b/src/routes/admin.ts @@ -18,6 +18,11 @@ import { getAllProviderHealth, adminSetProviderCircuit } from '../fiat/registry' import db from '../db' import { alertingService } from '../services/alerting' import { verifyAuditChain } from '../audit/chain' +import { + listBreakers, + manualTripBreaker, + manualResetBreaker, +} from '../agent/breakerService' const router = Router() const prisma = db as any @@ -1561,4 +1566,150 @@ router.get( } ) +// ── Agent circuit breaker (#345) ───────────────────────────────────────────── + +/** + * GET /api/v1/admin/agent/breakers + * List every circuit breaker (GLOBAL / PROTOCOL / USER) with current state. + */ +router.get( + '/agent/breakers', + requireAdminScope('agent'), + async (req: Request, res: Response) => { + try { + const breakers = await listBreakers() + auditLog(req, res, 'LIST_AGENT_BREAKERS', 'success', { + count: breakers.length, + }) + res.status(200).json({ success: true, data: breakers }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'LIST_AGENT_BREAKERS', 'failure', { error: message }) + res.status(500).json({ success: false, error: message }) + } + } +) + +/** + * POST /api/v1/admin/agent/breakers + * Manually trip (open) a breaker: { scope, scopeKey?, reason }. + * GLOBAL has no scopeKey; PROTOCOL/USER require one. + */ +router.post( + '/agent/breakers', + requireAdminScope('agent'), + async (req: Request, res: Response) => { + try { + const { scope, scopeKey, reason } = req.body ?? {} + + if (!['GLOBAL', 'PROTOCOL', 'USER'].includes(scope)) { + auditLog(req, res, 'TRIP_AGENT_BREAKER', 'failure', { + error: 'invalid_scope', + }) + res + .status(400) + .json({ + success: false, + error: 'scope must be GLOBAL, PROTOCOL or USER', + }) + return + } + + if (scope !== 'GLOBAL') { + if (typeof scopeKey !== 'string' || scopeKey.trim() === '') { + auditLog(req, res, 'TRIP_AGENT_BREAKER', 'failure', { + error: 'missing_scope_key', + }) + res + .status(400) + .json({ + success: false, + error: 'scopeKey is required for PROTOCOL and USER trips', + }) + return + } + } + + if (typeof reason !== 'string' || reason.trim() === '') { + auditLog(req, res, 'TRIP_AGENT_BREAKER', 'failure', { + error: 'missing_reason', + }) + res.status(400).json({ success: false, error: 'reason is required' }) + return + } + + const adminIdentity = res.locals.adminAuth + ? `${res.locals.adminAuth.name} (${res.locals.adminAuth.role})` + : 'unknown' + const scopeKeyValue = scope === 'GLOBAL' ? '' : scopeKey.trim() + const result = await manualTripBreaker( + scope, + scopeKeyValue, + reason.trim(), + adminIdentity + ) + + auditLog(req, res, 'TRIP_AGENT_BREAKER', 'success', { + scope, + scopeKey: scopeKeyValue, + breakerId: result.id, + }) + res.status(200).json({ success: true, data: result }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'TRIP_AGENT_BREAKER', 'failure', { error: message }) + res.status(500).json({ success: false, error: message }) + } + } +) + +/** + * POST /api/v1/admin/agent/breakers/:id/reset + * Manually close (reset) a breaker: { reason }. + */ +router.post( + '/agent/breakers/:id/reset', + requireAdminScope('agent'), + async (req: Request, res: Response) => { + try { + const { reason } = req.body ?? {} + + if (typeof reason !== 'string' || reason.trim() === '') { + auditLog(req, res, 'RESET_AGENT_BREAKER', 'failure', { + error: 'missing_reason', + }) + res.status(400).json({ success: false, error: 'reason is required' }) + return + } + + const adminIdentity = res.locals.adminAuth + ? `${res.locals.adminAuth.name} (${res.locals.adminAuth.role})` + : 'unknown' + const result = await manualResetBreaker( + req.params.id, + reason.trim(), + adminIdentity + ) + + if (!result) { + auditLog(req, res, 'RESET_AGENT_BREAKER', 'failure', { + error: 'not_found', + id: req.params.id, + }) + res.status(404).json({ success: false, error: 'Breaker not found' }) + return + } + + auditLog(req, res, 'RESET_AGENT_BREAKER', 'success', { + breakerId: result.id, + }) + res.status(200).json({ success: true, data: result }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'RESET_AGENT_BREAKER', 'failure', { error: message }) + res.status(500).json({ success: false, error: message }) + } + } +) + export default router diff --git a/src/routes/agent.ts b/src/routes/agent.ts index 42434a0..e925fc7 100644 --- a/src/routes/agent.ts +++ b/src/routes/agent.ts @@ -3,7 +3,9 @@ */ import express, { Request, Response } from 'express' import { getAgentStatus } from '../agent/loop' +import { getBreakerStatusForUser } from '../agent/breakerService' import { internalAuthGuard } from '../middleware/authGuard' +import { requireAuth } from '../middleware/authenticate' const router = express.Router() @@ -31,6 +33,7 @@ router.get('/status', internalAuthGuard, (req: Request, res: Response) => { nextScheduledCheck: status.nextScheduledCheck, lastError: status.lastError, healthStatus: status.healthStatus, + breakers: status.breakers, timestamp: new Date().toISOString(), }, }) @@ -42,4 +45,33 @@ router.get('/status', internalAuthGuard, (req: Request, res: Response) => { } }) +/** + * GET /api/v1/agent/status/breakers + * Per-user breaker status: what applies to the calling player. Plain-language + * only — never other users' loss figures or thresholds. + * + * PROTECTED: Requires a user session. + */ +router.get( + '/status/breakers', + requireAuth, + async (req: Request, res: Response) => { + try { + const auth = req as Request & { auth?: { userId?: string } } + const userId = auth.auth?.userId + if (!userId) { + res.status(401).json({ success: false, error: 'Unauthorized' }) + return + } + const status = await getBreakerStatusForUser(userId) + res.json({ success: true, data: status }) + } catch (error) { + res.status(500).json({ + success: false, + error: error instanceof Error ? error.message : 'Unknown error', + }) + } + } +) + export default router diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts index 93de184..d47c115 100644 --- a/src/utils/metrics.ts +++ b/src/utils/metrics.ts @@ -125,6 +125,41 @@ export const agentRebalancesTriggeredTotal = new client.Counter({ registers: [register], }) +// #345 — agent circuit breaker observability. + +export const agentBreakerState = new client.Gauge({ + name: 'agent_breaker_state', + help: 'Agent circuit breaker state by scope and key: 0 closed, 1 half-open, 2 open', + labelNames: ['scope', 'scopeKey'] as const, + registers: [register], +}) + +export const agentBreakerTripsTotal = new client.Counter({ + name: 'agent_breaker_trips_total', + help: 'Agent circuit breaker trips by scope and rule', + labelNames: ['scope', 'rule'] as const, + registers: [register], +}) + +/** + * Record a breaker's current state for dashboards. + */ +export function setAgentBreakerState( + scope: string, + scopeKey: string, + state: 'CLOSED' | 'HALF_OPEN' | 'OPEN' +): void { + const value = state === 'CLOSED' ? 0 : state === 'HALF_OPEN' ? 1 : 2 + agentBreakerState.set({ scope, scopeKey }, value) +} + +/** + * Record a breaker trip (or re-trip). + */ +export function recordAgentBreakerTrip(scope: string, rule: string): void { + agentBreakerTripsTotal.inc({ scope, rule }) +} + export const agentSnapshotDuration = new client.Histogram({ name: 'agent_snapshot_duration_seconds', help: 'Duration of balance snapshot operations in seconds', diff --git a/tests/integration/agent/breakerLoop.integration.test.ts b/tests/integration/agent/breakerLoop.integration.test.ts new file mode 100644 index 0000000..26fb200 --- /dev/null +++ b/tests/integration/agent/breakerLoop.integration.test.ts @@ -0,0 +1,252 @@ +/** + * Integration test (#345): agent circuit breaker blocks rebalancing. + * + * - rebalanceCheckJob with a GLOBAL OPEN breaker writes a BLOCKED decision + * and never touches the money path (Stellar contract submit). + * - compareProtocols excludes an OPEN-breaker protocol as a target even when + * it is the highest-APY protocol (never move INTO a broken protocol). + */ +import { rebalanceCheckJob } from '../../../src/agent/loop' +import { compareProtocols } from '../../../src/agent/router' + +const mockTriggerRebalance = jest.fn() +jest.mock('../../../src/stellar/contract', () => ({ + triggerRebalance: (...args: unknown[]) => mockTriggerRebalance(...args), +})) + +const mockScanAllProtocols = jest.fn() +const mockGetCurrentOnChainApy = jest.fn() +jest.mock('../../../src/agent/scanner', () => ({ + scanAllProtocols: (...args: unknown[]) => mockScanAllProtocols(...args), + getCurrentOnChainApy: (...args: unknown[]) => + mockGetCurrentOnChainApy(...args), +})) + +jest.mock('../../../src/utils/logger', () => ({ + logger: { info: jest.fn(), warn: jest.fn(), error: jest.fn() }, + logBackgroundJob: jest.fn(), +})) + +jest.mock('../../../src/utils/correlation', () => { + const actual = jest.requireActual('../../../src/utils/correlation') + return { ...actual } +}) + +jest.mock('../../../src/strategy/service', () => ({ + loadActiveFollowsForUsers: jest.fn().mockResolvedValue(new Map()), +})) + +jest.mock('../../../src/events/publisher', () => ({ + publishUserEvent: jest.fn().mockResolvedValue(undefined), +})) + +jest.mock('../../../src/services/alerting', () => ({ + alertingService: { emit: jest.fn().mockResolvedValue(undefined) }, +})) + +jest.mock('../../../src/outbox/dispatcher', () => ({ + dispatchInBackground: jest.fn(), +})) + +jest.mock('../../../src/outbox/service', () => ({ + enqueueOutboxOp: jest.fn().mockResolvedValue('op-1'), + deriveIdempotencyKey: jest.fn().mockReturnValue('idem-1'), +})) + +const mockPositionFindMany = jest.fn() + +const decisionRows: Array = [] +const mockRebalanceDecisionCreate = jest + .fn() + .mockImplementation(({ data }: any) => { + decisionRows.push(data) + return Promise.resolve({ id: `decision-${decisionRows.length}` }) + }) +const mockRebalanceDecisionUpdate = jest.fn().mockResolvedValue({ id: 'd' }) +const mockRebalanceDecisionFindFirst = jest.fn().mockResolvedValue(null) +const mockRebalanceDecisionFindUnique = jest.fn().mockResolvedValue(null) +const mockRebalanceDecisionGroupBy = jest.fn().mockResolvedValue([]) + +const mockAgentBreakerFindMany = jest.fn() +const mockAgentBreakerUpdate = jest + .fn() + .mockImplementation(({ data }: any) => + Promise.resolve({ id: 'bk-g', ...data }) + ) +const mockAgentBreakerCreate = jest + .fn() + .mockImplementation(({ data }: any) => + Promise.resolve({ id: 'bk-new', ...data }) + ) +const mockAgentBreakerUpsert = jest + .fn() + .mockImplementation(() => Promise.resolve({ id: 'bk-g' })) + +const agentLogRows: Array = [] +const mockAgentLogCreate = jest.fn().mockImplementation(({ data }: any) => { + agentLogRows.push(data) + return Promise.resolve({ id: `log-${agentLogRows.length}` }) +}) + +jest.mock('../../../src/db', () => { + const client: any = { + position: { + findMany: (...args: unknown[]) => mockPositionFindMany(...args), + }, + yieldSnapshot: { + findMany: jest.fn().mockResolvedValue([]), + }, + rebalanceDecision: { + create: (...args: unknown[]) => mockRebalanceDecisionCreate(...args), + update: (...args: unknown[]) => mockRebalanceDecisionUpdate(...args), + findFirst: (...args: unknown[]) => + mockRebalanceDecisionFindFirst(...args), + findUnique: (...args: unknown[]) => + mockRebalanceDecisionFindUnique(...args), + groupBy: (...args: unknown[]) => mockRebalanceDecisionGroupBy(...args), + }, + protocolRate: { + findFirst: jest.fn().mockResolvedValue({ fetchedAt: new Date() }), + }, + agentCircuitBreaker: { + findMany: (...args: unknown[]) => mockAgentBreakerFindMany(...args), + update: (...args: unknown[]) => mockAgentBreakerUpdate(...args), + create: (...args: unknown[]) => mockAgentBreakerCreate(...args), + upsert: (...args: unknown[]) => mockAgentBreakerUpsert(...args), + }, + agentLog: { + create: (...args: unknown[]) => mockAgentLogCreate(...args), + }, + auditPayloadHash: { + create: jest.fn().mockResolvedValue({ id: 'hash-1' }), + }, + user: { + findMany: jest.fn().mockResolvedValue([]), + }, + } + client.$transaction = (fn: (tx: unknown) => unknown) => fn(client) + return { __esModule: true, default: client } +}) + +const PROTOCOLS = [ + { + name: 'protocol-b', + apy: 9.9, + assetSymbol: 'USDC', + lastUpdated: new Date(), + isAvailable: true, + }, + { + name: 'protocol-c', + apy: 8.0, + assetSymbol: 'USDC', + lastUpdated: new Date(), + isAvailable: true, + }, +] + +describe('Agent circuit breaker (#345)', () => { + beforeEach(() => { + jest.clearAllMocks() + decisionRows.length = 0 + agentLogRows.length = 0 + + mockScanAllProtocols.mockResolvedValue(PROTOCOLS.map((p) => ({ ...p }))) + mockGetCurrentOnChainApy.mockResolvedValue(5.0) + mockTriggerRebalance.mockResolvedValue({ hash: 'tx-hash-001' }) + + const now = new Date() + mockAgentBreakerFindMany.mockResolvedValue([ + { + id: 'bk-global', + scope: 'GLOBAL', + scopeKey: '', + state: 'OPEN', + trippedRule: 'stale_data', + detail: { rule: 'stale_data', at: now.toISOString() }, + trippedAt: new Date(now.getTime() - 10 * 60 * 1000), + autoResetAt: new Date(now.getTime() + 50 * 60 * 1000), + resetBy: null, + resetAt: null, + updatedAt: now, + }, + ]) + + mockPositionFindMany.mockResolvedValue([ + { + id: 'pos-1', + userId: 'user-1', + protocolName: 'protocol-a', + status: 'ACTIVE', + amount: '1000', + user: { id: 'user-1', rebalanceStrategy: null, strategyConfig: null }, + }, + { + id: 'pos-2', + userId: 'user-2', + protocolName: 'protocol-a', + status: 'ACTIVE', + amount: '2000', + user: { id: 'user-2', rebalanceStrategy: null, strategyConfig: null }, + }, + ]) + }) + + describe('rebalanceCheckJob with a GLOBAL OPEN breaker', () => { + it('writes a BLOCKED decision and never submits a rebalance', async () => { + await rebalanceCheckJob() + + // One batch (protocol-a:DEFAULT:none), one BLOCKED decision. + expect(decisionRows).toHaveLength(1) + expect(decisionRows[0]).toMatchObject({ + batchKey: 'protocol-a:DEFAULT:none', + fromProtocol: 'protocol-a', + outcome: 'BLOCKED', + blockedReason: 'circuit_breaker_open', + affectedPositions: 2, + }) + expect(String(decisionRows[0].rationale)).toContain('GLOBAL') + + // Money path untouched. + expect(mockTriggerRebalance).not.toHaveBeenCalled() + expect(mockAgentBreakerCreate).not.toHaveBeenCalled() + + // The tick succeeded and reported zero rebalances. + const analyze = agentLogRows.find( + (l) => l.action === 'ANALYZE' && l.status === 'SUCCESS' + ) + expect(analyze).toBeDefined() + const analyzeInput = JSON.parse(analyze.inputData) + expect(analyzeInput.rebalancesTriggered).toBe(0) + expect(analyzeInput.breakers.globalOpen).toBe(true) + }) + }) + + describe('compareProtocols target exclusion', () => { + it('never chooses an OPEN-breaker protocol as target', async () => { + const comparison = await compareProtocols( + 'protocol-a', + '1000', + undefined, + ['protocol-b'] + ) + expect(comparison).not.toBeNull() + expect(comparison!.best.name).toBe('protocol-c') + expect(comparison!.best.name).not.toBe('protocol-b') + expect(mockScanAllProtocols).toHaveBeenCalled() + }) + + it('falls back to the blocked protocol when everything is excluded', async () => { + mockScanAllProtocols.mockResolvedValue( + PROTOCOLS.slice(0, 1).map((p) => ({ ...p })) + ) + const comparison = await compareProtocols( + 'protocol-a', + '1000', + undefined, + ['protocol-b'] + ) + expect(comparison).toBeNull() + }) + }) +}) diff --git a/tests/integration/agent/strategy-follow.integration.test.ts b/tests/integration/agent/strategy-follow.integration.test.ts index 88354f1..e8e6c02 100644 --- a/tests/integration/agent/strategy-follow.integration.test.ts +++ b/tests/integration/agent/strategy-follow.integration.test.ts @@ -130,6 +130,18 @@ beforeEach(() => { findFirst: jest.fn(), findMany: jest.fn(), } + // #345 — breaker evaluation runs on every tick. With no OPEN breaker the + // evaluation must pass through as a closed circuit so batches proceed. + mockDb.agentCircuitBreaker = { + findMany: jest.fn().mockResolvedValue([]), + create: jest.fn().mockResolvedValue({ id: 'bk' }), + update: jest.fn().mockResolvedValue({ id: 'bk' }), + } + mockDb.protocolRate = { + findFirst: jest.fn().mockResolvedValue({ fetchedAt: new Date() }), + } + mockDb.yieldSnapshot = { findMany: jest.fn().mockResolvedValue([]) } + mockDb.rebalanceDecision = { groupBy: jest.fn().mockResolvedValue([]) } }) // ─── 2. No-follow regression ───────────────────────────────────────────────── diff --git a/tests/unit/agent/breakerRules.test.ts b/tests/unit/agent/breakerRules.test.ts new file mode 100644 index 0000000..ef94d33 --- /dev/null +++ b/tests/unit/agent/breakerRules.test.ts @@ -0,0 +1,371 @@ +// #345 — pure circuit-breaker trip rules: drawdown, de-peg, oscillation and +// stale-data. All deterministic; the service layer feeds these measurements. +import { + evaluateAbnormalLossRule, + evaluateDepegRule, + evaluateOscillationRule, + evaluateStaleDataRule, + evaluateBreakerRules, + validateBreakerConfig, + ValuePoint, + BreakerRuleConfig, +} from '../../../src/agent/breakerRules' + +const now = new Date('2026-09-01T12:00:00.000Z') +const HOUR = 60 * 60 * 1000 + +function pts(values: Array<[number, number]>): ValuePoint[] { + return values.map(([hoursAgo, value]) => ({ + at: new Date(now.getTime() - hoursAgo * HOUR), + value, + })) +} + +describe('breakerRules', () => { + describe('evaluateAbnormalLossRule', () => { + it('trips when window drawdown exceeds lossPct', () => { + const r = evaluateAbnormalLossRule({ + series: pts([ + [0, 92], + [2, 100], + ]), + lossPct: 5, + windowHours: 24, + now, + }) + expect(r.tripped).toBe(true) + expect(r.rule).toBe('abnormal_loss') + expect(r.detail.drawdownPct).toBeCloseTo(-8) + }) + + it('does not trip at or better than the threshold', () => { + const r = evaluateAbnormalLossRule({ + series: pts([ + [0, 100], + [2, 92], + ]), + lossPct: 5, + windowHours: 24, + now, + }) + expect(r.tripped).toBe(false) + }) + + it('needs at least two in-window points', () => { + const r = evaluateAbnormalLossRule({ + series: pts([[0, 50]]), + lossPct: 5, + windowHours: 24, + now, + }) + expect(r.tripped).toBe(false) + expect(r.detail.reason).toBe('insufficient_history') + }) + + it('ignores points outside the window', () => { + const r = evaluateAbnormalLossRule({ + series: pts([ + [25, 200], + [48, 200], + ]), + lossPct: 5, + windowHours: 24, + now, + }) + expect(r.tripped).toBe(false) + expect(r.detail.points).toBe(0) + }) + + it('fails safe on non-positive values', () => { + const r = evaluateAbnormalLossRule({ + series: [ + { at: now, value: 0 }, + { at: new Date(now.getTime() - 2 * HOUR), value: 100 }, + ], + lossPct: 5, + windowHours: 24, + now, + }) + expect(r.tripped).toBe(false) + expect(r.detail.reason).toBe('non_positive_peak') + }) + + it('sorts the series oldest-first regardless of input order', () => { + const r = evaluateAbnormalLossRule({ + series: [ + { at: now, value: 880 }, + { at: new Date(now.getTime() - 4 * HOUR), value: 920 }, + { at: new Date(now.getTime() - 2 * HOUR), value: 1000 }, + ], + lossPct: 5, + windowHours: 24, + now, + }) + expect(r.tripped).toBe(true) + expect(r.detail.peak).toBe(1000) + }) + }) + + describe('evaluateDepegRule', () => { + it('trips when deviation exceeds depegBps', () => { + const r = evaluateDepegRule({ price: 1.02, depegBps: 150 }) + expect(r.tripped).toBe(true) + expect(r.detail.deviationBps).toBeCloseTo(200) + }) + + it('does not trip within the band', () => { + expect(evaluateDepegRule({ price: 1.01, depegBps: 150 }).tripped).toBe( + false + ) + expect(evaluateDepegRule({ price: 0.99, depegBps: 150 }).tripped).toBe( + false + ) + }) + + it('fails safe on a null price (no feed)', () => { + const r = evaluateDepegRule({ price: null, depegBps: 150 }) + expect(r.tripped).toBe(false) + expect(r.detail.reason).toBe('no_price_feed') + }) + + it('fails safe on non-finite or non-positive prices', () => { + expect(evaluateDepegRule({ price: NaN, depegBps: 150 }).tripped).toBe( + false + ) + expect( + evaluateDepegRule({ price: Infinity, depegBps: 150 }).tripped + ).toBe(false) + expect(evaluateDepegRule({ price: 0, depegBps: 150 }).tripped).toBe(false) + }) + }) + + describe('evaluateOscillationRule', () => { + it('trips at maxFlips', () => { + expect(evaluateOscillationRule({ flips: 3, maxFlips: 3 }).tripped).toBe( + true + ) + }) + + it('does not trip below maxFlips', () => { + expect(evaluateOscillationRule({ flips: 2, maxFlips: 3 }).tripped).toBe( + false + ) + }) + }) + + describe('evaluateStaleDataRule', () => { + it('trips when consecutive failures hit the threshold', () => { + const r = evaluateStaleDataRule({ + latestFetchedAt: now, + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + consecutiveFailures: 3, + now, + }) + expect(r.tripped).toBe(true) + expect(r.detail.reason).toBe('consecutive_scan_failures') + }) + + it('a single failure does not trip', () => { + const r = evaluateStaleDataRule({ + latestFetchedAt: now, + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + consecutiveFailures: 1, + now, + }) + expect(r.tripped).toBe(false) + }) + + it('trips on never-scanned even with zero failures', () => { + const r = evaluateStaleDataRule({ + latestFetchedAt: null, + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + consecutiveFailures: 0, + now, + }) + expect(r.tripped).toBe(true) + expect(r.detail.reason).toBe('never_scanned') + }) + + it('trips when the last successful fetch is older than maxStaleMinutes', () => { + const r = evaluateStaleDataRule({ + latestFetchedAt: new Date(now.getTime() - 3 * 60 * 60 * 1000), + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + consecutiveFailures: 0, + now, + }) + expect(r.tripped).toBe(true) + }) + + it('does not trip while the table is fresh', () => { + const r = evaluateStaleDataRule({ + latestFetchedAt: new Date(now.getTime() - 10 * 60 * 1000), + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + consecutiveFailures: 0, + now, + }) + expect(r.tripped).toBe(false) + }) + }) + + describe('evaluateBreakerRules', () => { + const cfg: BreakerRuleConfig = { + abnormalLoss: { enabled: true, lossPct: 5, windowHours: 24 }, + depeg: { enabled: true, depegBps: 150 }, + oscillation: { enabled: true, maxFlips: 3 }, + staleData: { + enabled: true, + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + }, + } + + it('returns no-trip with reason none when everything clears', () => { + const r = evaluateBreakerRules(cfg, { + abnormalLossSeries: pts([ + [0, 100], + [2, 100], + ]), + depegPrice: 1.0, + oscillationFlips: 1, + latestFetchedAt: now, + consecutiveFailures: 0, + now, + }) + expect(r.tripped).toBe(false) + expect(r.detail.reason).toBe('none') + }) + + it('evaluates in order: abnormal_loss beats depeg', () => { + const r = evaluateBreakerRules(cfg, { + abnormalLossSeries: pts([ + [0, 80], + [2, 100], + ]), + depegPrice: 1.1, + oscillationFlips: 0, + latestFetchedAt: now, + consecutiveFailures: 0, + now, + }) + expect(r.rule).toBe('abnormal_loss') + }) + + it('skips disabled rules', () => { + const off: BreakerRuleConfig = { + abnormalLoss: { enabled: false, lossPct: 5, windowHours: 24 }, + depeg: { enabled: false, depegBps: 150 }, + oscillation: { enabled: false, maxFlips: 3 }, + staleData: { + enabled: false, + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + }, + } + const r = evaluateBreakerRules(off, { + abnormalLossSeries: pts([ + [0, 1], + [1, 0.5], + ]), + depegPrice: 0.5, + oscillationFlips: 99, + latestFetchedAt: null, + consecutiveFailures: 99, + now, + }) + expect(r.tripped).toBe(false) + }) + + it('skips abnormal_loss when no series is supplied', () => { + const r = evaluateBreakerRules(cfg, { + depegPrice: 1.0, + oscillationFlips: 0, + latestFetchedAt: now, + consecutiveFailures: 0, + now, + }) + expect(r.tripped).toBe(false) + }) + + it('de-peg trips when it is the first failing rule in order', () => { + const r = evaluateBreakerRules(cfg, { + abnormalLossSeries: pts([ + [0, 100], + [2, 100], + ]), + depegPrice: 1.02, + oscillationFlips: 0, + latestFetchedAt: now, + consecutiveFailures: 0, + now, + }) + expect(r.rule).toBe('depeg') + }) + }) + + describe('validateBreakerConfig', () => { + it('accepts a well-formed config', () => { + expect(() => + validateBreakerConfig({ + abnormalLoss: { enabled: true, lossPct: 5, windowHours: 24 }, + depeg: { enabled: true, depegBps: 150 }, + oscillation: { enabled: true, maxFlips: 3 }, + staleData: { + enabled: true, + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + }, + }) + ).not.toThrow() + }) + + it('rejects a non-positive lossPct when enabled', () => { + expect(() => { + validateBreakerConfig({ + abnormalLoss: { enabled: true, lossPct: 0, windowHours: 24 }, + depeg: { enabled: false, depegBps: 150 }, + oscillation: { enabled: false, maxFlips: 3 }, + staleData: { + enabled: false, + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + }, + }) + }).toThrow(/BREAKER_LOSS_PCT/) + }) + + it('rejects maxFlips below 2 when enabled', () => { + expect(() => { + validateBreakerConfig({ + abnormalLoss: { enabled: false, lossPct: 5, windowHours: 24 }, + depeg: { enabled: false, depegBps: 150 }, + oscillation: { enabled: true, maxFlips: 1 }, + staleData: { + enabled: false, + maxStaleMinutes: 120, + maxConsecutiveFailures: 3, + }, + }) + }).toThrow(/BREAKER_MAX_FLIPS/) + }) + + it('allows a bad value when the rule is disabled', () => { + expect(() => + validateBreakerConfig({ + abnormalLoss: { enabled: false, lossPct: 0, windowHours: 0 }, + depeg: { enabled: false, depegBps: -5 }, + oscillation: { enabled: false, maxFlips: 1 }, + staleData: { + enabled: false, + maxStaleMinutes: 0, + maxConsecutiveFailures: 0, + }, + }) + ).not.toThrow() + }) + }) +}) diff --git a/tests/unit/agent/breakerServiceStatus.test.ts b/tests/unit/agent/breakerServiceStatus.test.ts new file mode 100644 index 0000000..789c61b --- /dev/null +++ b/tests/unit/agent/breakerServiceStatus.test.ts @@ -0,0 +1,148 @@ +// #345 — breaker service status surfaces: user-facing status must never leak +// another user's (or any protocol's) breaker details, and the GLOBAL state is +// exposed in plain language only. +import { + getBreakerStatusForUser, + getBreakerStatusSummary, +} from '../../../src/agent/breakerService' + +jest.mock('../../../src/db', () => { + const mockFindMany = jest.fn() + const client: any = { + agentCircuitBreaker: { + findMany: (...args: unknown[]) => mockFindMany(...args), + }, + } + return { + __esModule: true, + default: client, + __mockFindMany: mockFindMany, + } +}) + +jest.mock('../../../src/utils/logger', () => ({ + logger: { info: jest.fn(), warn: jest.fn(), error: jest.fn() }, +})) +jest.mock('../../../src/events/publisher', () => ({ + publishUserEvent: jest.fn().mockResolvedValue(undefined), +})) +jest.mock('../../../src/services/alerting', () => ({ + alertingService: { emit: jest.fn().mockResolvedValue(undefined) }, +})) +jest.mock('../../../src/utils/metrics', () => ({ + setAgentBreakerState: jest.fn(), + recordAgentBreakerTrip: jest.fn(), +})) + +// eslint-disable-next-line @typescript-eslint/no-var-requires +const dbMock = require('../../../src/db') +const mockFindMany: jest.Mock = dbMock.__mockFindMany + +describe('breakerService status surfaces', () => { + beforeEach(() => { + mockFindMany.mockReset() + }) + + describe('getBreakerStatusForUser', () => { + it('reports only the calling user plus the GLOBAL state', async () => { + mockFindMany.mockResolvedValue([ + { + id: 'g', + scope: 'GLOBAL', + scopeKey: '', + state: 'OPEN', + trippedRule: 'stale_data', + }, + { + id: 'p', + scope: 'PROTOCOL', + scopeKey: 'protocol-b', + state: 'OPEN', + trippedRule: 'abnormal_loss', + }, + { + id: 'u-other', + scope: 'USER', + scopeKey: 'user-other', + state: 'OPEN', + trippedRule: 'oscillation', + }, + { + id: 'u-a', + scope: 'USER', + scopeKey: 'user-a', + state: 'HALF_OPEN', + trippedRule: 'abnormal_loss', + }, + { + id: 'u-a-closed', + scope: 'USER', + scopeKey: 'user-a', + state: 'CLOSED', + trippedRule: null, + }, + ]) + + const status = await getBreakerStatusForUser('user-a') + + // GLOBAL + the user's own breaker, plain language only. + expect(status).toEqual({ + global: 'stale_data', + affectingYou: ['rebalancing paused for your account: abnormal_loss'], + }) + const serialized = JSON.stringify(status) + expect(serialized).not.toContain('user-other') + expect(serialized).not.toContain('protocol-b') + expect(serialized).not.toContain('oscillation') + }) + + it('reports nothing when no open breaker applies', async () => { + mockFindMany.mockResolvedValue([ + { + id: 'g', + scope: 'GLOBAL', + scopeKey: '', + state: 'CLOSED', + trippedRule: null, + }, + { + id: 'u-a', + scope: 'USER', + scopeKey: 'user-a', + state: 'CLOSED', + trippedRule: null, + }, + ]) + await expect(getBreakerStatusForUser('user-a')).resolves.toEqual({ + global: null, + affectingYou: [], + }) + }) + }) + + describe('getBreakerStatusSummary', () => { + it('returns a closed (null) global summary before any tick', () => { + expect(getBreakerStatusSummary()).toEqual({ global: null }) + }) + }) + + describe('withdrawal independence (#345 acceptance criterion)', () => { + it('breaker code never gates or imports the withdrawal/outbox path', () => { + const fs = require('node:fs') + const path = require('node:path') + const fixtureDir = path.join(__dirname, '../../../src') + + const breakerSource = fs.readFileSync( + path.join(fixtureDir, 'agent/breakerService.ts'), + 'utf8' + ) + // A documented reference to the outbox loop gate is expected in the + // header comment; only a real import would couple the breaker to it. + const codeOnly = breakerSource.replace(/\/\*[\s\S]*?\*\//g, '') + + expect(codeOnly).not.toMatch(/from '.*\boutbox\b/) + expect(codeOnly).not.toMatch(/from '.*\bwithdraw\b/) + expect(codeOnly).not.toContain('isUserHalted') + }) + }) +}) diff --git a/tests/unit/agent/breakerState.test.ts b/tests/unit/agent/breakerState.test.ts new file mode 100644 index 0000000..d501d9e --- /dev/null +++ b/tests/unit/agent/breakerState.test.ts @@ -0,0 +1,252 @@ +// #345 — pure circuit-breaker state machine: CLOSED / OPEN / HALF_OPEN +// transitions, cooldown backoff, manual trip/reset, and the describe summary. +import { + applyBreakerEvaluation, + applyManualTrip, + applyManualReset, + describeBreaker, + BreakerRecord, + BreakerTransitionConfig, +} from '../../../src/agent/breakerState' + +const cfg: BreakerTransitionConfig = { + cooldownMs: 60 * 60 * 1000, + maxCooldownMs: 4 * 60 * 60 * 1000, + sustainedClearChecks: 3, +} + +const closed: BreakerRecord = { + state: 'CLOSED', + trippedRule: null, + detail: null, + trippedAt: null, + autoResetAt: null, +} + +const trip = { + tripped: true, + rule: 'abnormal_loss' as const, + detail: { drawdownPct: -7 }, +} + +const clean = { + tripped: false, + rule: 'stale_data' as const, + detail: { reason: 'none' }, +} + +function at(minutesAhead: number): Date { + return new Date(Date.UTC(2026, 8, 1, 12, 0, 0) + minutesAhead * 60 * 1000) +} + +function detailOf(r: BreakerRecord): Record { + return r.detail as Record +} + +describe('breakerState', () => { + describe('CLOSED', () => { + it('a clean evaluation leaves it closed', () => { + const r = applyBreakerEvaluation(closed, clean, at(0), cfg) + expect(r.state).toBe('CLOSED') + expect(r).toBe(closed) + }) + + it('a trip opens it with the base cooldown', () => { + const t0 = at(0) + const r = applyBreakerEvaluation(closed, trip, t0, cfg) + expect(r.state).toBe('OPEN') + expect(r.trippedRule).toBe('abnormal_loss') + expect(r.trippedAt?.getTime()).toBe(t0.getTime()) + expect(r.autoResetAt?.getTime()).toBe(t0.getTime() + cfg.cooldownMs) + expect(detailOf(r)._machine).toEqual({ + clearCount: 0, + cooldownMs: cfg.cooldownMs, + }) + }) + }) + + describe('OPEN', () => { + const openAt = at(0) + const open: BreakerRecord = applyBreakerEvaluation( + closed, + trip, + openAt, + cfg + ) + + it('stays OPEN on a trip and records the failed evaluation', () => { + const later = at(5) + const r = applyBreakerEvaluation(open, trip, later, cfg) + expect(r.state).toBe('OPEN') + expect(r.autoResetAt?.getTime()).toBe(openAt.getTime() + cfg.cooldownMs) + expect(detailOf(r).lastEvaluation.rule).toBe('abnormal_loss') + expect(detailOf(r).lastEvaluation.at).toBe(later.toISOString()) + }) + + it('stays OPEN while the cooldown has not elapsed', () => { + const r = applyBreakerEvaluation(open, clean, at(5), cfg) + expect(r.state).toBe('OPEN') + expect(detailOf(r)._machine.clearCount).toBe(1) + }) + + it('stays OPEN before sustainedClearChecks clean evaluations', () => { + const beforeCooldown = at(cfg.cooldownMs / 60000 - 10) + let r = open + for (let i = 0; i < cfg.sustainedClearChecks; i++) { + r = applyBreakerEvaluation( + r, + clean, + new Date(beforeCooldown.getTime() + i * 3 * 60000), + cfg + ) + } + expect(r.state).toBe('OPEN') + expect(detailOf(r)._machine.clearCount).toBe(cfg.sustainedClearChecks) + }) + + it('opens to HALF_OPEN once cooldown elapsed and clears are sustained', () => { + const afterCooldown = at(cfg.cooldownMs / 60000 + 1) + let r = open + for (let i = 1; i <= cfg.sustainedClearChecks; i++) { + r = applyBreakerEvaluation( + r, + clean, + new Date(afterCooldown.getTime() + i * 60000), + cfg + ) + } + expect(r.state).toBe('HALF_OPEN') + expect(detailOf(r)._machine.clearCount).toBe(cfg.sustainedClearChecks) + }) + }) + + describe('HALF_OPEN', () => { + function toHalfOpen(t: Date): BreakerRecord { + let r = applyBreakerEvaluation(closed, trip, t, cfg) + // Fast-forward past cooldown with sustained clean checks. + for (let i = 1; i <= cfg.sustainedClearChecks; i++) { + r = applyBreakerEvaluation( + r, + clean, + new Date(t.getTime() + cfg.cooldownMs + i * 60000), + cfg + ) + } + expect(r.state).toBe('HALF_OPEN') + return r + } + + it('a clean probe closes the breaker', () => { + const r = applyBreakerEvaluation(toHalfOpen(at(0)), clean, at(80), cfg) + expect(r.state).toBe('CLOSED') + expect(r.trippedRule).toBeNull() + expect(r.trippedAt).toBeNull() + expect(r.autoResetAt).toBeNull() + expect(detailOf(r)._machine).toEqual({ + clearCount: 0, + cooldownMs: cfg.cooldownMs, + }) + }) + + it('a failed probe re-opens with doubled cooldown', () => { + const t0 = at(0) + const half = toHalfOpen(t0) + const probeAt = at(80) + const r = applyBreakerEvaluation(half, trip, probeAt, cfg) + expect(r.state).toBe('OPEN') + expect(detailOf(r)._machine.cooldownMs).toBe(cfg.cooldownMs * 2) + expect(r.autoResetAt?.getTime()).toBe( + probeAt.getTime() + cfg.cooldownMs * 2 + ) + }) + + it('cooldown doubling is capped at maxCooldownMs', () => { + let r: BreakerRecord = { ...closed } + const t0 = at(0) + let t = new Date(t0.getTime()) + for (let i = 0; i < 6; i++) { + r = applyBreakerEvaluation(r, trip, t, cfg) + expect(r.state).toBe('OPEN') + // Advance past the (possibly backed-off) cooldown, then sustain. + const cooldown = detailOf(r)._machine.cooldownMs + t = new Date(t.getTime() + cooldown) + for (let j = 1; j <= cfg.sustainedClearChecks; j++) { + r = applyBreakerEvaluation( + r, + clean, + new Date(t.getTime() + j * 60000), + cfg + ) + } + expect(r.state).toBe('HALF_OPEN') + t = new Date(t.getTime() + (cfg.sustainedClearChecks + 1) * 60000) + } + expect(detailOf(r)._machine.cooldownMs).toBe(cfg.maxCooldownMs) + }) + }) + + describe('applyManualTrip', () => { + it('opens a closed breaker with rule manual', () => { + const r = applyManualTrip(closed, at(0), cfg) + expect(r.state).toBe('OPEN') + expect(r.trippedRule).toBe('manual') + expect(r.autoResetAt?.getTime()).toBe(at(0).getTime() + cfg.cooldownMs) + }) + + it('is a no-op when already manually open', () => { + const t = at(0) + const manual = applyManualTrip(closed, t, cfg) + const again = applyManualTrip(manual, at(1), cfg) + expect(again).toBe(manual) + }) + + it('overrides a rule trip with manual', () => { + const r = applyManualTrip( + applyBreakerEvaluation(closed, trip, at(0), cfg), + at(1), + cfg + ) + expect(r.trippedRule).toBe('manual') + }) + }) + + describe('applyManualReset', () => { + it('closes any state and clears trip fields', () => { + const opened = applyBreakerEvaluation(closed, trip, at(0), cfg) + const r = applyManualReset( + opened, + at(5), + 'Nadia (super)', + 'market recovered' + ) + expect(r.state).toBe('CLOSED') + expect(r.trippedRule).toBeNull() + expect(r.trippedAt).toBeNull() + expect(r.autoResetAt).toBeNull() + expect(detailOf(r)._lastReset).toMatchObject({ + resetBy: 'Nadia (super)', + reason: 'market recovered', + }) + expect(detailOf(r).rule).toBeDefined() + }) + }) + + describe('describeBreaker', () => { + it('reports state, cooldown and clear count', () => { + const opened = applyBreakerEvaluation(closed, trip, at(0), cfg) + const d = describeBreaker(opened) + expect(d.state).toBe('OPEN') + expect(d.trippedRule).toBe('abnormal_loss') + expect(d.cooldownMs).toBe(cfg.cooldownMs) + expect(d.clearCount).toBe(0) + expect(d.autoResetAt).toBe(opened.autoResetAt?.toISOString()) + }) + + it('survives a missing machine section', () => { + const d = describeBreaker(applyManualReset(closed, at(0), 'a', 'r')) + expect(d.state).toBe('CLOSED') + expect(d.cooldownMs).toBe(0) + expect(d.clearCount).toBe(0) + }) + }) +}) From 35b83e21e5f4da636c82ce6f3fc1dba0afa6c4ca Mon Sep 17 00:00:00 2001 From: WEB3NOVA Date: Tue, 1 Sep 2026 10:03:37 +0100 Subject: [PATCH 2/2] style(admin): apply prettier formatting --- src/routes/admin.ts | 20 ++++++++------------ 1 file changed, 8 insertions(+), 12 deletions(-) diff --git a/src/routes/admin.ts b/src/routes/admin.ts index 4805a61..e42f9c7 100644 --- a/src/routes/admin.ts +++ b/src/routes/admin.ts @@ -1606,12 +1606,10 @@ router.post( auditLog(req, res, 'TRIP_AGENT_BREAKER', 'failure', { error: 'invalid_scope', }) - res - .status(400) - .json({ - success: false, - error: 'scope must be GLOBAL, PROTOCOL or USER', - }) + res.status(400).json({ + success: false, + error: 'scope must be GLOBAL, PROTOCOL or USER', + }) return } @@ -1620,12 +1618,10 @@ router.post( auditLog(req, res, 'TRIP_AGENT_BREAKER', 'failure', { error: 'missing_scope_key', }) - res - .status(400) - .json({ - success: false, - error: 'scopeKey is required for PROTOCOL and USER trips', - }) + res.status(400).json({ + success: false, + error: 'scopeKey is required for PROTOCOL and USER trips', + }) return } }