feat(omnibase_infra): add circuit breaker event emission [OMN-5293]

scripts/check_contract_topic_parity.py

@@ -303,6 +303,7 @@
     "onex.evt.omnimemory.memory-stored.v1": "pre-migration; needs contract.yaml in omnimemory repo | owner: jonah | expiry: 2026-06-01",
     "onex.evt.omnimemory.memory-updated.v1": "pre-migration; needs contract.yaml in omnimemory repo | owner: jonah | expiry: 2026-06-01",
     # --- platform/cross-cutting topics ---
+    "onex.evt.omnibase-infra.circuit-breaker.v1": "new topic (OMN-5293); publisher-only, no node contract.yaml yet | owner: jonah | expiry: 2026-09-01",
     "onex.evt.omnibase-infra.gmail-archive-purged.v1": "pre-migration; needs contract.yaml in omnibase_infra | owner: jonah | expiry: 2026-06-01",
     "onex.evt.pattern.discovered.v1": "pre-migration; needs contract.yaml in omniintelligence | owner: jonah | expiry: 2026-06-01",
     "onex.evt.platform.resolution-decided.v1": "pre-migration; needs contract.yaml in omnibase_infra | owner: jonah | expiry: 2026-06-01",

src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py

@@ -19,6 +19,7 @@ class EnumOmnibaseInfraTopic(str, Enum):
     Members are sorted by (kind, event_name, version).
     """
     EVT_BASELINES_COMPUTED_V1 = "onex.evt.omnibase-infra.baselines-computed.v1"  # onex.evt.omnibase-infra.baselines-computed.v1
+    EVT_CIRCUIT_BREAKER_STATE_V1 = "onex.evt.omnibase-infra.circuit-breaker-state.v1"  # onex.evt.omnibase-infra.circuit-breaker-state.v1
     EVT_DB_ERROR_V1 = "onex.evt.omnibase-infra.db-error.v1"  # onex.evt.omnibase-infra.db-error.v1
     EVT_EFFECTIVENESS_DATA_CHANGED_V1 = "onex.evt.omnibase-infra.effectiveness-data-changed.v1"  # onex.evt.omnibase-infra.effectiveness-data-changed.v1
     EVT_GMAIL_INTENT_RECEIVED_V1 = "onex.evt.omnibase-infra.gmail-intent-received.v1"  # onex.evt.omnibase-infra.gmail-intent-received.v1

src/omnibase_infra/event_bus/topic_constants.py

@@ -539,6 +539,19 @@ def derive_dlq_topic_for_event_type(
 Ticket: OMN-2255
 """
 
+# Circuit Breaker State topics (OMN-5293)
+TOPIC_CIRCUIT_BREAKER_STATE: Final[str] = (
+    "onex.evt.omnibase-infra.circuit-breaker-state.v1"
+)
+"""Circuit breaker state transition events emitted by MixinAsyncCircuitBreaker.
+
+Emitted on CLOSED→OPEN, OPEN→HALF_OPEN, and HALF_OPEN→CLOSED transitions.
+
+Producer: MixinAsyncCircuitBreaker (any class using the mixin with event_bus set)
+Consumer: Omnidash /circuit-breakers dashboard
+Ticket: OMN-5293
+"""
+
 # Grouped constants for wiring health monitoring
 WIRING_HEALTH_MONITORED_TOPICS: Final[tuple[str, ...]] = (
     TOPIC_SESSION_OUTCOME_CURRENT,

src/omnibase_infra/models/resilience/__init__.py

@@ -6,7 +6,11 @@
 from omnibase_infra.models.resilience.model_circuit_breaker_config import (
     ModelCircuitBreakerConfig,
 )
+from omnibase_infra.models.resilience.model_circuit_breaker_state_event import (
+    ModelCircuitBreakerStateEvent,
+)
 
 __all__ = [
     "ModelCircuitBreakerConfig",
+    "ModelCircuitBreakerStateEvent",
 ]

src/omnibase_infra/models/resilience/model_circuit_breaker_state_event.py

@@ -0,0 +1,54 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Event model for circuit breaker state transitions.
+
+Emitted whenever a circuit breaker transitions between states:
+  CLOSED → OPEN, OPEN → HALF_OPEN, or HALF_OPEN → CLOSED.
+
+Producer: CircuitBreakerEventPublisher
+Consumer: omnidash /circuit-breaker dashboard (OMN-5293)
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from omnibase_infra.enums.enum_circuit_state import EnumCircuitState
+
+
+class ModelCircuitBreakerStateEvent(BaseModel):
+    """Payload for a circuit breaker state transition event.
+
+    Attributes:
+        service_name: Identifier of the service whose circuit changed.
+        state: New circuit breaker state after the transition.
+        previous_state: State before the transition.
+        failure_count: Failure counter at the time of the transition.
+        threshold: Configured failure threshold for this circuit.
+        timestamp: UTC time the transition occurred.
+        correlation_id: Optional trace correlation ID.
+    """
+
+    service_name: str = Field(description="Service identifier for the circuit breaker")
+    state: EnumCircuitState = Field(description="New circuit breaker state")
+    previous_state: EnumCircuitState = Field(
+        description="Previous circuit breaker state before transition"
+    )
+    failure_count: int = Field(ge=0, description="Failure count at time of transition")
+    threshold: int = Field(ge=1, description="Configured failure threshold")
+    timestamp: datetime = Field(description="UTC timestamp of the state transition")
+    correlation_id: UUID | None = Field(
+        default=None, description="Optional correlation ID for distributed tracing"
+    )
+
+    model_config = ConfigDict(
+        frozen=True,
+        extra="forbid",
+        from_attributes=True,
+    )
+
+
+__all__: list[str] = ["ModelCircuitBreakerStateEvent"]

src/omnibase_infra/services/__init__.py

@@ -47,6 +47,9 @@
 )
 from omnibase_infra.services.corpus_capture import CorpusCapture
 from omnibase_infra.services.service_capability_query import ServiceCapabilityQuery
