From 5ae6f167d3db3601cc5ca7a1ee23e695a3dde6fc Mon Sep 17 00:00:00 2001 From: Matt Owen Date: Thu, 6 Aug 2026 16:32:37 -0400 Subject: [PATCH 1/4] observability: auto-bootstrap Phoenix wiring from env MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ObservabilityIntegration existed but was inert — nothing ever called `.connect()`, so PHOENIX_ENDPOINT / PHOENIX_PROJECT had no effect and no LLM spans ever reached the collector. Add `ObservabilityIntegration.bootstrap_from_env()` as a singleton factory that idempotently instantiates + connects when both env vars are set (returns None otherwise). Wire it into every process entry point: - `clearwing.main()` covers all CLI subcommands, including `clearwing operate --machine-fd 3` used by embedding hosts. - `create_app()` covers the webui/FastAPI server and registers a shutdown hook so the OTel batch processor flushes on graceful exit. Safe to call from multiple entry points concurrently; the classvar `_singleton` guard makes the call idempotent. `disconnect()` releases the singleton so a fresh bootstrap can rebuild — primarily useful for tests. --- clearwing/__init__.py | 6 +++++ clearwing/observability/integration.py | 35 +++++++++++++++++++++++++- clearwing/ui/web/app.py | 9 +++++++ 3 files changed, 49 insertions(+), 1 deletion(-) diff --git a/clearwing/__init__.py b/clearwing/__init__.py index 7e5132c8..0e0b5cd5 100644 --- a/clearwing/__init__.py +++ b/clearwing/__init__.py @@ -9,7 +9,13 @@ def main(): """Main entry point for Clearwing.""" + from .observability.integration import ObservabilityIntegration from .ui.cli import CLI + # Auto-wire Phoenix / OTLP tracing when PHOENIX_ENDPOINT is set. No-op + # otherwise, so plain `clearwing …` invocations without the env vars pay + # zero cost. + ObservabilityIntegration.bootstrap_from_env() + cli = CLI() cli.run() diff --git a/clearwing/observability/integration.py b/clearwing/observability/integration.py index 4329a11f..19a5bef1 100644 --- a/clearwing/observability/integration.py +++ b/clearwing/observability/integration.py @@ -3,7 +3,8 @@ from __future__ import annotations import logging -from typing import Any +import os +from typing import Any, ClassVar from clearwing.core.events import EventBus, EventType @@ -32,8 +33,16 @@ class ObservabilityIntegration: # ... run agent ... obs.disconnect() print(obs.metrics.format_prometheus()) + + For env-driven auto-wiring at process startup, prefer + :meth:`bootstrap_from_env` — it is a no-op unless Phoenix env vars are set. """ + # Process-wide singleton established by :meth:`bootstrap_from_env` so + # startup hooks in the CLI and webui don't double-subscribe. Tests may + # clear this by calling ``disconnect()``, which unsets the reference. + _singleton: ClassVar["ObservabilityIntegration | None"] = None + def __init__(self, debug: bool = False, exporters: list = None): if exporters is None: exporters = [] @@ -49,6 +58,26 @@ def __init__(self, debug: bool = False, exporters: list = None): self._connected = False self._handlers = {} + @classmethod + def bootstrap_from_env(cls) -> "ObservabilityIntegration | None": + """Idempotently instantiate + connect when Phoenix env vars are set. + + Returns the shared singleton, or ``None`` if + ``PHOENIX_ENDPOINT`` / ``PHOENIX_PROJECT`` are unset. Safe to call + from every process entry point (CLI ``main()``, FastAPI + ``create_app()``, machine-fd subcommand); subsequent calls return the + already-connected instance without re-subscribing to the EventBus. + """ + if cls._singleton is not None: + return cls._singleton + if not os.environ.get("PHOENIX_ENDPOINT") or not os.environ.get("PHOENIX_PROJECT"): + return None + instance = cls() + instance.connect() + cls._singleton = instance + logger.info("ObservabilityIntegration bootstrapped from env") + return instance + def connect(self) -> None: """Subscribe to EventBus events.""" if self._connected: @@ -80,6 +109,10 @@ def disconnect(self) -> None: self.tracer.shutdown() self._connected = False + # Release the bootstrap singleton so a follow-up bootstrap can create + # a fresh instance (primarily useful for tests). + if type(self)._singleton is self: + type(self)._singleton = None @property def spans(self) -> list: diff --git a/clearwing/ui/web/app.py b/clearwing/ui/web/app.py index 990a0951..952f8d23 100644 --- a/clearwing/ui/web/app.py +++ b/clearwing/ui/web/app.py @@ -21,6 +21,7 @@ from clearwing.agent.runtime import Command from clearwing.core.events import EventBus, EventType from clearwing.observability import MetricsCollector +from clearwing.observability.integration import ObservabilityIntegration logger = logging.getLogger(__name__) @@ -54,6 +55,14 @@ def create_app(): allow_headers=["*"], ) + # Auto-wire Phoenix / OTLP tracing when PHOENIX_ENDPOINT is set. Flush the + # OTel batch processor on graceful shutdown so nothing gets dropped. + _obs = ObservabilityIntegration.bootstrap_from_env() + if _obs is not None: + @app.on_event("shutdown") + def _flush_observability() -> None: + _obs.disconnect() + # --------------------------------------------------------------- # Authentication for privileged endpoints # --------------------------------------------------------------- From e1a511fe5374d99205732453320aa73ead617bb5 Mon Sep 17 00:00:00 2001 From: Matt Owen Date: Thu, 6 Aug 2026 16:35:01 -0400 Subject: [PATCH 2/4] observability: use OpenInference semantic conventions on LLM spans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The synthetic ``llm_call`` span was rendered as a generic "internal" span in the Arize Phoenix UI because the attribute names didn't match Phoenix's OpenInference-based classifier. Switch to the documented OpenInference conventions: - openinference.span.kind = "LLM" (was: span.kind = "llm") - llm.model_name (was: llm.model) - llm.token_count.prompt (was: llm.token_count.input) - llm.token_count.completion (was: llm.token_count.output) - llm.token_count.total (new — sum) - llm.token_count.cached (unchanged) - llm.provider, llm.cost_usd (unchanged) Reference: https://github.com/Arize-ai/openinference/blob/main/spec/semantic_conventions.md Also coerce int-typed token counts up front so downstream span serializers (OTel expects concrete ints/floats) don't have to re-parse. --- clearwing/observability/integration.py | 35 +++++++++++++++++--------- 1 file changed, 23 insertions(+), 12 deletions(-) diff --git a/clearwing/observability/integration.py b/clearwing/observability/integration.py index 19a5bef1..ece74bde 100644 --- a/clearwing/observability/integration.py +++ b/clearwing/observability/integration.py @@ -146,26 +146,37 @@ def _on_cost_update(self, data: Any) -> None: if not isinstance(data, dict): return model = data.get("model", "unknown") + input_tokens = int(data.get("input_tokens", 0) or 0) + output_tokens = int(data.get("output_tokens", 0) or 0) + cached_tokens = int(data.get("cached_tokens", 0) or 0) + self.metrics.increment("llm_calls_total", labels={"model": model}) - self.metrics.increment("input_tokens_total", value=float(data.get("input_tokens", 0))) - self.metrics.increment("output_tokens_total", value=float(data.get("output_tokens", 0))) + self.metrics.increment("input_tokens_total", value=float(input_tokens)) + self.metrics.increment("output_tokens_total", value=float(output_tokens)) self.metrics.set_gauge("total_cost_usd", data.get("total_cost_usd", 0.0)) - # Emit a synthetic span so Arize sees per-call LLM telemetry + # Emit a synthetic span so Arize Phoenix sees per-call LLM telemetry. + # Attribute names follow the OpenInference semantic conventions — + # https://github.com/Arize-ai/openinference/blob/main/spec/semantic_conventions.md + # — so Phoenix renders the span in its LLM view rather than as a + # generic "internal" span. import time now = time.time() - elapsed_s = data.get("elapsed_ms", 0) / 1000.0 - with self.tracer.span("llm_call", attributes={ - "llm.model": model, + elapsed_ms = data.get("elapsed_ms", 0) or 0 + elapsed_s = elapsed_ms / 1000.0 + attributes = { + "openinference.span.kind": "LLM", + "llm.model_name": model, "llm.provider": data.get("provider", "unknown"), - "llm.token_count.input": data.get("input_tokens", 0), - "llm.token_count.output": data.get("output_tokens", 0), - "llm.token_count.cached": data.get("cached_tokens", 0), + "llm.token_count.prompt": input_tokens, + "llm.token_count.completion": output_tokens, + "llm.token_count.total": input_tokens + output_tokens, + "llm.token_count.cached": cached_tokens, "llm.cost_usd": data.get("total_cost_usd", 0.0), - "span.kind": "llm", - }) as s: - # Backdate start to reflect actual call timing + } + with self.tracer.span("llm_call", attributes=attributes) as s: + # Backdate start so the span duration reflects actual call latency. if elapsed_s > 0: s.start_time = now - elapsed_s From 5b275e24905929789b3c872625c0361d05a82253 Mon Sep 17 00:00:00 2001 From: Matt Owen Date: Thu, 6 Aug 2026 16:35:06 -0400 Subject: [PATCH 3/4] observability: derive Phoenix SpanKind from openinference.span.kind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ``_to_readable_span`` hard-coded ``SpanKind.INTERNAL`` on every export, so even after the LLM span carried the correct OpenInference ``openinference.span.kind = "LLM"`` attribute, the OTel envelope Phoenix received still classified it as an internal generic span. Introduce ``_span_kind_from_openinference`` that maps the attribute to the most fitting OTel kind: - LLM / TOOL → CLIENT (outbound call semantics) - CHAIN / AGENT / * → INTERNAL (orchestration) The attribute itself is preserved on the exported span — Phoenix keys off both the string and the OTel enum. --- clearwing/observability/phoenix.py | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/clearwing/observability/phoenix.py b/clearwing/observability/phoenix.py index d43ac1d2..330c8a6a 100644 --- a/clearwing/observability/phoenix.py +++ b/clearwing/observability/phoenix.py @@ -15,10 +15,29 @@ logger = logging.getLogger(__name__) +def _span_kind_from_openinference(span: Span): + """Derive OTel SpanKind from an ``openinference.span.kind`` attribute. + + Phoenix's UI keys off both the attribute string and the OTel ``SpanKind`` + enum. We keep the attribute on the outgoing span (so Phoenix classifies it + correctly) and additionally translate to the most fitting OTel kind: + + - ``"LLM"`` / ``"TOOL"`` → ``SpanKind.CLIENT`` (outbound call) + - ``"CHAIN"`` / ``"AGENT"`` → ``SpanKind.INTERNAL`` (internal orchestration) + - unset / anything else → ``SpanKind.INTERNAL`` + """ + from opentelemetry.trace import SpanKind + + kind = span.attributes.get("openinference.span.kind") + if kind in ("LLM", "TOOL"): + return SpanKind.CLIENT + return SpanKind.INTERNAL + + def _to_readable_span(span: Span, resource): """Bridge a Clearwing Span to an OTel ReadableSpan.""" from opentelemetry.sdk.trace import ReadableSpan as _ReadableSpan - from opentelemetry.trace import SpanContext, SpanKind, TraceFlags + from opentelemetry.trace import SpanContext, TraceFlags from opentelemetry.trace.status import Status, StatusCode trace_id = int(span.trace_id, 16) & ((1 << 128) - 1) @@ -37,7 +56,7 @@ def _to_readable_span(span: Span, resource): resource=resource, attributes=span.attributes, events=tuple(), - kind=SpanKind.INTERNAL, + kind=_span_kind_from_openinference(span), status=status, start_time=int(span.start_time * 1e9), end_time=int(span.end_time * 1e9), From 48af566a4a5ede9aadc750ff0681f359bd4976a3 Mon Sep 17 00:00:00 2001 From: Matt Owen Date: Thu, 6 Aug 2026 16:35:21 -0400 Subject: [PATCH 4/4] observability: enrich COST_UPDATE payload + emit from sourcehunt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related fixes so Phoenix LLM spans surface every real LLM call with useful attributes: 1. Enrich the ``COST_UPDATE`` payload emitted from ``CostTracker.record_llm_call`` with ``cached_tokens``, ``elapsed_ms``, and ``provider``. Adds keyword-only ``elapsed_ms`` and ``provider`` args (backward-compatible — legacy callers still work and get safe defaults). 2. Fix a latent bug: ``EventBus.emit`` was called on the class, not an instance, so ``event_type`` was silently binding to ``self`` and the ``COST_UPDATE`` handler never fired. Route through ``EventBus()`` (the singleton) instead. 3. Call ``CostTracker().record_llm_call`` from the sourcehunt hunter's per-turn LLM loop as well. It previously called ``_estimate_cost_usd`` (a local wrapper around ``CostTracker.estimate_cost``) that never touched the singleton, so Phoenix saw operator LLM spans but no hunter ones. The local ``total_cost_usd`` tally is preserved unchanged — the additional call only surfaces telemetry. 4. Measure wall-clock latency around both LLM call sites so the backdated Phoenix span duration reflects real API latency. Tests: cover the new payload fields, the OpenInference attribute names on the emitted span, the ``_span_kind_from_openinference`` mapping, and ``bootstrap_from_env`` singleton semantics. --- clearwing/agent/runtime.py | 12 ++- clearwing/observability/telemetry.py | 14 +++- clearwing/sourcehunt/hunter.py | 21 +++++- tests/test_observability.py | 109 ++++++++++++++++++++++++++- tests/test_telemetry.py | 54 +++++++++++++ 5 files changed, 203 insertions(+), 7 deletions(-) diff --git a/clearwing/agent/runtime.py b/clearwing/agent/runtime.py index 01e600f3..3bd19930 100644 --- a/clearwing/agent/runtime.py +++ b/clearwing/agent/runtime.py @@ -258,12 +258,16 @@ async def _aassistant_step(self, state: dict[str, Any]) -> dict[str, Any]: system, chat_messages = _coerce_chat_messages(messages) system = "\n\n".join(part for part in (sys_prompt, system) if part) or sys_prompt + import time as _time + + _llm_start = _time.perf_counter() response = await self.llm.achat_stream( messages=chat_messages, system=system, tools=self.native_tools or None, on_text_delta=self.on_text_delta, ) + _llm_elapsed_ms = (_time.perf_counter() - _llm_start) * 1000.0 assistant_text = response_text(response) # tool_calls are raw genai ToolCall objects (.call_id/.fn_name/ @@ -289,7 +293,13 @@ async def _aassistant_step(self, state: dict[str, Any]) -> dict[str, Any]: input_tokens = (usage.prompt_tokens or 0) if usage else 0 output_tokens = (usage.completion_tokens or 0) if usage else 0 if self.cost_tracker and (input_tokens or output_tokens): - self.cost_tracker.record_llm_call(input_tokens, output_tokens, self.model_name) + self.cost_tracker.record_llm_call( + input_tokens, + output_tokens, + self.model_name, + elapsed_ms=_llm_elapsed_ms, + provider=getattr(self.llm, "provider_name", None), + ) state["total_cost_usd"] = self.cost_tracker.total_cost_usd state["total_tokens"] = self.cost_tracker.input_tokens + self.cost_tracker.output_tokens if self.audit_logger: diff --git a/clearwing/observability/telemetry.py b/clearwing/observability/telemetry.py index c8ba0fac..3d352089 100644 --- a/clearwing/observability/telemetry.py +++ b/clearwing/observability/telemetry.py @@ -112,6 +112,9 @@ def record_llm_call( output_tokens: int, model: str, cached_tokens: int = 0, + *, + elapsed_ms: float | None = None, + provider: str | None = None, ) -> None: """Record token usage for a single LLM call and update the running cost. @@ -119,6 +122,12 @@ def record_llm_call( pricing is used. ``cached_tokens`` bills at the model's cached rate. When an ``EventBus`` is available a ``COST_UPDATE`` event is emitted after updating counters. + + ``elapsed_ms`` (wall-clock latency of the call) and ``provider`` are + optional; when supplied they ride along in the ``COST_UPDATE`` payload + so ``ObservabilityIntegration`` can attach them to the synthetic + Phoenix LLM span. Keyword-only to keep call sites explicit and future + additions non-breaking. """ cost = self.estimate_cost(input_tokens, output_tokens, model, cached_tokens) @@ -128,14 +137,17 @@ def record_llm_call( self.total_cost_usd += cost try: - EventBus.emit( + EventBus().emit( EventType.COST_UPDATE, { "input_tokens": input_tokens, "output_tokens": output_tokens, + "cached_tokens": cached_tokens, "cost": cost, "total_cost_usd": self.total_cost_usd, "model": model, + "provider": provider or "unknown", + "elapsed_ms": elapsed_ms or 0, }, ) except Exception: diff --git a/clearwing/sourcehunt/hunter.py b/clearwing/sourcehunt/hunter.py index 02ab7e9f..0cc29e89 100644 --- a/clearwing/sourcehunt/hunter.py +++ b/clearwing/sourcehunt/hunter.py @@ -1456,11 +1456,15 @@ async def arun(self) -> HunterRunResult: messages = await self.summarizer.summarize(messages, self.llm) logger.info("Hunter context summarized: %d → %d messages", pre, len(messages)) + import time as _time + + _llm_start = _time.perf_counter() response = await self.llm.achat( messages=messages, system=self.prompt, tools=self.tools, ) + _llm_elapsed_ms = (_time.perf_counter() - _llm_start) * 1000.0 # Preserve the provider's reasoning_content alongside the # visible text. `response.first_text` only returns the # first Text part — reasoning/thinking blocks are separate @@ -1495,12 +1499,25 @@ async def arun(self) -> HunterRunResult: # response where nothing was cache-served. details = getattr(response.usage, "prompt_tokens_details", None) cached_tokens = (getattr(details, "cached_tokens", None) or 0) if details else 0 + _prompt_toks = response.usage.prompt_tokens or 0 + _completion_toks = response.usage.completion_tokens or 0 total_cost_usd += _estimate_cost_usd( - response.usage.prompt_tokens or 0, - response.usage.completion_tokens or 0, + _prompt_toks, + _completion_toks, self.llm.model_name, cached_tokens, ) + # Also push through the singleton so ObservabilityIntegration + # emits a Phoenix LLM span for this hunter call. + if _prompt_toks or _completion_toks: + CostTracker().record_llm_call( + _prompt_toks, + _completion_toks, + self.llm.model_name, + cached_tokens=cached_tokens, + elapsed_ms=_llm_elapsed_ms, + provider=getattr(self.llm, "provider_name", None), + ) last_assistant_text = response.first_text or "" if last_assistant_text: diff --git a/tests/test_observability.py b/tests/test_observability.py index 07b3bec4..2e837ce7 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -442,7 +442,7 @@ def test_phoenix_disabled_by_default(self): assert len(obs.tracer._exporters) == 1 def test_cost_update_emits_llm_span(self): - """cost_update event creates an llm_call span for Arize.""" + """cost_update event creates an llm_call span with OpenInference attrs.""" obs = ObservabilityIntegration() obs._on_cost_update({ "model": "claude-opus-4-6", @@ -457,10 +457,113 @@ def test_cost_update_emits_llm_span(self): llm_spans = obs._in_memory.get_spans("llm_call") assert len(llm_spans) == 1 s = llm_spans[0] - assert s.attributes["llm.model"] == "claude-opus-4-6" - assert s.attributes["llm.token_count.input"] == 2000 + # OpenInference semantic conventions: + # https://github.com/Arize-ai/openinference/blob/main/spec/semantic_conventions.md + assert s.attributes["openinference.span.kind"] == "LLM" + assert s.attributes["llm.model_name"] == "claude-opus-4-6" + assert s.attributes["llm.provider"] == "anthropic" + assert s.attributes["llm.token_count.prompt"] == 2000 + assert s.attributes["llm.token_count.completion"] == 400 + assert s.attributes["llm.token_count.total"] == 2400 + assert s.attributes["llm.token_count.cached"] == 500 assert s.attributes["llm.cost_usd"] == 0.12 + def test_cost_update_records_elapsed_and_provider(self): + """The emitted LLM span's duration reflects elapsed_ms.""" + obs = ObservabilityIntegration() + obs._on_cost_update({ + "model": "claude-sonnet-4-6", + "provider": "anthropic", + "input_tokens": 100, + "output_tokens": 50, + "total_cost_usd": 0.001, + "elapsed_ms": 1234, + }) + obs.tracer.flush() + llm_spans = obs._in_memory.get_spans("llm_call") + assert len(llm_spans) == 1 + s = llm_spans[0] + assert s.attributes["llm.provider"] == "anthropic" + # duration_ms should be within a small tolerance of elapsed_ms (span + # closed roughly at "now"; start backdated by elapsed_ms). + assert abs(s.duration_ms - 1234) < 500 + + def test_bootstrap_from_env_noop_without_endpoint(self, monkeypatch): + monkeypatch.delenv("PHOENIX_ENDPOINT", raising=False) + monkeypatch.delenv("PHOENIX_PROJECT", raising=False) + # Clear any singleton left over from a prior test. + if ObservabilityIntegration._singleton is not None: + ObservabilityIntegration._singleton.disconnect() + assert ObservabilityIntegration.bootstrap_from_env() is None + assert ObservabilityIntegration.bootstrap_from_env() is None + assert ObservabilityIntegration._singleton is None + + def test_bootstrap_from_env_connects_when_endpoint_set(self, monkeypatch): + monkeypatch.setenv("PHOENIX_ENDPOINT", "http://phoenix:6006") + monkeypatch.setenv("PHOENIX_PROJECT", "clearwing-test") + if ObservabilityIntegration._singleton is not None: + ObservabilityIntegration._singleton.disconnect() + first = ObservabilityIntegration.bootstrap_from_env() + try: + assert first is not None + assert first._connected is True + second = ObservabilityIntegration.bootstrap_from_env() + assert second is first + finally: + if first is not None: + first.disconnect() + + def test_bootstrap_from_env_disconnect_cleans_up(self, monkeypatch): + monkeypatch.setenv("PHOENIX_ENDPOINT", "http://phoenix:6006") + monkeypatch.setenv("PHOENIX_PROJECT", "clearwing-test") + if ObservabilityIntegration._singleton is not None: + ObservabilityIntegration._singleton.disconnect() + first = ObservabilityIntegration.bootstrap_from_env() + assert first is not None + first.disconnect() + assert ObservabilityIntegration._singleton is None + second = ObservabilityIntegration.bootstrap_from_env() + try: + assert second is not None + assert second is not first + finally: + if second is not None: + second.disconnect() + + def test_phoenix_span_kind_derived_from_openinference(self): + """PhoenixExporter's _to_readable_span picks OTel kind from attrs.""" + from opentelemetry.trace import SpanKind + + from clearwing.observability.phoenix import _span_kind_from_openinference + from clearwing.observability.tracer import Span + + llm_span = Span( + trace_id="t1", + span_id="s1", + name="llm_call", + attributes={"openinference.span.kind": "LLM"}, + ) + assert _span_kind_from_openinference(llm_span) == SpanKind.CLIENT + + tool_span = Span( + trace_id="t1", + span_id="s2", + name="tool", + attributes={"openinference.span.kind": "TOOL"}, + ) + assert _span_kind_from_openinference(tool_span) == SpanKind.CLIENT + + chain_span = Span( + trace_id="t1", + span_id="s3", + name="chain", + attributes={"openinference.span.kind": "CHAIN"}, + ) + assert _span_kind_from_openinference(chain_span) == SpanKind.INTERNAL + + unset_span = Span(trace_id="t1", span_id="s4", name="unset") + assert _span_kind_from_openinference(unset_span) == SpanKind.INTERNAL + # --------------------------------------------------------------------------- # PhoenixExporter tests diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index edc55d22..2dbe8073 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -1,5 +1,6 @@ """Tests for the CostTracker telemetry module.""" +from clearwing.core.events import EventBus, EventType from clearwing.observability.telemetry import CostSummary, CostTracker, ToolUsage @@ -131,6 +132,59 @@ def test_reset(self): assert t.tool_calls == 0 assert t.by_tool == {} + def test_record_llm_call_emits_cost_update_with_elapsed_and_provider(self): + """New keyword args ride along in the COST_UPDATE payload.""" + received: list[dict] = [] + + def handler(data): + received.append(data) + + bus = EventBus() + bus.subscribe(EventType.COST_UPDATE, handler) + try: + t = CostTracker() + t.record_llm_call( + 1000, + 500, + "claude-sonnet-4-6", + cached_tokens=100, + elapsed_ms=1234.5, + provider="anthropic", + ) + finally: + bus.unsubscribe(EventType.COST_UPDATE, handler) + + assert len(received) == 1 + payload = received[0] + assert payload["input_tokens"] == 1000 + assert payload["output_tokens"] == 500 + assert payload["cached_tokens"] == 100 + assert payload["model"] == "claude-sonnet-4-6" + assert payload["provider"] == "anthropic" + assert payload["elapsed_ms"] == 1234.5 + + def test_record_llm_call_backward_compatible_without_new_kwargs(self): + """Legacy callers without the new kwargs still emit a valid payload.""" + received: list[dict] = [] + + def handler(data): + received.append(data) + + bus = EventBus() + bus.subscribe(EventType.COST_UPDATE, handler) + try: + t = CostTracker() + t.record_llm_call(1000, 500, "claude-sonnet-4-6") + finally: + bus.unsubscribe(EventType.COST_UPDATE, handler) + + assert len(received) == 1 + payload = received[0] + # Falls back to safe defaults. + assert payload["provider"] == "unknown" + assert payload["elapsed_ms"] == 0 + assert payload["cached_tokens"] == 0 + def test_pricing_table(self): assert "claude-sonnet-4-6" in CostTracker.PRICING assert "claude-opus-4-6" in CostTracker.PRICING