Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions contextual_orchestrator/cost_ledger.py
Original file line number Diff line number Diff line change
Expand Up @@ -330,6 +330,7 @@ class NoopUsageTelemetrySink:
"""Default sink for callers that do not wire telemetry yet."""

def emit_usage(self, event: UsageTelemetryEvent) -> None:
"""Discard prompt-safe usage telemetry when export is not configured."""
return None


Expand All @@ -342,12 +343,14 @@ def __init__(self, max_events: int = 512) -> None:
self._lock = threading.Lock()

def emit_usage(self, event: UsageTelemetryEvent) -> None:
"""Retain the newest prompt-safe usage events up to the configured limit."""
with self._lock:
self._events.append(event)
if len(self._events) > self._max_events:
del self._events[: len(self._events) - self._max_events]

def events(self) -> List[UsageTelemetryEvent]:
"""Return a snapshot of the retained usage telemetry events."""
with self._lock:
return list(self._events)

Expand All @@ -363,6 +366,7 @@ class UsageTelemetryHealth:
last_error_type: Optional[str] = None

def as_dict(self) -> Dict[str, Any]:
"""Return operator-safe health counters as a serializable mapping."""
return {
"records_accepted": self.records_accepted,
"records_stored": self.records_stored,
Expand Down Expand Up @@ -442,6 +446,7 @@ def flush(self, timeout: Optional[float] = None) -> bool:
return True

def telemetry_health(self) -> Dict[str, Any]:
"""Return current asynchronous ledger persistence and export health."""
with self._lock:
return self._health.as_dict()

Expand Down
7 changes: 7 additions & 0 deletions contextual_orchestrator/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -4441,7 +4441,10 @@ def build_server(
clearfolio_url = clearfolio_url.rstrip("/")

class Handler(BaseHTTPRequestHandler):
"""Handle authenticated orchestration, administration, and health routes."""

def do_GET(self) -> None: # noqa: N802
"""Dispatch GET requests after applying the route's authorization scope."""
parsed = urllib.parse.urlparse(self.path)
path = parsed.path
query = urllib.parse.parse_qs(parsed.query)
Expand Down Expand Up @@ -4791,6 +4794,7 @@ def do_GET(self) -> None: # noqa: N802
self._send_error(500, "internal_error", "internal server error")

def do_PATCH(self) -> None: # noqa: N802
"""Apply an authenticated agent-pool worker update."""
try:
self._authorize("admin")
path = urllib.parse.urlparse(self.path).path
Expand All @@ -4814,6 +4818,7 @@ def do_PATCH(self) -> None: # noqa: N802
self._send_error(500, "internal_error", "internal server error")

def do_DELETE(self) -> None: # noqa: N802
"""Delete an authenticated agent-pool worker resource."""
try:
self._authorize("admin")
path = urllib.parse.urlparse(self.path).path
Expand All @@ -4834,6 +4839,7 @@ def do_DELETE(self) -> None: # noqa: N802
self._send_error(500, "internal_error", "internal server error")

def do_POST(self) -> None: # noqa: N802
"""Dispatch authenticated completion, agent, and simulation writes."""
try:
path = urllib.parse.urlparse(self.path).path
scope = "admin" if path == "/admin/simulate" or path.startswith("/api/v1/agent_pools/") else "inference"
Expand Down Expand Up @@ -5684,6 +5690,7 @@ def _read_json(self) -> dict[str, Any]:
return _coerce_json(raw) if raw else {}

def log_message(self, format: str, *args: object) -> None:
"""Suppress default request logging to keep service output structured."""
return

def _send_error(
Expand Down
Loading