+from omnibase_infra.services.service_circuit_breaker_event_publisher import (
+    CircuitBreakerEventPublisher,
+)
 from omnibase_infra.services.service_llm_endpoint_health import (
     TOPIC_LLM_ENDPOINT_HEALTH,
     ServiceLlmEndpointHealth,
@@ -117,6 +120,8 @@
     "TOPIC_LLM_ENDPOINT_HEALTH",
     # Topic catalog service (OMN-2311)
     "ServiceTopicCatalog",
+    # Circuit breaker event publisher (OMN-5293)
+    "CircuitBreakerEventPublisher",
     # Contract publisher service (OMN-1752)
     "ContractPublisherError",
     "ContractPublishingInfraError",

src/omnibase_infra/services/service_circuit_breaker_event_publisher.py

@@ -0,0 +1,133 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Circuit breaker state-transition event publisher.
+
+Publishes ModelCircuitBreakerStateEvent to the Kafka topic
+``onex.evt.omnibase-infra.circuit-breaker.v1`` whenever a circuit breaker
+transitions between states (CLOSED ↔ OPEN ↔ HALF_OPEN).
+
+Design:
+    - Stateless: caller passes all transition context on each call.
+    - Fault-tolerant: publish errors are logged but not re-raised so that
+      a bus outage never blocks the circuit breaker's own state machine.
+    - Dependency-injected: accepts any ProtocolEventBusLike implementation
+      for testability (pass an in-memory bus in unit tests).
+
+Usage::
+
+    publisher = CircuitBreakerEventPublisher(event_bus=kafka_bus)
+
+    # Inside _record_circuit_failure / _reset_circuit_breaker overrides:
+    await publisher.publish_transition(
+        service_name="kafka.production",
+        new_state=EnumCircuitState.OPEN,
+        previous_state=EnumCircuitState.CLOSED,
+        failure_count=5,
+        threshold=5,
+    )
+
+Related:
+    - SUFFIX_CIRCUIT_BREAKER_STATE: topic constant (OMN-5293)
+    - ModelCircuitBreakerStateEvent: event payload model
+    - omnidash /circuit-breaker: consumer dashboard
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import UTC, datetime
+from uuid import UUID, uuid4
+
+from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope
+from omnibase_infra.enums.enum_circuit_state import EnumCircuitState
+from omnibase_infra.models.resilience.model_circuit_breaker_state_event import (
+    ModelCircuitBreakerStateEvent,
+)
+from omnibase_infra.protocols.protocol_event_bus_like import ProtocolEventBusLike
+from omnibase_infra.topics import SUFFIX_CIRCUIT_BREAKER_STATE
+
+logger = logging.getLogger(__name__)
+
+
+class CircuitBreakerEventPublisher:
+    """Publishes circuit breaker state transition events to the event bus.
+
+    All publish errors are caught and logged; they never propagate to the
+    circuit breaker itself so that a bus outage cannot block state transitions.
+
+    Attributes:
+        _event_bus: The underlying event bus implementation.
+        _topic: Resolved Kafka topic string for circuit breaker events.
+    """
+
+    def __init__(self, event_bus: ProtocolEventBusLike) -> None:
+        self._event_bus = event_bus
+        self._topic: str = SUFFIX_CIRCUIT_BREAKER_STATE
+
+    async def publish_transition(
+        self,
+        service_name: str,
+        new_state: EnumCircuitState,
+        previous_state: EnumCircuitState,
+        failure_count: int,
+        threshold: int,
+        correlation_id: UUID | None = None,
+    ) -> None:
+        """Publish a state-transition event for a circuit breaker.
+
+        Args:
+            service_name: Service identifier (e.g. "kafka.production").
+            new_state: The circuit breaker's new state after the transition.
+            previous_state: The state before the transition.
+            failure_count: Current failure count at time of transition.
+            threshold: Configured failure threshold for this circuit.
+            correlation_id: Optional trace correlation ID.
+        """
+        now = datetime.now(UTC)
+        event = ModelCircuitBreakerStateEvent(
+            service_name=service_name,
+            state=new_state,
+            previous_state=previous_state,
+            failure_count=failure_count,
+            threshold=threshold,
+            timestamp=now,
+            correlation_id=correlation_id,
+        )
+
+        envelope: ModelEventEnvelope[object] = ModelEventEnvelope(
+            envelope_id=uuid4(),
+            payload=json.loads(event.model_dump_json()),
+            envelope_timestamp=now,
+            correlation_id=correlation_id if correlation_id else uuid4(),
+            source="circuit_breaker_event_publisher",
+        )
+
+        try:
+            await self._event_bus.publish_envelope(
+                envelope,  # type: ignore[arg-type]
+                topic=self._topic,
+            )
+            logger.debug(
+                "Circuit breaker state published",
+                extra={
+                    "service_name": service_name,
+                    "new_state": new_state.value,
+                    "previous_state": previous_state.value,
+                    "failure_count": failure_count,
+                    "topic": self._topic,
+                },
+            )
+        except Exception as exc:  # noqa: BLE001 — never block circuit breaker on bus errors
+            logger.error(  # noqa: TRY400 — avoid leaking stack traces in error logs
+                "Failed to publish circuit breaker state event: %s",
+                exc,
+                extra={
+                    "service_name": service_name,
+                    "new_state": new_state.value,
+                    "topic": self._topic,
+                },
+            )
+
+
+__all__: list[str] = ["CircuitBreakerEventPublisher"]

src/omnibase_infra/topics/__init__.py

@@ -32,6 +32,7 @@
     ALL_PROVISIONED_TOPIC_SPECS,
     ALL_VALIDATION_TOPIC_SPECS,
     SUFFIX_BASELINES_COMPUTED,
+    SUFFIX_CIRCUIT_BREAKER_STATE,
     SUFFIX_CONTRACT_DEREGISTERED,
     SUFFIX_CONTRACT_REGISTERED,
     SUFFIX_FSM_STATE_TRANSITIONS,
@@ -141,6 +142,7 @@
     "SUFFIX_OMNIINTELLIGENCE_ROUTING_DECISION_CMD",
     # OmniBase Infra domain suffix constants
     "SUFFIX_BASELINES_COMPUTED",
+    "SUFFIX_CIRCUIT_BREAKER_STATE",
     "SUFFIX_GMAIL_ARCHIVE_PURGED",
     # OmniMemory domain suffix constants
     "SUFFIX_OMNIMEMORY_DOCUMENT_DISCOVERED",

src/omnibase_infra/topics/platform_topic_suffixes.py

@@ -411,6 +411,18 @@
 Consumer: omnidash /baselines dashboard
 """
 
+SUFFIX_CIRCUIT_BREAKER_STATE: str = "onex.evt.omnibase-infra.circuit-breaker.v1"
+"""Topic suffix for circuit breaker state transition events.
+
+Published by CircuitBreakerEventPublisher whenever a circuit breaker transitions
+between states: CLOSED → OPEN, OPEN → HALF_OPEN, or HALF_OPEN → CLOSED.
+
+Each event carries service_name, state, failure_count, threshold, and timestamp.
+
+Producer: CircuitBreakerEventPublisher (MixinAsyncCircuitBreaker integrations)
+Consumer: omnidash /circuit-breaker dashboard (OMN-5293)
+"""
+
 # Full topic name (not a suffix) — named as such to be unambiguous.
 # Used by monitor_logs.py postgres error emitter and downstream consumers.
 TOPIC_DB_ERROR_V1: str = "onex.evt.omnibase-infra.db-error.v1"
@@ -446,6 +458,15 @@
             "cleanup.policy": "delete",
         },  # 7 days
     ),
+    # Circuit breaker state transitions (3 partitions — low-throughput, state-change-driven)
+    ModelTopicSpec(
+        suffix=SUFFIX_CIRCUIT_BREAKER_STATE,
+        partitions=3,
+        kafka_config={
+            "retention.ms": "604800000",
+            "cleanup.policy": "delete",
+        },  # 7 days
+    ),
 )
 """Omnibase_infra domain topic specs for internal effect nodes.
 

src/omnibase_infra/validation/validation_exemptions.yaml

@@ -945,6 +945,14 @@ pattern_exemptions:
     documentation:
       - docs/patterns/correlation_id_tracking.md
     ticket: OMN-1002
+  - file_pattern: 'model_circuit_breaker_state_event\.py'
+    violation_pattern: "Field 'service_name' might reference an entity"
+    reason: >
+      service_name in ModelCircuitBreakerStateEvent mirrors the same identifier used by MixinAsyncCircuitBreaker.service_name — a free-form string label (e.g. "kafka.production", "db.primary") identifying which service the circuit protects. It is not a reference to a domain entity; it is a telemetry label. The ID + display_name pattern does not apply here.
+
+    documentation:
+      - docs/patterns/circuit_breaker_implementation.md
+    ticket: OMN-5293
   # ModelDispatchLogContext - Dispatch-specific logging
   - file_pattern: 'model_dispatch_log_context\.py'
     class_pattern: "Class 'ModelDispatchLogContext'"