diff --git a/scripts/check_contract_topic_parity.py b/scripts/check_contract_topic_parity.py index 74409f6cc8..3c6c3510ab 100644 --- a/scripts/check_contract_topic_parity.py +++ b/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", diff --git a/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py b/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py index 6bedd93f7b..25a375bf54 100644 --- a/src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py +++ b/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 diff --git a/src/omnibase_infra/event_bus/topic_constants.py b/src/omnibase_infra/event_bus/topic_constants.py index 517c5a82f5..349a1087a3 100644 --- a/src/omnibase_infra/event_bus/topic_constants.py +++ b/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, diff --git a/src/omnibase_infra/models/resilience/__init__.py b/src/omnibase_infra/models/resilience/__init__.py index 840eed09af..fa815cae4c 100644 --- a/src/omnibase_infra/models/resilience/__init__.py +++ b/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", ] diff --git a/src/omnibase_infra/models/resilience/model_circuit_breaker_state_event.py b/src/omnibase_infra/models/resilience/model_circuit_breaker_state_event.py new file mode 100644 index 0000000000..08d16bd303 --- /dev/null +++ b/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"] diff --git a/src/omnibase_infra/services/__init__.py b/src/omnibase_infra/services/__init__.py index c3732373e8..7672da48c0 100644 --- a/src/omnibase_infra/services/__init__.py +++ b/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", diff --git a/src/omnibase_infra/services/service_circuit_breaker_event_publisher.py b/src/omnibase_infra/services/service_circuit_breaker_event_publisher.py new file mode 100644 index 0000000000..f893a90868 --- /dev/null +++ b/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"] diff --git a/src/omnibase_infra/topics/__init__.py b/src/omnibase_infra/topics/__init__.py index 37b0a6006a..4042b01055 100644 --- a/src/omnibase_infra/topics/__init__.py +++ b/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", diff --git a/src/omnibase_infra/topics/platform_topic_suffixes.py b/src/omnibase_infra/topics/platform_topic_suffixes.py index 95d0de9db4..438c32ef88 100644 --- a/src/omnibase_infra/topics/platform_topic_suffixes.py +++ b/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. diff --git a/src/omnibase_infra/validation/validation_exemptions.yaml b/src/omnibase_infra/validation/validation_exemptions.yaml index 32bd8be79a..abb3d01f75 100644 --- a/src/omnibase_infra/validation/validation_exemptions.yaml +++ b/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'" diff --git a/tests/unit/services/test_circuit_breaker_event_publisher.py b/tests/unit/services/test_circuit_breaker_event_publisher.py new file mode 100644 index 0000000000..1a4399246d --- /dev/null +++ b/tests/unit/services/test_circuit_breaker_event_publisher.py @@ -0,0 +1,165 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Unit tests for CircuitBreakerEventPublisher. + +Tests verify: + - publish_transition() builds a well-formed ModelCircuitBreakerStateEvent + - the event is forwarded to the event bus with the correct topic + - bus errors are swallowed (never propagated to callers) + - correlation_id is threaded through when provided +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import UUID, uuid4 + +import pytest + +from omnibase_infra.enums.enum_circuit_state import EnumCircuitState +from omnibase_infra.models.resilience.model_circuit_breaker_state_event import ( + ModelCircuitBreakerStateEvent, +) +from omnibase_infra.services.service_circuit_breaker_event_publisher import ( + CircuitBreakerEventPublisher, +) +from omnibase_infra.topics import SUFFIX_CIRCUIT_BREAKER_STATE + + +@pytest.mark.unit +class TestCircuitBreakerEventPublisher: + """Tests for CircuitBreakerEventPublisher.""" + + def _make_publisher(self) -> tuple[CircuitBreakerEventPublisher, AsyncMock]: + mock_bus = MagicMock() + mock_bus.publish_envelope = AsyncMock() + publisher = CircuitBreakerEventPublisher(event_bus=mock_bus) + return publisher, mock_bus.publish_envelope + + @pytest.mark.asyncio + async def test_publish_transition_calls_event_bus(self) -> None: + """publish_transition() should call event_bus.publish_envelope exactly once.""" + publisher, publish_envelope = self._make_publisher() + + await publisher.publish_transition( + service_name="kafka.test", + new_state=EnumCircuitState.OPEN, + previous_state=EnumCircuitState.CLOSED, + failure_count=5, + threshold=5, + ) + + publish_envelope.assert_awaited_once() + + @pytest.mark.asyncio + async def test_publish_uses_correct_topic(self) -> None: + """The event bus must be called with the circuit breaker topic suffix.""" + publisher, publish_envelope = self._make_publisher() + + await publisher.publish_transition( + service_name="db.primary", + new_state=EnumCircuitState.OPEN, + previous_state=EnumCircuitState.CLOSED, + failure_count=3, + threshold=3, + ) + + _, kwargs = publish_envelope.call_args + assert kwargs["topic"] == SUFFIX_CIRCUIT_BREAKER_STATE + + @pytest.mark.asyncio + async def test_payload_fields(self) -> None: + """The envelope payload should contain correct field values.""" + publisher, publish_envelope = self._make_publisher() + corr_id = uuid4() + + await publisher.publish_transition( + service_name="http.gateway", + new_state=EnumCircuitState.HALF_OPEN, + previous_state=EnumCircuitState.OPEN, + failure_count=0, + threshold=5, + correlation_id=corr_id, + ) + + envelope = publish_envelope.call_args[0][0] + payload = envelope.payload + assert payload["service_name"] == "http.gateway" + assert payload["state"] == EnumCircuitState.HALF_OPEN.value + assert payload["previous_state"] == EnumCircuitState.OPEN.value + assert payload["failure_count"] == 0 + assert payload["threshold"] == 5 + + @pytest.mark.asyncio + async def test_bus_error_is_swallowed(self) -> None: + """A publish error must not propagate — circuit breaker must remain functional.""" + mock_bus = MagicMock() + mock_bus.publish_envelope = AsyncMock(side_effect=RuntimeError("bus down")) + publisher = CircuitBreakerEventPublisher(event_bus=mock_bus) + + # Should not raise + await publisher.publish_transition( + service_name="kafka.prod", + new_state=EnumCircuitState.OPEN, + previous_state=EnumCircuitState.CLOSED, + failure_count=5, + threshold=5, + ) + + @pytest.mark.asyncio + async def test_correlation_id_forwarded_in_envelope(self) -> None: + """When a correlation_id is provided it must appear in the envelope.""" + publisher, publish_envelope = self._make_publisher() + corr_id = uuid4() + + await publisher.publish_transition( + service_name="valkey", + new_state=EnumCircuitState.CLOSED, + previous_state=EnumCircuitState.HALF_OPEN, + failure_count=0, + threshold=5, + correlation_id=corr_id, + ) + + envelope = publish_envelope.call_args[0][0] + assert envelope.correlation_id == corr_id + + @pytest.mark.asyncio + async def test_all_three_state_transitions_publish(self) -> None: + """All valid state transitions should publish without error.""" + transitions = [ + (EnumCircuitState.OPEN, EnumCircuitState.CLOSED), + (EnumCircuitState.HALF_OPEN, EnumCircuitState.OPEN), + (EnumCircuitState.CLOSED, EnumCircuitState.HALF_OPEN), + ] + for new_state, prev_state in transitions: + publisher, publish_envelope = self._make_publisher() + await publisher.publish_transition( + service_name="svc", + new_state=new_state, + previous_state=prev_state, + failure_count=2, + threshold=5, + ) + publish_envelope.assert_awaited_once() + + def test_model_circuit_breaker_state_event_schema(self) -> None: + """ModelCircuitBreakerStateEvent must be constructable with valid data.""" + event = ModelCircuitBreakerStateEvent( + service_name="kafka.dev", + state=EnumCircuitState.OPEN, + previous_state=EnumCircuitState.CLOSED, + failure_count=5, + threshold=5, + timestamp=datetime.now(UTC), + ) + assert event.state == EnumCircuitState.OPEN + assert event.service_name == "kafka.dev" + assert event.correlation_id is None + + def test_topic_constant_format(self) -> None: + """SUFFIX_CIRCUIT_BREAKER_STATE must follow ONEX topic naming conventions.""" + assert SUFFIX_CIRCUIT_BREAKER_STATE.startswith("onex.evt.") + assert "circuit-breaker" in SUFFIX_CIRCUIT_BREAKER_STATE + assert SUFFIX_CIRCUIT_BREAKER_STATE.endswith(".v1")