How breaker.onApiError Feeds API Error Storm Trips from OTel Spans into the Circuit Breaker

The breaker.onApiError callback receives agent IDs from OpenTelemetry api_error events and increments internal error counters, triggering circuit breaker trips when configured thresholds are exceeded.

The Munder-Difflin runtime implements a tightly-coupled observability pipeline that transforms OpenTelemetry (OTel) log spans into circuit breaker trips. This integration ensures that every observable API failure automatically contributes to guard-rail logic that protects agents from unsafe execution states. The system connects OTLP-JSON log ingestion to circuit breaker state changes through a three-stage publisher-subscriber pattern.

The Three-Stage Pipeline from OTel Logs to Circuit Breaker

The complete data flow moves from OTLP log ingestion through subscriber notification to circuit breaker state mutation. Each stage is implemented in specific source files within the repository.

Stage 1: Ingesting OTel Logs via TelemetryCollector

Claude Code workers send OTLP-JSON logs to the local TelemetryCollector via the /v1/logs endpoint. When a log record contains event.name equal to api_error (or includes the word "error"), the ingestLogs method in src/main/telemetry.ts creates a telemetry event.

In src/main/telemetry.ts (lines 65-68), the collector inspects incoming log attributes:

// Simplified ingestion logic from src/main/telemetry.ts
ingestLogs(resourceLogs: any[]) {
  for (const log of resourceLogs) {
    if (log.attributes?.find((a: any) => a.key === 'event.name')?.value?.stringValue === 'api_error') {
      this.publishApiError(agentId);
    }
  }
}

The collector extracts the agent.id from resource attributes to identify the offending agent for downstream circuit breaker targeting.

Stage 2: Publishing to apiError Subscribers

The TelemetryCollector maintains a private apiErrorSubs set containing callbacks registered via the public onApiError method. For every detected api_error, the collector iterates over this set and invokes each subscriber with the offending agentId.

In src/main/telemetry.ts (lines 45-48), the subscription mechanism is defined:

class TelemetryCollector {
  private apiErrorSubs = new Set<(agentId: string) => void>();

  onApiError(callback: (agentId: string) => void): () => void {
    this.apiErrorSubs.add(callback);
    return () => this.apiErrorSubs.delete(callback);
  }
}

This pub-sub decoupling allows the circuit breaker to subscribe to error events without direct dependency on the OTLP parsing logic.

Stage 3: Circuit Breaker Consumes the Error and Trips

The main process wires the circuit breaker to the telemetry collector in src/main/index.ts (lines 272-275). The subscription callback invokes breaker.recordError(agentId), which increments internal counters and evaluates storm thresholds.

// From src/main/index.ts - the critical wire-up
telemetry.onApiError((agentId) => breaker.recordError(agentId));

Once the error rate crosses configured thresholds, the breaker trips—setting the agent state to steering, constrained, or stopped—and emits a control:breakerState event for UI consumption.

Wiring the Components Together

The integration between OpenTelemetry ingestion and circuit breaker protection requires explicit subscription setup during application initialization. The following pattern matches the implementation in src/main/index.ts:

import { TelemetryCollector } from './telemetry';
import { CircuitBreaker } from './breaker';

// Create the collector (renderer receives events via emit)
const telemetry = new TelemetryCollector({
  emit: (ch, payload) => webContents.send(ch, payload),
});

// Initialize the breaker with policy callback
const breaker = new CircuitBreaker(() => {
  // Policy logic evaluating usage and error rates
});

// Critical integration point: api_error storm seam
telemetry.onApiError((agentId) => breaker.recordError(agentId));

This wiring ensures that OpenTelemetry log entries serve as the sole input channel for API error storms, guaranteeing that every observable failure is accounted for in the guard-rail logic.

Example OTel Log Payload Structure

Workers emit OTLP-JSON payloads that trigger the pipeline. The TelemetryCollector receives these on the /v1/logs endpoint and parses the event.name field:

{
  "resourceLogs": [
    {
      "resource": {
        "attributes": [
          { "key": "agent.id", "value": { "stringValue": "creed-mqp3l5wn" } }
        ]
      },
      "scopeLogs": [
        {
          "logRecords": [
            {
              "attributes": [
                { "key": "event.name", "value": { "stringValue": "api_error" } },
                { "key": "error", "value": { "stringValue": "Rate limit exceeded" } }
              ]
            }
          ]
        }
      ]
    }
  ]
}

When processing this payload, ingestLogs detects the api_error event name, extracts the agent ID "creed-mqp3l5wn", and notifies all apiErrorSubs callbacks.

How the Circuit Breaker Processes Error Storms

The CircuitBreaker class maintains per-agent error counts and evaluates threshold violations. When recordError receives an agent ID from the telemetry subscription, it updates internal state and potentially triggers a trip:

class CircuitBreaker {
  private errorCounts = new Map<string, number>();
  private threshold = 5; // Configurable threshold

  recordError(agentId: string) {
    const prev = this.errorCounts.get(agentId) ?? 0;
    this.errorCounts.set(agentId, prev + 1);
    
    if (this.errorCounts.get(agentId)! > this.threshold) {
      this.trip(agentId);
    }
  }

  private trip(agentId: string) {
    // Set level to steering, constrained, or stopped per policy
    this.emitBreakerState(agentId, 'steering', 'API error storm');
  }
}

The breaker then emits control:breakerState events that the renderer consumes via useTelemetry/useHive hooks to update UI meters and enforce execution constraints.

Summary

• OTel Integration: The TelemetryCollector in src/main/telemetry.ts parses OTLP-JSON logs and identifies api_error events by inspecting the event.name attribute.
• Pub-Sub Pattern: The onApiError method registers callbacks in a private apiErrorSubs set, decoupling log ingestion from circuit breaker logic.
• Critical Wire-Up: src/main/index.ts lines 272-275 connect the telemetry system to the breaker via telemetry.onApiError((agentId) => breaker.recordError(agentId)).
• Storm Detection: The circuit breaker increments per-agent counters and trips to steering, constrained, or stopped states when error thresholds are exceeded.
• Event Propagation: Tripped breakers emit control:breakerState events consumed by the renderer to update UI state and enforce guard-rails.

Frequently Asked Questions

What specific OTel log attribute triggers the api_error event?

The TelemetryCollector inspects the event.name attribute within log record attributes. When this value equals "api_error" or contains the word "error", the system creates an api_error telemetry event and notifies subscribers. The agent ID is extracted from the resource attributes under the agent.id key.

How does the circuit breaker determine which agent to penalize?

The agent identifier propagates through the OpenTelemetry resource attributes in the OTLP payload. When ingestLogs processes a log record, it extracts the agent.id value from the resource attributes and passes this string to all apiErrorSubs callbacks. The breaker.recordError(agentId) method uses this ID to increment the specific agent's error counter in the internal errorCounts Map.

Can the error threshold for circuit breaker trips be configured?

Yes, the CircuitBreaker class maintains configurable thresholds evaluated in the recordError method. While the simplified example shows a hardcoded threshold value, the actual implementation in src/main/breaker.ts accepts policy callbacks during initialization that determine threshold violations and resulting state transitions (steering, constrained, or stopped).

Is there any input channel for circuit breaker errors besides OTel logs?

No. According to the Munder-Difflin source code, the OpenTelemetry log entry is the sole source of api_error storm detection. The circuit breaker has no separate input channel, ensuring that every observable API failure captured in OTel spans is accounted for in the guard-rail logic. This design guarantees comprehensive error tracking through the telemetry.onApiError subscription pattern.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}