from __future__ import annotations
import json
import os
import threading
import time
from collections import deque
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
EVENT_LEVELS = {"trace": 10, "debug": 20, "info": 30, "warn": 40, "error": 50, "fatal": 60}
DEFAULT_EVENT_LEVEL = "info"
DEFAULT_EVENT_BUFFER = 1000
def _env_bool(name: str, default: bool = True) -> bool:
value = os.environ.get(name)
if value is None:
return default
return value.strip().lower() not in {"0", "false", "off", "no", ""}
def _env_int(name: str, default: int) -> int:
try:
return max(1, int(str(os.environ.get(name, default)).strip()))
except Exception:
return default
def _level_value(level: str) -> int:
return EVENT_LEVELS.get(str(level or "").strip().lower(), EVENT_LEVELS[DEFAULT_EVENT_LEVEL])
def _redact_value(value: Any) -> Any:
if isinstance(value, dict):
out: dict[str, Any] = {}
for key, item in value.items():
lowered = str(key).lower()
secret_key = lowered in {"authorization", "password", "secret", "token", "api_key", "apikey", "access_token", "refresh_token"} or lowered.endswith(("_key", "_token", "_secret", "_password"))
if secret_key:
out[key] = "[redacted]"
else:
out[key] = _redact_value(item)
return out
if isinstance(value, list):
return [_redact_value(item) for item in value]
return value
@dataclass(frozen=True)
class EventConfig:
enabled: bool
level: str
buffer_size: int
@classmethod
def from_env(cls) -> "EventConfig":
return cls(
enabled=_env_bool("CLAUDE_ANY_EVENT_LOG", True),
level=os.environ.get("CLAUDE_ANY_EVENT_LEVEL", DEFAULT_EVENT_LEVEL).strip().lower() or DEFAULT_EVENT_LEVEL,
buffer_size=_env_int("CLAUDE_ANY_EVENT_BUFFER", DEFAULT_EVENT_BUFFER),
)
class EventBus:
def __init__(self, config: EventConfig | None = None) -> None:
self.config = config or EventConfig.from_env()
self._events: deque[dict[str, Any]] = deque(maxlen=self.config.buffer_size)
self._condition = threading.Condition()
self._next_id = 1
def publish(
self,
*,
level: str,
category: str,
message: str,
source: str = "router",
session_id: str | None = None,
request_id: str | None = None,
provider: str | None = None,
model: str | None = None,
data: dict[str, Any] | None = None,
) -> dict[str, Any] | None:
if not self.config.enabled:
return None
level_name = str(level or DEFAULT_EVENT_LEVEL).strip().lower()
if _level_value(level_name) < _level_value(self.config.level):
return None
with self._condition:
event = {
"id": self._next_id,
"time": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"),
"ts": time.time(),
"level": level_name,
"source": source,
"category": str(category or "router"),
"session_id": session_id or "",
"request_id": request_id or "",
"provider": provider or "",
"model": model or "",
"message": str(message or ""),
"data": _redact_value(data or {}),
}
self._next_id += 1
self._events.append(event)
self._condition.notify_all()
return event
def recent(self, *, limit: int = 200, min_id: int | None = None, level: str | None = None, category: str | None = None) -> list[dict[str, Any]]:
with self._condition:
events = list(self._events)
if min_id is not None:
events = [event for event in events if int(event.get("id") or 0) > min_id]
if level:
threshold = _level_value(level)
events = [event for event in events if _level_value(str(event.get("level") or "")) >= threshold]
if category:
events = [event for event in events if str(event.get("category") or "").startswith(category)]
if limit <= 0:
return []
return events[-limit:]
def wait_after(self, last_id: int, timeout: float = 15.0) -> list[dict[str, Any]]:
deadline = time.time() + max(0.1, timeout)
with self._condition:
while not any(int(event.get("id") or 0) > last_id for event in self._events):
remaining = deadline - time.time()
if remaining <= 0:
break
self._condition.wait(min(remaining, 1.0))
return [event for event in self._events if int(event.get("id") or 0) > last_id]
def render_events_html(events_path: str = "/ca/events/stream", recent_path: str = "/ca/events/recent") -> str:
events_path_js = json.dumps(events_path)
recent_path_js = json.dumps(recent_path)
return f"""
Claude Any Router Events
"""