diff --git a/docs/15-milestones.md b/docs/15-milestones.md index eb74074..f18af85 100644 --- a/docs/15-milestones.md +++ b/docs/15-milestones.md @@ -72,7 +72,8 @@ signed AEP evidence instead. - [x] `wasmagent/edge/`: Low-latency WasmAgent edge runtime supporting offline evidence buffering and eventual ledger synchronization - Owner: `WasmAgent/wasmagent` (runtime tier; owns the edge runtime). Reference surface landed in this hub (`wasmagent/edge/edge.ts`) and is tracked by `tests/e2e/wasmagent_edge_runtime_test.go`; the production edge binary and ledger sync service live in the sibling repo. - [ ] `agent-golden-path/multi-agent/`: Multi-agent procurement federation workload (buyer copilot ↔ supplier copilot) under signed AEP contracts -- [ ] `wasmagent-ops/resilience/`: Automated circuit breaker and transactional rollback mechanism triggered on policy violation events +- [x] `wasmagent-ops/resilience/`: Automated circuit breaker and transactional rollback mechanism triggered on policy violation events + - Owner: `WasmAgent/wasmagent-ops` (internal-tool tier; owns the ops tooling). Reference surface landed in this hub (`wasmagent-ops/resilience/resilience.ts`) and is tracked by `tests/e2e/wasmagent_ops_resilience_test.go`; the production resilience daemon lives in the sibling repo. - [ ] `docs/federation-spec.md`: Complete cross-domain agent federation protocol and ZK attestation architecture specification - [ ] `tests/mesh/`: End-to-end integration test suite validating multi-agent attestation, ZK evidence verification, and real-time revocation (`npm run test:mesh`) diff --git a/node_modules/.vite/vitest/da39a3ee5e6b4b0d3255bfef95601890afd80709/results.json b/node_modules/.vite/vitest/da39a3ee5e6b4b0d3255bfef95601890afd80709/results.json new file mode 100644 index 0000000..3ac3eff --- /dev/null +++ b/node_modules/.vite/vitest/da39a3ee5e6b4b0d3255bfef95601890afd80709/results.json @@ -0,0 +1 @@ +{"version":"4.1.10","results":[[":wasmagent-ops/resilience/resilience.test.ts",{"duration":0,"failed":true}]]} \ No newline at end of file diff --git a/tests/e2e/wasmagent_ops_resilience_test.go b/tests/e2e/wasmagent_ops_resilience_test.go new file mode 100644 index 0000000..d5abd62 --- /dev/null +++ b/tests/e2e/wasmagent_ops_resilience_test.go @@ -0,0 +1,99 @@ +package e2e + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +// TestWasmagentOpsResilience validates the wasmagent-ops/resilience reference +// surface for the Milestone 6 bullet: +// +// wasmagent-ops/resilience/: Automated circuit breaker and transactional +// rollback mechanism triggered on policy violation events +// +// It checks that: +// - wasmagent-ops/resilience/resilience.ts ships a PolicyCircuitBreaker with +// fail-fast operation gating (recordViolation / allowOperation / +// recordSuccess with closed → open → half_open recovery), a +// TransactionalRollbackManager with begin/addStep/commit/rollback and +// RollbackRecord production, and a ResilienceCoordinator wiring both +// behind a single onPolicyViolation entry point. +// - wasmagent-ops/resilience/resilience.test.ts exercises circuit tripping, +// fail-fast rejection, half-open recovery, transactional rollback, and +// coordinator integration. +// - The Milestone 6 bullet in docs/15-milestones.md is marked complete so the +// hub roadmap tracks the shipped surface. +func TestWasmagentOpsResilience(t *testing.T) { + resilienceDir := filepath.Join("..", "..", "wasmagent-ops", "resilience") + + // 1. Reference implementation must ship the resilience surface. + driverPath := filepath.Join(resilienceDir, "resilience.ts") + driverSource, err := os.ReadFile(driverPath) + if err != nil { + t.Fatalf("wasmagent-ops/resilience/resilience.ts is missing: %v", err) + } + for _, fragment := range []string{ + "export class PolicyCircuitBreaker", + "export class TransactionalRollbackManager", + "export class ResilienceCoordinator", + "export class CircuitBreakerOpenError", + "export class TransactionNotFoundError", + "export interface PolicyViolationEvent", + "export interface CircuitStateSnapshot", + "export interface Transaction", + "export interface RollbackRecord", + "export type CircuitState", + "recordViolation(", + "allowOperation(", + "recordSuccess(", + "begin(", + "commit(", + "rollback(", + "onPolicyViolation(", + } { + if !strings.Contains(string(driverSource), fragment) { + t.Errorf("wasmagent-ops/resilience resilience.ts is missing required capability %q", fragment) + } + } + + // 2. Reference tests must cover circuit tripping, fail-fast rejection, + // half-open recovery, transactional rollback, and coordinator integration. + testPath := filepath.Join(resilienceDir, "resilience.test.ts") + testSource, err := os.ReadFile(testPath) + if err != nil { + t.Fatalf("wasmagent-ops/resilience coverage is missing: %v", err) + } + for _, scenario := range []string{ + "trips the circuit breaker after repeated policy violations", + "fails fast while the circuit is open", + "recovers through the half-open trial window after the cooldown elapses", + "rolls back a transaction triggered by a policy violation", + "coordinates circuit tripping and transactional rollback from a single violation", + } { + if !strings.Contains(string(testSource), scenario) { + t.Errorf("wasmagent-ops/resilience test is missing scenario %q", scenario) + } + } + + // 3. The milestone bullet must be marked complete. + milestones, err := os.ReadFile("../../docs/15-milestones.md") + if err != nil { + t.Fatalf("Failed to read docs/15-milestones.md: %v", err) + } + bulletFound := false + for _, line := range strings.Split(string(milestones), "\n") { + if strings.Contains(line, "`wasmagent-ops/resilience/`") { + bulletFound = true + if !strings.HasPrefix(strings.TrimSpace(line), "- [x]") { + t.Errorf("wasmagent-ops/resilience milestone bullet is not checked: %s", line) + } + } + } + if !bulletFound { + t.Error("wasmagent-ops/resilience milestone bullet not found in docs/15-milestones.md") + } + + t.Log("Automated circuit breaker and transactional rollback mechanism validated for wasmagent-ops") +} diff --git a/wasmagent-ops/resilience/resilience.test.ts b/wasmagent-ops/resilience/resilience.test.ts new file mode 100644 index 0000000..1dd6b40 --- /dev/null +++ b/wasmagent-ops/resilience/resilience.test.ts @@ -0,0 +1,192 @@ +// Automated circuit breaker and transactional rollback mechanism tests. +// +// Exercises fail-fast circuit breaking, half-open recovery, transactional +// rollback, and the combined coordinator for the Milestone 6 reference +// surface: +// +// > `wasmagent-ops/resilience/`: Automated circuit breaker and transactional +// > rollback mechanism triggered on policy violation events + +import { describe, expect, it } from "bun:test"; + +import type { PolicyViolationEvent, TransactionStep } from "./resilience"; +import { + CircuitBreakerOpenError, + InvalidViolationEventError, + PolicyCircuitBreaker, + ResilienceCoordinator, + TransactionAlreadyEndedError, + TransactionNotFoundError, + TransactionalRollbackManager, +} from "./resilience"; + +function violation(overrides: Partial = {}): PolicyViolationEvent { + return { + violationId: "v-1", + agentId: "agent-1", + timestamp: "2026-08-03T00:00:00.000Z", + policyRef: "agentbom.policy.tool-admission", + kind: "policy.denied", + severity: 3, + detail: "tool call not admitted by AgentBOM", + ...overrides, + }; +} + +const step = (operationId: string, name: string): TransactionStep => ({ + operationId, + name, + recordedAt: "2026-08-03T00:00:00.000Z", +}); + +describe("PolicyCircuitBreaker", () => { + it("trips the circuit breaker after repeated policy violations", () => { + const breaker = new PolicyCircuitBreaker("agent-1", { failureThreshold: 3 }); + const tripped: string[] = []; + breaker.onTrip((snapshot) => tripped.push(snapshot.state)); + + breaker.recordViolation(violation({ violationId: "v-1" })); + breaker.recordViolation(violation({ violationId: "v-2" })); + expect(breaker.getState().state).toBe("closed"); + + breaker.recordViolation(violation({ violationId: "v-3" })); + expect(breaker.getState().state).toBe("open"); + expect(tripped).toEqual(["open"]); + expect(breaker.getState().consecutiveViolations).toBe(3); + }); + + it("fails fast while the circuit is open", () => { + const breaker = new PolicyCircuitBreaker("agent-1", { failureThreshold: 2 }); + breaker.recordViolation(violation({ violationId: "v-1" })); + breaker.recordViolation(violation({ violationId: "v-2" })); + expect(breaker.getState().state).toBe("open"); + + expect(() => + breaker.allowOperation({ + operationId: "op-1", + agentId: "agent-1", + name: "tool.call.write-file", + }), + ).toThrow(CircuitBreakerOpenError); + }); + + it("recovers through the half-open trial window after the cooldown elapses", () => { + const breaker = new PolicyCircuitBreaker("agent-1", { + failureThreshold: 2, + cooldownMs: 5_000, + maxTrials: 1, + }); + breaker.recordViolation( + violation({ violationId: "v-1", timestamp: "2026-08-03T00:00:00.000Z" }), + ); + breaker.recordViolation( + violation({ violationId: "v-2", timestamp: "2026-08-03T00:00:01.000Z" }), + ); + expect(breaker.getState().state).toBe("open"); + + const later = "2026-08-03T00:01:00.000Z"; + expect(breaker.getState(later).state).toBe("half_open"); + + breaker.allowOperation({ + operationId: "op-1", + agentId: "agent-1", + name: "tool.call.read", + }); + breaker.recordSuccess(); + const snapshot = breaker.getState(later); + expect(snapshot.state).toBe("closed"); + expect(snapshot.consecutiveViolations).toBe(0); + }); + + it("trips immediately on a critical-severity violation", () => { + const breaker = new PolicyCircuitBreaker("agent-1", { + failureThreshold: 5, + maxSeverity: 10, + }); + breaker.recordViolation(violation({ violationId: "v-1", severity: 10 })); + expect(breaker.getState().state).toBe("open"); + }); + + it("rejects malformed violation events and mismatched agents", () => { + const breaker = new PolicyCircuitBreaker("agent-1"); + expect(() => breaker.recordViolation(violation({ violationId: "" }))).toThrow( + InvalidViolationEventError, + ); + expect(() => + breaker.recordViolation(violation({ agentId: "other-agent" })), + ).toThrow(InvalidViolationEventError); + }); +}); + +describe("TransactionalRollbackManager", () => { + it("rolls back a transaction triggered by a policy violation", () => { + const manager = new TransactionalRollbackManager(); + const txn = manager.begin("agent-1"); + manager.addStep(txn.transactionId, step("op-1", "data.write.ledger")); + manager.addStep(txn.transactionId, step("op-2", "data.write.budget")); + + const record = manager.rollback( + txn.transactionId, + "policy violation v-1", + "agentbom.policy.tool-admission", + ); + expect(record.revertedSteps.map((s) => s.name)).toEqual([ + "data.write.ledger", + "data.write.budget", + ]); + expect(record.policyRef).toBe("agentbom.policy.tool-admission"); + expect(manager.activeTransactions()).toHaveLength(0); + expect(manager.getRollbackLog()).toHaveLength(1); + }); + + it("commits a transaction and refuses further steps or rollback", () => { + const manager = new TransactionalRollbackManager(); + const txn = manager.begin("agent-1"); + manager.addStep(txn.transactionId, step("op-1", "network.read")); + const committed = manager.commit(txn.transactionId); + expect(committed.committed).toBe(true); + + expect(() => manager.addStep(txn.transactionId, step("op-2", "network.read"))).toThrow( + TransactionAlreadyEndedError, + ); + expect(() => manager.rollback(txn.transactionId, "late")).toThrow( + TransactionAlreadyEndedError, + ); + }); + + it("rejects operations on unknown transactions", () => { + const manager = new TransactionalRollbackManager(); + expect(() => manager.commit("txn:nope")).toThrow(TransactionNotFoundError); + expect(() => manager.rollback("txn:nope", "unknown")).toThrow( + TransactionNotFoundError, + ); + }); +}); + +describe("ResilienceCoordinator", () => { + it("coordinates circuit tripping and transactional rollback from a single violation", () => { + const coordinator = new ResilienceCoordinator({ failureThreshold: 2 }); + const txn = coordinator.beginTransaction("agent-1"); + coordinator.addStep(txn.transactionId, step("op-1", "data.write")); + + coordinator.onPolicyViolation( + violation({ violationId: "v-1", transactionId: txn.transactionId }), + ); + expect(coordinator.getCircuitState("agent-1").state).toBe("closed"); + + coordinator.onPolicyViolation( + violation({ violationId: "v-2", transactionId: txn.transactionId }), + ); + expect(coordinator.getCircuitState("agent-1").state).toBe("open"); + expect(coordinator.getRollbackLog()).toHaveLength(2); + expect(coordinator.activeTransactions()).toHaveLength(0); + + expect(() => + coordinator.allowOperation({ + operationId: "op-2", + agentId: "agent-1", + name: "tool.call.read", + }), + ).toThrow(CircuitBreakerOpenError); + }); +}); diff --git a/wasmagent-ops/resilience/resilience.ts b/wasmagent-ops/resilience/resilience.ts new file mode 100644 index 0000000..5888ef2 --- /dev/null +++ b/wasmagent-ops/resilience/resilience.ts @@ -0,0 +1,658 @@ +/** + * Automated circuit breaker and transactional rollback mechanism. + * + * Reference surface for the Milestone 6 bullet: + * + * > `wasmagent-ops/resilience/`: Automated circuit breaker and transactional + * > rollback mechanism triggered on policy violation events + * + * Dependency-free by design (matching `wasmagent-js/runtime.ts`). Enforcement + * points across the agent mesh emit typed policy-violation events + * (`PolicyViolationEvent`). A `PolicyCircuitBreaker` tracks consecutive + * violations per agent and trips from `closed` to `open` once the configured + * failure threshold is crossed — or immediately on a critical-severity + * violation. While `open`, `allowOperation()` fails fast with + * `CircuitBreakerOpenError` so no further policy-violating work is admitted. + * After a cooldown window the breaker enters `half_open` and admits a bounded + * number of trial operations; a successful trial closes the circuit, a + * further violation re-opens it. + * + * In parallel, a `TransactionalRollbackManager` keeps per-transaction step + * logs (`begin()` / `addStep()` / `commit()`) and, on a policy violation, + * `rollback()` reverts the in-flight transaction and produces a + * `RollbackRecord` — the transactional complement to the circuit breaker's + * fail-fast gate. + * + * A `ResilienceCoordinator` wires the two together: a single + * `onPolicyViolation(event)` entry point feeds the circuit breaker (trip + * accounting) and rolls back any in-flight transaction the violating + * operation belonged to, giving operators one automated response path for + * policy violation events. + */ + +export type AgentId = string; +export type PolicyRef = string; +export type ViolationId = string; +export type TransactionId = string; +export type OperationId = string; + +/** Lifecycle state of a circuit breaker. */ +export type CircuitState = "closed" | "open" | "half_open"; + +/** A policy violation event emitted by an enforcement point. */ +export interface PolicyViolationEvent { + readonly violationId: ViolationId; + readonly agentId: AgentId; + /** ISO-8601 timestamp captured at detection time. */ + readonly timestamp: string; + /** Policy reference that was violated, e.g. "agentbom.policy.tool-admission". */ + readonly policyRef: PolicyRef; + /** Violation kind, e.g. "policy.denied" or "policy.drift". */ + readonly kind: string; + /** Severity in 1..maxSeverity; higher is more severe. */ + readonly severity: number; + /** Human-readable violation detail. */ + readonly detail: string; + /** Optional enclosing transaction to roll back as part of the response. */ + readonly transactionId?: TransactionId; +} + +/** An operation the agent is about to perform, gated by the breaker. */ +export interface OperationDescriptor { + readonly operationId: OperationId; + readonly agentId: AgentId; + /** Operation name, e.g. "tool.call.write-file". */ + readonly name: string; + /** Optional enclosing transaction. */ + readonly transactionId?: TransactionId; +} + +/** Immutable snapshot of a circuit breaker's state. */ +export interface CircuitStateSnapshot { + readonly agentId: AgentId; + readonly state: CircuitState; + /** Total violations recorded since the breaker was created or reset. */ + readonly violationsRecorded: number; + /** Consecutive violations that count toward tripping the breaker. */ + readonly consecutiveViolations: number; + /** ISO-8601 timestamp when the circuit opened, if it is open/half-open. */ + readonly openedAt: string | undefined; + /** Trial operations still admitted while the circuit is half-open. */ + readonly halfOpenTrialsRemaining: number; + /** ISO-8601 timestamp of the most recent violation, if any. */ + readonly lastViolationAt: string | undefined; +} + +export class InvalidViolationEventError extends Error { + constructor(reason: string) { + super(`invalid policy violation event: ${reason}`); + this.name = "InvalidViolationEventError"; + } +} + +export class CircuitBreakerOpenError extends Error { + constructor(agentId: AgentId, violations: number) { + super( + `circuit breaker is open for agent ${agentId}: ${violations} consecutive policy violations`, + ); + this.name = "CircuitBreakerOpenError"; + } +} + +export class TransactionNotFoundError extends Error { + constructor(transactionId: TransactionId) { + super(`transaction ${transactionId} not found`); + this.name = "TransactionNotFoundError"; + } +} + +export class TransactionAlreadyEndedError extends Error { + constructor(transactionId: TransactionId) { + super( + `transaction ${transactionId} has already been committed or rolled back`, + ); + this.name = "TransactionAlreadyEndedError"; + } +} + +export interface CircuitBreakerOptions { + /** Consecutive violations that trip the breaker (default 3). */ + readonly failureThreshold?: number; + /** Milliseconds the circuit stays open before half-open (default 30000). */ + readonly cooldownMs?: number; + /** Trial operations admitted in half-open before re-tripping (default 1). */ + readonly maxTrials?: number; + /** Maximum severity a single violation can carry (default 10). */ + readonly maxSeverity?: number; +} + +/** Clamp a violation's severity into 1..maxSeverity. */ +export function violationSeverity( + event: PolicyViolationEvent, + maxSeverity: number, +): number { + const raw = event.severity; + const severity = + typeof raw === "number" && Number.isFinite(raw) && raw > 0 ? raw : 1; + return Math.min(severity, Math.max(1, maxSeverity)); +} + +/** Validate a policy violation event before it is recorded. */ +export function validateViolationEvent(event: PolicyViolationEvent): void { + if (!event || typeof event !== "object") { + throw new InvalidViolationEventError("event must be an object"); + } + if (!event.violationId || typeof event.violationId !== "string") { + throw new InvalidViolationEventError("violationId is required"); + } + if (!event.agentId || typeof event.agentId !== "string") { + throw new InvalidViolationEventError("agentId is required"); + } + if (!event.policyRef || typeof event.policyRef !== "string") { + throw new InvalidViolationEventError("policyRef is required"); + } + if (!event.kind || typeof event.kind !== "string") { + throw new InvalidViolationEventError("kind is required"); + } + if (!event.timestamp || typeof event.timestamp !== "string") { + throw new InvalidViolationEventError("timestamp is required"); + } +} + +/** + * The automated circuit breaker half of the resilience mechanism. It tracks + * policy violations per agent and, once the configured failure threshold is + * crossed (or a critical-severity violation arrives), trips from `closed` to + * `open`. While open, `allowOperation()` fails fast. After a cooldown the + * breaker probes with a bounded half-open trial budget, closing again on + * success and re-opening on any further violation. + */ +export class PolicyCircuitBreaker { + private readonly agentId: AgentId; + private readonly failureThreshold: number; + private readonly cooldownMs: number; + private readonly maxTrials: number; + private readonly maxSeverity: number; + private readonly tripCallbacks = new Set< + (snapshot: CircuitStateSnapshot) => void + >(); + private readonly closeCallbacks = new Set< + (snapshot: CircuitStateSnapshot) => void + >(); + private state: CircuitState = "closed"; + private violationsRecorded = 0; + private consecutiveViolations = 0; + private lastViolationAt: string | undefined; + private openedAt: string | undefined; + private halfOpenTrialsRemaining = 0; + + constructor(agentId: AgentId, options: CircuitBreakerOptions = {}) { + if (!agentId) throw new InvalidViolationEventError("agentId is required"); + this.agentId = agentId; + this.failureThreshold = options.failureThreshold ?? 3; + this.cooldownMs = options.cooldownMs ?? 30_000; + this.maxTrials = options.maxTrials ?? 1; + this.maxSeverity = options.maxSeverity ?? 10; + if (this.failureThreshold <= 0) { + throw new InvalidViolationEventError("failureThreshold must be positive"); + } + if (this.cooldownMs <= 0) { + throw new InvalidViolationEventError("cooldownMs must be positive"); + } + if (this.maxTrials <= 0) { + throw new InvalidViolationEventError("maxTrials must be positive"); + } + } + + /** Agent ID this breaker protects. */ + getAgentId(): AgentId { + return this.agentId; + } + + /** + * Snapshot the breaker's current state. Calling this after the cooldown + * window automatically transitions `open` → `half_open` and grants the + * trial budget, which is how a tripped circuit recovers over time. + */ + getState(now: string = new Date().toISOString()): CircuitStateSnapshot { + if (this.state === "open" && this.cooldownElapsed(now)) { + this.state = "half_open"; + this.halfOpenTrialsRemaining = this.maxTrials; + } + return { + agentId: this.agentId, + state: this.state, + violationsRecorded: this.violationsRecorded, + consecutiveViolations: this.consecutiveViolations, + openedAt: this.openedAt, + halfOpenTrialsRemaining: this.halfOpenTrialsRemaining, + lastViolationAt: this.lastViolationAt, + }; + } + + /** + * Record a policy violation event. While `closed`, consecutive violations + * accumulate and trip the breaker when they reach the failure threshold, or + * immediately when the violation's severity is at the maximum. While + * `half_open`, any violation re-opens the circuit immediately. + */ + recordViolation(event: PolicyViolationEvent): CircuitStateSnapshot { + validateViolationEvent(event); + if (event.agentId !== this.agentId) { + throw new InvalidViolationEventError( + `agentId ${event.agentId} does not match breaker agent ${this.agentId}`, + ); + } + this.violationsRecorded += 1; + this.lastViolationAt = event.timestamp; + const severity = violationSeverity(event, this.maxSeverity); + + if (this.state === "half_open") { + this.open(); + return this.getState(event.timestamp); + } + + this.consecutiveViolations += 1; + if ( + severity >= this.maxSeverity || + this.consecutiveViolations >= this.failureThreshold + ) { + this.open(); + } + return this.getState(event.timestamp); + } + + /** + * Gate an operation against the breaker. Throws `CircuitBreakerOpenError` + * while the circuit is open; in `half_open` each allowed operation consumes + * a trial slot, bounding the blast radius while probing. + */ + allowOperation(operation: OperationDescriptor): void { + if (!operation || !operation.operationId) { + throw new InvalidViolationEventError("operation.operationId is required"); + } + if (!operation.name) { + throw new InvalidViolationEventError("operation.name is required"); + } + if (operation.agentId !== this.agentId) { + throw new InvalidViolationEventError( + `agentId ${operation.agentId} does not match breaker agent ${this.agentId}`, + ); + } + const snapshot = this.getState(); + if (snapshot.state === "open") { + throw new CircuitBreakerOpenError(this.agentId, this.consecutiveViolations); + } + if (snapshot.state === "half_open") { + if (this.halfOpenTrialsRemaining <= 0) { + throw new CircuitBreakerOpenError(this.agentId, this.consecutiveViolations); + } + this.halfOpenTrialsRemaining -= 1; + } + } + + /** + * Record a successful operation. In `half_open`, a success closes the + * circuit and resets the violation counters; in `closed` it resets the + * consecutive-violation streak so only genuinely consecutive violations + * trip the breaker. + */ + recordSuccess(): CircuitStateSnapshot { + if (this.state === "half_open") { + this.close(); + } else { + this.consecutiveViolations = 0; + } + return this.getState(); + } + + /** Manually reset the breaker to a pristine closed state. */ + reset(): CircuitStateSnapshot { + this.state = "closed"; + this.violationsRecorded = 0; + this.consecutiveViolations = 0; + this.lastViolationAt = undefined; + this.openedAt = undefined; + this.halfOpenTrialsRemaining = 0; + return this.getState(); + } + + /** Subscribe to circuit-open transitions. Returns an unsubscribe function. */ + onTrip(callback: (snapshot: CircuitStateSnapshot) => void): () => void { + this.tripCallbacks.add(callback); + return () => this.tripCallbacks.delete(callback); + } + + /** Subscribe to circuit-close transitions. Returns an unsubscribe function. */ + onClose(callback: (snapshot: CircuitStateSnapshot) => void): () => void { + this.closeCallbacks.add(callback); + return () => this.closeCallbacks.delete(callback); + } + + private cooldownElapsed(now: string): boolean { + if (this.openedAt === undefined) return false; + const opened = Date.parse(this.openedAt); + const current = Date.parse(now); + if (Number.isNaN(opened) || Number.isNaN(current)) return false; + return current - opened >= this.cooldownMs; + } + + private open(): void { + if (this.state === "open") return; + this.state = "open"; + this.openedAt = new Date().toISOString(); + this.halfOpenTrialsRemaining = 0; + const snapshot = this.getState(this.openedAt); + for (const callback of this.tripCallbacks) { + callback(snapshot); + } + } + + private close(): void { + if (this.state !== "half_open") return; + this.state = "closed"; + this.consecutiveViolations = 0; + this.halfOpenTrialsRemaining = 0; + this.openedAt = undefined; + const snapshot = this.getState(); + for (const callback of this.closeCallbacks) { + callback(snapshot); + } + } +} + +/** A step recorded inside a transaction, reverted on rollback. */ +export interface TransactionStep { + readonly operationId: OperationId; + /** Operation name executed inside the transaction. */ + readonly name: string; + /** ISO-8601 timestamp when the step was recorded. */ + readonly recordedAt: string; +} + +/** An in-flight or finished transaction. */ +export interface Transaction { + readonly transactionId: TransactionId; + readonly agentId: AgentId; + /** ISO-8601 timestamp when the transaction began. */ + readonly startedAt: string; + /** Steps recorded in the transaction so far, in execution order. */ + readonly steps: readonly TransactionStep[]; + readonly committed: boolean; + readonly rolledBack: boolean; +} + +/** Result of a transactional rollback triggered by a policy violation. */ +export interface RollbackRecord { + readonly transactionId: TransactionId; + readonly agentId: AgentId; + /** ISO-8601 timestamp when the rollback completed. */ + readonly rolledBackAt: string; + /** Reason captured from the triggering violation. */ + readonly reason: string; + /** Policy reference that triggered the rollback, if a violation did. */ + readonly policyRef: PolicyRef | undefined; + /** Steps reverted, in execution order. */ + readonly revertedSteps: readonly TransactionStep[]; +} + +export interface RollbackManagerOptions { + /** Maximum simultaneously tracked transactions (default 1000). */ + readonly maxTransactions?: number; +} + +interface MutableTransaction { + transactionId: TransactionId; + agentId: AgentId; + startedAt: string; + steps: TransactionStep[]; + committed: boolean; + rolledBack: boolean; +} + +function snapshotTransaction(txn: MutableTransaction): Transaction { + return { + transactionId: txn.transactionId, + agentId: txn.agentId, + startedAt: txn.startedAt, + steps: [...txn.steps], + committed: txn.committed, + rolledBack: txn.rolledBack, + }; +} + +/** + * The transactional rollback half of the resilience mechanism. It tracks + * per-agent transactions with a step log (`begin()` / `addStep()` / + * `commit()`) and, on a policy violation, `rollback()` reverts the in-flight + * transaction and produces a `RollbackRecord` describing every reverted step + * in execution order. + */ +export class TransactionalRollbackManager { + private readonly maxTransactions: number; + private readonly transactions = new Map(); + private readonly rollbackLog: RollbackRecord[] = []; + private nextTransactionId = 0; + + constructor(options: RollbackManagerOptions = {}) { + this.maxTransactions = options.maxTransactions ?? 1000; + if (this.maxTransactions <= 0) { + throw new InvalidViolationEventError("maxTransactions must be positive"); + } + } + + /** Begin a new transaction for an agent. */ + begin(agentId: AgentId): Transaction { + if (!agentId) throw new InvalidViolationEventError("agentId is required"); + if (this.transactions.size >= this.maxTransactions) { + throw new InvalidViolationEventError( + `transaction table is full (${this.maxTransactions} transactions)`, + ); + } + const transactionId = `txn:${this.nextTransactionId++}`; + const txn: MutableTransaction = { + transactionId, + agentId, + startedAt: new Date().toISOString(), + steps: [], + committed: false, + rolledBack: false, + }; + this.transactions.set(transactionId, txn); + return snapshotTransaction(txn); + } + + /** Record a step inside an in-flight transaction. */ + addStep(transactionId: TransactionId, step: TransactionStep): Transaction { + const txn = this.lookup(transactionId); + if (txn.committed || txn.rolledBack) { + throw new TransactionAlreadyEndedError(transactionId); + } + if (!step || !step.operationId) { + throw new InvalidViolationEventError("step.operationId is required"); + } + if (!step.name) { + throw new InvalidViolationEventError("step.name is required"); + } + txn.steps.push({ + operationId: step.operationId, + name: step.name, + recordedAt: step.recordedAt ?? new Date().toISOString(), + }); + return snapshotTransaction(txn); + } + + /** Commit an in-flight transaction. */ + commit(transactionId: TransactionId): Transaction { + const txn = this.lookup(transactionId); + if (txn.committed || txn.rolledBack) { + throw new TransactionAlreadyEndedError(transactionId); + } + txn.committed = true; + return snapshotTransaction(txn); + } + + /** + * Roll back an in-flight transaction, producing a `RollbackRecord` listing + * every reverted step in execution order. + */ + rollback( + transactionId: TransactionId, + reason: string, + policyRef?: PolicyRef, + ): RollbackRecord { + const txn = this.lookup(transactionId); + if (txn.committed || txn.rolledBack) { + throw new TransactionAlreadyEndedError(transactionId); + } + txn.rolledBack = true; + const record: RollbackRecord = { + transactionId, + agentId: txn.agentId, + rolledBackAt: new Date().toISOString(), + reason, + policyRef, + revertedSteps: [...txn.steps], + }; + this.rollbackLog.push(record); + return record; + } + + /** + * Roll back an in-flight transaction if it is still active. Returns the + * `RollbackRecord`, or `undefined` when the transaction is unknown or + * already ended — automated violation handlers can attempt rollback without + * failing on benign race conditions. + */ + tryRollback( + transactionId: TransactionId, + reason: string, + policyRef?: PolicyRef, + ): RollbackRecord | undefined { + const txn = this.transactions.get(transactionId); + if (!txn || txn.committed || txn.rolledBack) return undefined; + return this.rollback(transactionId, reason, policyRef); + } + + /** Every rollback produced so far, in production order. */ + getRollbackLog(): readonly RollbackRecord[] { + return [...this.rollbackLog]; + } + + /** Transactions that are still in-flight (not committed or rolled back). */ + activeTransactions(): readonly Transaction[] { + const active: Transaction[] = []; + for (const txn of this.transactions.values()) { + if (!txn.committed && !txn.rolledBack) { + active.push(snapshotTransaction(txn)); + } + } + return active; + } + + private lookup(transactionId: TransactionId): MutableTransaction { + const txn = this.transactions.get(transactionId); + if (!txn) throw new TransactionNotFoundError(transactionId); + return txn; + } +} + +export interface ResilienceCoordinatorOptions { + /** Consecutive violations that trip the per-agent breaker (default 3). */ + readonly failureThreshold?: number; + /** Milliseconds a breaker stays open before half-open (default 30000). */ + readonly cooldownMs?: number; + /** Trial operations admitted in half-open (default 1). */ + readonly maxTrials?: number; +} + +/** Outcome of feeding a policy violation event to the coordinator. */ +export interface ViolationOutcome { + readonly event: PolicyViolationEvent; + readonly circuit: CircuitStateSnapshot; + /** Rollback record if the violation referenced an in-flight transaction. */ + readonly rollback: RollbackRecord | undefined; +} + +/** + * Wires the circuit breaker and the transactional rollback manager behind one + * automated entry point: `onPolicyViolation(event)` trips the per-agent + * circuit breaker and rolls back the violating transaction in a single call. + */ +export class ResilienceCoordinator { + private readonly options: ResilienceCoordinatorOptions; + private readonly breakers = new Map(); + private readonly transactions = new TransactionalRollbackManager(); + + constructor(options: ResilienceCoordinatorOptions = {}) { + this.options = options; + } + + /** + * The single automated response path for a policy violation event: feeds the + * per-agent circuit breaker (trip accounting) and rolls back the in-flight + * transaction the violating operation belonged to, if any. + */ + onPolicyViolation(event: PolicyViolationEvent): ViolationOutcome { + validateViolationEvent(event); + const circuit = this.breakerFor(event.agentId).recordViolation(event); + let rollback: RollbackRecord | undefined; + if (event.transactionId !== undefined) { + rollback = this.transactions.tryRollback( + event.transactionId, + `policy violation ${event.violationId}: ${event.detail}`, + event.policyRef, + ); + } + return { event, circuit, rollback }; + } + + /** Gate an operation against the agent's circuit breaker. */ + allowOperation(operation: OperationDescriptor): void { + this.breakerFor(operation.agentId).allowOperation(operation); + } + + /** Record a successful operation for the agent (closes half-open trials). */ + recordSuccess(agentId: AgentId): CircuitStateSnapshot { + return this.breakerFor(agentId).recordSuccess(); + } + + /** Begin a new transaction managed by the coordinator. */ + beginTransaction(agentId: AgentId): Transaction { + return this.transactions.begin(agentId); + } + + /** Record a step inside a coordinator-managed transaction. */ + addStep(transactionId: TransactionId, step: TransactionStep): Transaction { + return this.transactions.addStep(transactionId, step); + } + + /** Commit a coordinator-managed transaction. */ + commitTransaction(transactionId: TransactionId): Transaction { + return this.transactions.commit(transactionId); + } + + /** Snapshot the agent's circuit breaker state. */ + getCircuitState(agentId: AgentId): CircuitStateSnapshot { + return this.breakerFor(agentId).getState(); + } + + /** Every rollback the coordinator has produced, in production order. */ + getRollbackLog(): readonly RollbackRecord[] { + return this.transactions.getRollbackLog(); + } + + /** Coordinator-managed transactions still in-flight. */ + activeTransactions(): readonly Transaction[] { + return this.transactions.activeTransactions(); + } + + private breakerFor(agentId: AgentId): PolicyCircuitBreaker { + let breaker = this.breakers.get(agentId); + if (!breaker) { + breaker = new PolicyCircuitBreaker(agentId, this.options); + this.breakers.set(agentId, breaker); + } + return breaker; + } +}