diff --git a/backend/package/yuxi/channels/infra/__init__.py b/backend/package/yuxi/channels/infra/__init__.py index fe816ef3..bfbd3fdb 100644 --- a/backend/package/yuxi/channels/infra/__init__.py +++ b/backend/package/yuxi/channels/infra/__init__.py @@ -1,6 +1,6 @@ +from yuxi.channels.infra.broadcast import EventBroadcaster from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError, CircuitState from yuxi.channels.infra.config_watcher import ConfigWatcher -from yuxi.channels.infra.broadcast import EventBroadcaster from yuxi.channels.infra.external_process import ExternalProcessManager from yuxi.channels.infra.text_chunker import ChunkMode, TextChunk, TextChunker diff --git a/backend/package/yuxi/channels/infra/circuit_breaker.py b/backend/package/yuxi/channels/infra/circuit_breaker.py index b01780b9..7b153b63 100644 --- a/backend/package/yuxi/channels/infra/circuit_breaker.py +++ b/backend/package/yuxi/channels/infra/circuit_breaker.py @@ -19,16 +19,27 @@ class CircuitBreakerOpenError(Exception): pass +CircuitBreakerEventCallback = ( + Callable[[str, str, dict[str, Any]], Awaitable[None]] | Callable[[str, str, dict[str, Any]], None] | None +) + + class CircuitBreaker: def __init__( self, failure_threshold: int = 5, recovery_timeout: float = 60.0, half_open_max_calls: int = 3, + channel_id: str = "", + operation: str = "default", + on_event: CircuitBreakerEventCallback = None, ): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.half_open_max_calls = half_open_max_calls + self.channel_id = channel_id + self.operation = operation + self._on_event = on_event self.state = CircuitState.CLOSED self._failure_count = 0 @@ -44,6 +55,7 @@ class CircuitBreaker: self.state = CircuitState.HALF_OPEN self._half_open_calls = 0 logger.info("Circuit breaker transitioned to HALF_OPEN") + await self._emit_event("on_half_open", {}) else: raise CircuitBreakerOpenError( f"Circuit breaker is OPEN, retry after {self.recovery_timeout - elapsed:.0f}s" @@ -69,6 +81,7 @@ class CircuitBreaker: self._failure_count = 0 self._half_open_calls = 0 logger.info("Circuit breaker closed after successful HALF_OPEN probe") + await self._emit_event("on_close", {}) elif self.state == CircuitState.CLOSED and self._failure_count > 0: self._failure_count = 0 @@ -82,6 +95,24 @@ class CircuitBreaker: logger.warning( f"Circuit breaker OPEN after {self._failure_count} failures (threshold={self.failure_threshold})" ) + await self._emit_event("on_open", {"failure_count": self._failure_count}) elif self.state == CircuitState.HALF_OPEN: self.state = CircuitState.OPEN logger.warning("Circuit breaker re-OPENed after HALF_OPEN failure") + await self._emit_event("on_open", {"failure_count": self._failure_count, "from": "half_open"}) + + async def _emit_event(self, event: str, detail: dict[str, Any]) -> None: + if self._on_event is None: + return + payload = { + "channel_id": self.channel_id, + "operation": self.operation, + "state": self.state, + **detail, + } + try: + result = self._on_event(event, self.channel_id, payload) # type: ignore[call-arg] + if asyncio.iscoroutine(result): + await result + except Exception: + logger.debug(f"CircuitBreaker event callback error: {event}", exc_info=True) diff --git a/backend/package/yuxi/channels/infra/config_watcher.py b/backend/package/yuxi/channels/infra/config_watcher.py index 58859602..8af12909 100644 --- a/backend/package/yuxi/channels/infra/config_watcher.py +++ b/backend/package/yuxi/channels/infra/config_watcher.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio import hashlib import json -from typing import Any, TYPE_CHECKING +from typing import TYPE_CHECKING, Any from yuxi.utils.logging_config import logger diff --git a/backend/package/yuxi/channels/infra/text_chunker.py b/backend/package/yuxi/channels/infra/text_chunker.py index 017b7ed3..25e7f508 100644 --- a/backend/package/yuxi/channels/infra/text_chunker.py +++ b/backend/package/yuxi/channels/infra/text_chunker.py @@ -2,8 +2,8 @@ from __future__ import annotations import re from collections.abc import Iterator -from enum import StrEnum from dataclasses import dataclass +from enum import StrEnum class ChunkMode(StrEnum):