feat(observability): instrument models, conductor, memory, ledger (Units 2-6)

Wire the observability facade through the load-bearing layers:
- models: llm.call/llm.structured spans with GenAI attrs, token/cost metrics,
full prompt+response captured at DEBUG (litellm, stub, openai-compat, router).
- conductor+governor: run/turn/agent.turn spans, run/turn/agent context binding,
event-append logs, agent-turn latency, governor budget-trip metrics+logs.
- memory: memory.recall + memory.index.search spans; DEBUG-logs the EXACT
salience-ranked memory text (with scores) each agent receives; index hit/latency
metrics and keyword-fallback logging; context.build structure logging.
- ledger/events/projections: per-append log+counter, projection.apply, invalid-kind.

Note: litellm_provider.py carries some pre-existing in-progress changes that ride
along with the instrumentation on this feature branch.

Co-Authored-By: Codex <codex@openai.com>

src/core/conductor.py

@@ -32,11 +32,13 @@ the observer never participates in cognition.
 from __future__ import annotations
 
 import logging
+import time
 from collections import deque
 from pathlib import Path
 from typing import TYPE_CHECKING
 from uuid import uuid4
 
+from src import observability as obs
 from src.core.events import Event
 from src.core.governor import BudgetExceeded, Governor
 from src.core.ledger import Ledger
@@ -96,6 +98,8 @@ class Conductor:
         self.run_id = str(uuid4())
         self.turn = 0
         self.governor.reset()
+        obs.set_context(run_id=self.run_id, turn=self.turn)
+        obs.log("run.started", run_id=self.run_id, seed=seed, goal=getattr(self.scenario, "goal", ""))
         genesis_start = Event(
             run_id=self.run_id,
             turn=self.turn,
@@ -158,6 +162,7 @@ class Conductor:
             self.turn += 1
             self.governor.begin_turn(self.turn)
             self.governor.check(self.turn)
+            obs.set_context(turn=self.turn)
             self._pending.extend(agent for agent, _ in self._trigger_queue)
             self._trigger_queue.clear()
             self._pending.extend(self._tick_scheduled_agents())
@@ -195,38 +200,45 @@ class Conductor:
         self.turn += 1
         self.governor.begin_turn(self.turn)
         self.governor.check(self.turn)
+        obs.set_context(turn=self.turn)
 
         projection = self.projection
 
-        # ── phase 1: event-triggered (subscription) agents ────────────────────
-        while self._trigger_queue:
-            agent, _trigger = self._trigger_queue.popleft()
-            self._run_agent(agent, projection)
+        with obs.span("turn", **{"mal.turn": self.turn}):
+            # ── phase 1: event-triggered (subscription) agents ────────────────
+            while self._trigger_queue:
+                agent, _trigger = self._trigger_queue.popleft()
+                self._run_agent(agent, projection)
 
-        # ── phase 2: tick-based scheduled agents ──────────────────────────────
-        for agent in self._tick_scheduled_agents():
-            self._run_agent(agent, projection)
+            # ── phase 2: tick-based scheduled agents ──────────────────────────
+            for agent in self._tick_scheduled_agents():
+                self._run_agent(agent, projection)
 
     def _run_agent(self, agent: "Agent", projection: StageProjection) -> None:
-        self.governor.check(self.turn)
-        try:
-            event = agent.act(
-                run_id=self.run_id,
-                turn=self.turn,
-                projection=projection,
-                recent_events=self.ledger.events,
-            )
-        except BudgetExceeded:
-            raise  # an intentional stop from the governor — never swallow it
-        except Exception as exc:  # noqa: BLE001 — one agent's crash must not silence the cast
-            self._note_agent_error(agent, exc)
-            return
-        usage = getattr(agent, "last_usage", {})
-        tokens = int(usage.get("total_tokens", 0) or 0)
-        cost_usd = float(usage.get("cost_usd", 0.0) or 0.0)
-        self.governor.record_call(tokens=tokens, cost_usd=cost_usd)
-        self._append(event)
-        projection.apply(event)
+        self.governor.check(self.turn)  # before the span: a budget stop is not an agent turn
+        name = getattr(agent, "name", agent.__class__.__name__)
+        start = time.perf_counter()
+        with obs.bind(agent=name), obs.span("agent.turn", **{"mal.agent": name, "mal.turn": self.turn}):
+            try:
+                event = agent.act(
+                    run_id=self.run_id,
+                    turn=self.turn,
+                    projection=projection,
+                    recent_events=self.ledger.events,
+                )
+            except BudgetExceeded:
+                raise  # an intentional stop from the governor — never swallow it
+            except Exception as exc:  # noqa: BLE001 — one agent's crash must not silence the cast
+                self._note_agent_error(agent, exc)
+                return
+            usage = getattr(agent, "last_usage", {})
+            tokens = int(usage.get("total_tokens", 0) or 0)
+            cost_usd = float(usage.get("cost_usd", 0.0) or 0.0)
+            obs.add_span_attrs(**{"event.kind": event.kind, "mal.tokens": tokens, "mal.cost_usd": cost_usd})
+            self.governor.record_call(tokens=tokens, cost_usd=cost_usd)
+            self._append(event)
+            projection.apply(event)
+        obs.record_agent_turn(name, time.perf_counter() - start)
 
     def _note_agent_error(self, agent: "Agent", exc: Exception) -> None:
         """Record (and log) an agent's failed turn without aborting the tick.
@@ -237,6 +249,7 @@ class Conductor:
         name = getattr(agent, "name", agent.__class__.__name__)
         self.agent_errors.append({"turn": str(self.turn), "agent": name, "error": str(exc)})
         logger.warning("agent %s failed on turn %d: %s", name, self.turn, exc, exc_info=exc)
+        obs.log("agent.error", level="warning", agent=name, turn=self.turn, error=str(exc))
 
     def _maybe_snapshot(self) -> None:
         if not self.snapshot_every or not self.snapshot_path:
@@ -249,6 +262,9 @@ class Conductor:
 
     def _append(self, event: Event) -> Event:
         appended = self.ledger.append(event)
+        obs.log(
+            "event.append", level="debug", id=appended.id, kind=appended.kind, actor=appended.actor, turn=appended.turn
+        )
         if self.observer:
             self.observer.consume(appended)
         self._notify_subscribers(appended)

src/core/context.py

@@ -17,6 +17,7 @@ Changing the prompt strategy for all agents is a one-file edit here.
 
 from __future__ import annotations
 
+from src import observability as obs
 from src.core.events import Event
 from src.core.memory import EpisodicMemory
 from src.core.projections import StageProjection
@@ -53,7 +54,7 @@ class ContextBuilder:
 
         goal_block = f"SHARED GOAL\n{projection.goal}\n\n" if projection.goal else ""
 
-        return (
+        prompt = (
             f"IDENTITY\n{persona}\n\n"
             f"{goal_block}"
             f"CURRENT SCENE\n{projection.current_scene}\n\n"
@@ -61,6 +62,20 @@ class ContextBuilder:
             f"YOUR MEMORY (recent events you witnessed)\n{memory_text}\n\n"
             f"VISITOR DISTURBANCES\n{visitor_lines}"
         )
+        # Structure + size of the assembled context (the full prompt is logged by the
+        # agent layer as ``agent.prompt``; here we record which sections were present).
+        sections = ["IDENTITY", "CURRENT SCENE", "BLACKBOARD", "MEMORY", "VISITOR"]
+        if goal_block:
+            sections.insert(1, "SHARED GOAL")
+        obs.log(
+            "context.build",
+            level="debug",
+            agent=agent_name,
+            sections=sections,
+            prompt_chars=len(prompt),
+            memory_chars=len(memory_text),
+        )
+        return prompt
 
     @staticmethod
     def _blackboard_block(agent_notes: list[str], window: int = 6) -> str:

src/core/events.py

@@ -7,6 +7,8 @@ from uuid import uuid4
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
+from src import observability as obs
+
 # ── event kinds ───────────────────────────────────────────────────────────────
 #
 # `kind` is an OPEN, format-validated string — NOT a closed enum.  This is the
@@ -58,6 +60,7 @@ class Event(BaseModel):
     @classmethod
     def _validate_kind(cls, value: str) -> str:
         if not is_valid_kind(value):
+            obs.log("event.invalid", level="warning", kind=value)
             raise ValueError(
                 f"invalid event kind {value!r}: must be a lowercase, dot-namespaced "
                 "identifier such as 'agent.spoke' or 'clue.found'"

src/core/governor.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 
+from src import observability as obs
+
 
 @dataclass
 class Governor:
@@ -35,18 +37,29 @@ class Governor:
 
     def check(self, turn: int) -> None:
         if turn > self.max_turns:
-            raise BudgetExceeded(f"Turn cap {self.max_turns} reached", reason="max_turns")
+            self._trip("max_turns", f"Turn cap {self.max_turns} reached")
         if self._total_calls >= self.max_total_calls:
-            raise BudgetExceeded(f"Total call cap {self.max_total_calls} reached", reason="max_total_calls")
+            self._trip("max_total_calls", f"Total call cap {self.max_total_calls} reached")
         if self._calls_this_turn >= self.max_calls_per_turn:
-            raise BudgetExceeded(
-                f"Per-turn call cap {self.max_calls_per_turn} reached on turn {turn}",
-                reason="max_calls_per_turn",
-            )
+            self._trip("max_calls_per_turn", f"Per-turn call cap {self.max_calls_per_turn} reached on turn {turn}")
         if self.max_total_tokens is not None and self._total_tokens >= self.max_total_tokens:
-            raise BudgetExceeded(f"Total token cap {self.max_total_tokens} reached", reason="max_total_tokens")
+            self._trip("max_total_tokens", f"Total token cap {self.max_total_tokens} reached")
         if self.hourly_budget_usd is not None and self._spend_usd >= self.hourly_budget_usd:
-            raise BudgetExceeded(f"Spend cap ${self.hourly_budget_usd:.2f} reached", reason="hourly_budget_usd")
+            self._trip("hourly_budget_usd", f"Spend cap ${self.hourly_budget_usd:.2f} reached")
+
+    def _trip(self, reason: str, message: str) -> None:
+        """Record the budget trip as a metric + log, then raise the stop."""
+        obs.record_governor_trip(reason)
+        obs.log(
+            "governor.trip",
+            level="warning",
+            reason=reason,
+            message=message,
+            total_calls=self._total_calls,
+            total_tokens=self._total_tokens,
+            spend_usd=round(self._spend_usd, 4),
+        )
+        raise BudgetExceeded(message, reason=reason)
 
     def record_call(self, tokens: int = 0, cost_usd: float = 0.0) -> None:
         self._calls_this_turn += 1

src/core/ledger.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Iterable
 
+from src import observability as obs
 from src.core.events import Event
 
 
@@ -21,6 +22,8 @@ class Ledger:
             return event
         self._events.append(event)
         self._seen_ids.add(event.id)
+        obs.log("ledger.append", level="debug", id=event.id, kind=event.kind, actor=event.actor, turn=event.turn)
+        obs.incr("ledger.events", 1, kind=event.kind)
         return event
 
     def extend(self, events: Iterable[Event]) -> None:
@@ -28,6 +31,6 @@ class Ledger:
             self.append(event)
 
     def reset(self) -> None:
+        obs.log("ledger.reset", level="debug", events=len(self._events))
         self._events.clear()
         self._seen_ids.clear()
-

src/core/memory.py

@@ -35,6 +35,7 @@ import math
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
+from src import observability as obs
 from src.core.events import Event
 
 if TYPE_CHECKING:  # pragma: no cover - typing only
@@ -92,9 +93,22 @@ class EpisodicMemory:
         return result[-self.max_recent :]
 
     def format_for_prompt(self, events: tuple[Event, ...]) -> str:
-        recalled = self.visible(events)
-        lines = [f"[turn {e.turn:03d}][{e.kind}] {text}" for e in recalled if (text := _displayable(e))]
-        return "\n".join(lines) if lines else "(no prior memory)"
+        with obs.span("memory.recall", **{"mal.agent": self.agent_name, "memory.mode": "episodic"}):
+            recalled = self.visible(events)
+            lines = [f"[turn {e.turn:03d}][{e.kind}] {text}" for e in recalled if (text := _displayable(e))]
+            memory = "\n".join(lines) if lines else "(no prior memory)"
+            obs.add_span_attrs(**{"memory.visible_count": len(recalled)})
+            obs.observe("memory.visible_count", len(recalled), agent=self.agent_name)
+            # DEBUG: the EXACT memory string this agent will receive (what it "sees").
+            obs.log(
+                "memory.recall",
+                level="debug",
+                agent=self.agent_name,
+                mode="episodic",
+                visible_count=len(recalled),
+                memory=memory,
+            )
+            return memory
 
 
 # ── layer 2: salience-scored memory ──────────────────────────────────────────
@@ -175,6 +189,7 @@ class SalienceMemory:
             hits = self.index.search(query, k=len(candidates))
         except Exception as exc:  # noqa: BLE001 — relevance is best-effort, never fatal
             logger.warning("memory index unavailable, using keyword relevance: %s", exc)
+            obs.log("memory.index.fallback", level="warning", agent=self.agent_name, error=str(exc))
             return None
         eligible = {e.id for e in candidates}
         ranked = [h.id for h in hits if h.id in eligible]
@@ -201,19 +216,47 @@ class SalienceMemory:
         return sorted(top, key=lambda e: e.turn)
 
     def format_for_prompt(self, events: tuple[Event, ...], current_turn: int, query: str) -> str:
-        candidates = self._candidates(events)
-        relevance = self._relevance_map(candidates, query)
-
-        def _score(e: Event) -> float:
-            rel = None if relevance is None else relevance.get(e.id, 0.0)
-            return self.score(e, current_turn, query, relevance=rel)
-
-        top = sorted(candidates, key=_score, reverse=True)[: self.top_k]
-        recalled = sorted(top, key=lambda e: e.turn)
-        lines = [
-            f"[turn {e.turn:03d}][{e.kind}][sal={_score(e):.2f}] {text}" for e in recalled if (text := _displayable(e))
-        ]
-        return "\n".join(lines) if lines else "(no salient memories)"
+        with obs.span(
+            "memory.recall",
+            **{"mal.agent": self.agent_name, "memory.mode": "salience", "memory.top_k": self.top_k},
+        ):
+            candidates = self._candidates(events)
+            relevance = self._relevance_map(candidates, query)
+
+            def _score(e: Event) -> float:
+                rel = None if relevance is None else relevance.get(e.id, 0.0)
+                return self.score(e, current_turn, query, relevance=rel)
+
+            top = sorted(candidates, key=_score, reverse=True)[: self.top_k]
+            recalled = sorted(top, key=lambda e: e.turn)
+            lines = [
+                f"[turn {e.turn:03d}][{e.kind}][sal={_score(e):.2f}] {text}"
+                for e in recalled
+                if (text := _displayable(e))
+            ]
+            memory = "\n".join(lines) if lines else "(no salient memories)"
+            scores = {e.id: round(_score(e), 3) for e in recalled}
+            obs.add_span_attrs(
+                **{
+                    "memory.visible_count": len(recalled),
+                    "memory.query": query,
+                    "memory.semantic": relevance is not None,
+                }
+            )
+            obs.observe("memory.visible_count", len(recalled), agent=self.agent_name)
+            # DEBUG: the EXACT salience-ranked memory this agent will receive, with scores.
+            obs.log(
+                "memory.recall",
+                level="debug",
+                agent=self.agent_name,
+                mode="salience",
+                query=query,
+                visible_count=len(recalled),
+                semantic=relevance is not None,
+                scores=scores,
+                memory=memory,
+            )
+            return memory
 
 
 # ── layer 3: reflection trigger ───────────────────────────────────────────────

src/core/memory_index.py

@@ -35,8 +35,10 @@ duplicates) — this is what makes the index rebuildable rather than authoritati
 from __future__ import annotations
 
 import os
+import time
 from typing import TYPE_CHECKING, Protocol, runtime_checkable
 
+from src import observability as obs
 from src.core.events import Event
 
 if TYPE_CHECKING:  # pragma: no cover - typing only
@@ -162,13 +164,31 @@ class _Mem0BackendBase:
         """Semantic search; map hits back to :class:`Event` via stored metadata."""
         if not query or k <= 0:
             return []
-        mem = self._memory()
-        events: list[Event] = []
-        for hit in self._query(mem, query, k):
-            event = _event_from_metadata(hit.get("metadata"))
-            if event is not None:
-                events.append(event)
-        return events
+        with obs.span(
+            "memory.index.search",
+            **{"memory.query": query, "memory.k": k, "memory.backend": type(self).__name__},
+        ):
+            started = time.perf_counter()
+            mem = self._memory()
+            events: list[Event] = []
+            for hit in self._query(mem, query, k):
+                event = _event_from_metadata(hit.get("metadata"))
+                if event is not None:
+                    events.append(event)
+            elapsed_ms = (time.perf_counter() - started) * 1000
+            obs.add_span_attrs(**{"memory.hits": len(events), "memory.latency_ms": round(elapsed_ms, 2)})
+            obs.observe("memory.index.hits", len(events))
+            obs.observe("memory.index.latency_ms", elapsed_ms)
+            obs.log(
+                "memory.index.search",
+                level="debug",
+                backend=type(self).__name__,
+                query=query,
+                k=k,
+                hits=len(events),
+                latency_ms=round(elapsed_ms, 2),
+            )
+            return events
 
 
 # ── local (off-the-grid) backend ──────────────────────────────────────────────

src/core/projections.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 
+from src import observability as obs
 from src.core.events import Event
 
 
@@ -15,6 +16,7 @@ class StageProjection:
     user_artifacts: list[str] = field(default_factory=list)
 
     def apply(self, event: Event) -> None:
+        obs.log("projection.apply", level="debug", kind=event.kind, actor=event.actor, turn=event.turn)
         if event.kind == "run.started":
             self.seed = str(event.payload["seed"])
             self.goal = str(event.payload.get("goal", "")) or self.goal

src/models/litellm_provider.py

@@ -36,6 +36,7 @@ from __future__ import annotations
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
+from src import observability as obs
 from src.models.openai_compat import OpenAICompatProvider
 from src.models.provider import ModelProvider, model_error
 
@@ -64,28 +65,40 @@ class LiteLLMProvider(ModelProvider):
     """Transport retries LiteLLM makes on a transient call failure — a dropped
     connection, a timeout, a 5xx.  Lets a flaky endpoint self-heal mid-demo before the
     call gives up and returns the failure sentinel."""
+    structured_mode: str = "json_schema"
+    """Instructor mode for :meth:`complete_structured` (an ``instructor.Mode`` member name,
+    case-insensitive).  Defaults to ``json_schema`` — vLLM **guided decoding** via
+    ``response_format``, which is parser-independent: it constrains the output to the schema
+    (``kind`` can't be an unauthorised value) without needing a tool-call parser.  This is
+    deliberate: not every served model ships a tool parser (e.g. MiniCPM4.1 emits a custom
+    ``<|tool_call_start|>`` format vLLM 0.21.0 has no parser for), so Instructor's default
+    ``tools`` mode 400s there.  ``json`` (plain ``json_object`` + schema-in-prompt) is the
+    fallback if a backend rejects ``json_schema``; ``tools`` restores the old behaviour."""
     _last_usage: dict = field(default_factory=dict, init=False, repr=False)
     _last_cost: float = field(default=0.0, init=False, repr=False)
     _last_reasoning: str = field(default="", init=False, repr=False)
 
     def complete(self, role: str, prompt: str) -> str:
         litellm = self._litellm()
-        try:
-            response = litellm.completion(
-                model=self.model,
-                api_base=self.api_base,
-                api_key=self._resolved_api_key(),
-                messages=self._messages(role, prompt),
-                temperature=self.temperature,
-                max_tokens=self.max_tokens,
-                num_retries=self.num_retries,
-            )
-            text = (response.choices[0].message.content or "").strip()
-            self._capture_usage(litellm, response, prompt, text)
-            return text
-        except Exception as exc:
-            self._zero_usage()
-            return model_error(exc)
+        with obs.span("llm.call", **self._span_request_attrs(role)):
+            try:
+                response = litellm.completion(
+                    model=self.model,
+                    api_base=self.api_base,
+                    api_key=self._resolved_api_key(),
+                    messages=self._messages(role, prompt),
+                    temperature=self.temperature,
+                    max_tokens=self.max_tokens,
+                    num_retries=self.num_retries,
+                )
+                text = (response.choices[0].message.content or "").strip()
+                self._capture_usage(litellm, response, prompt, text)
+                self._emit_telemetry(role, prompt, text, structured=False)
+                return text
+            except Exception as exc:
+                self._zero_usage()
+                obs.log("llm.error", level="warning", model=self.model, role=role, error=str(exc))
+                return model_error(exc)
 
     def complete_structured(
         self,
@@ -115,25 +128,32 @@ class LiteLLMProvider(ModelProvider):
                 "instructor package is required for complete_structured(). Install it with: uv pip install instructor"
             ) from exc
 
-        client = instructor.from_litellm(litellm.completion)
-        try:
-            result, response = client.create_with_completion(
-                model=self.model,
-                api_base=self.api_base,
-                api_key=self._resolved_api_key(),
-                messages=self._messages(role, prompt),
-                response_model=response_model,
-                max_retries=self.max_retries,
-                num_retries=self.num_retries,
-                temperature=self.temperature,
-                max_tokens=self.max_tokens,
-            )
-            text = getattr(result, "text", "") or ""
-            self._capture_usage(litellm, response, prompt, text)
-            return result
-        except Exception:
-            self._zero_usage()
-            raise
+        # Guided-JSON by default (see ``structured_mode``): constrain the output to the
+        # schema via vLLM's ``response_format`` rather than tool calling, so a model with no
+        # tool-call parser still returns a validated payload instead of a 400.
+        mode = getattr(instructor.Mode, self.structured_mode.upper(), instructor.Mode.JSON_SCHEMA)
+        client = instructor.from_litellm(litellm.completion, mode=mode)
+        with obs.span("llm.structured", **{**self._span_request_attrs(role), "llm.mode": self.structured_mode}):
+            try:
+                result, response = client.create_with_completion(
+                    model=self.model,
+                    api_base=self.api_base,
+                    api_key=self._resolved_api_key(),
+                    messages=self._messages(role, prompt),
+                    response_model=response_model,
+                    max_retries=self.max_retries,
+                    num_retries=self.num_retries,
+                    temperature=self.temperature,
+                    max_tokens=self.max_tokens,
+                )
+                text = getattr(result, "text", "") or ""
+                self._capture_usage(litellm, response, prompt, text)
+                self._emit_telemetry(role, prompt, text, structured=True)
+                return result
+            except Exception as exc:
+                self._zero_usage()
+                obs.log("llm.error", level="warning", model=self.model, role=role, structured=True, error=str(exc))
+                raise
 
     @property
     def last_reasoning(self) -> str:
@@ -150,6 +170,65 @@ class LiteLLMProvider(ModelProvider):
         """Metered USD cost of the most recent call (0.0 offline)."""
         return self._last_cost
 
+    # ── telemetry (shared by complete / complete_structured) ────────────────────
+
+    def _span_request_attrs(self, role: str) -> dict:
+        """GenAI request attributes for an LLM span — never includes the api key."""
+        return {
+            "gen_ai.system": "litellm",
+            "gen_ai.request.model": self.model,
+            "gen_ai.request.temperature": self.temperature,
+            "gen_ai.request.max_tokens": self.max_tokens,
+            "llm.api_base": self.api_base or "",
+            "mal.role": role,
+        }
+
+    def _emit_telemetry(self, role: str, prompt: str, text: str, *, structured: bool) -> None:
+        """Attach usage/cost/prompt to the active span, count the call, and log it.
+
+        The full prompt + completion + reasoning ride on the span (truncated in the
+        UI store) and on a DEBUG ``llm.exchange`` log, so a reviewer can read exactly
+        what was sent to each model. INFO ``llm.call`` carries the metered summary.
+        """
+        usage = self._last_usage or {}
+        prompt_tokens = int(usage.get("prompt_tokens", 0) or 0)
+        completion_tokens = int(usage.get("completion_tokens", 0) or 0)
+        obs.add_span_attrs(
+            **{
+                "gen_ai.usage.input_tokens": prompt_tokens,
+                "gen_ai.usage.output_tokens": completion_tokens,
+                "llm.cost_usd": self._last_cost,
+                "llm.structured": structured,
+                "llm.prompt": prompt,
+                "llm.completion": text,
+                "llm.reasoning": self._last_reasoning or "",
+            }
+        )
+        obs.record_llm_call(
+            self.model,
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            cost_usd=self._last_cost,
+        )
+        obs.log(
+            "llm.call",
+            role=role,
+            model=self.model,
+            structured=structured,
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            cost_usd=round(self._last_cost, 6),
+        )
+        obs.log(
+            "llm.exchange",
+            level="debug",
+            role=role,
+            model=self.model,
+            prompt=prompt,
+            completion=text,
+            reasoning=self._last_reasoning or "",
+        )
+
     # ── call helpers (shared by complete / complete_structured) ─────────────────
 
     @staticmethod

src/models/openai_compat.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import os
 from dataclasses import dataclass, field
 
+from src import observability as obs
 from src.models.provider import ModelProvider, model_error
 
 
@@ -72,31 +73,61 @@ class OpenAICompatProvider(ModelProvider):
 
         client = self._get_client()
         system = self._system_for_role(role)
-        try:
-            resp = client.chat.completions.create(
-                model=self.model,
-                messages=[
-                    {"role": "system", "content": system},
-                    {"role": "user", "content": prompt},
-                ],
-                max_tokens=self.max_tokens,
-                temperature=self.temperature,
-            )
-            text = resp.choices[0].message.content.strip()
-            usage = getattr(resp, "usage", None)
-            if usage is not None:
-                self._last_usage = {
-                    "prompt_tokens": getattr(usage, "prompt_tokens", 0) or 0,
-                    "completion_tokens": getattr(usage, "completion_tokens", 0) or 0,
-                    "total_tokens": getattr(usage, "total_tokens", 0) or 0,
-                }
-            else:
-                p, c = estimate_tokens(prompt), estimate_tokens(text)
-                self._last_usage = {"prompt_tokens": p, "completion_tokens": c, "total_tokens": p + c}
-            return text
-        except Exception as exc:
-            self._last_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
-            return model_error(exc)
+        span_attrs = {
+            "gen_ai.system": "openai-compatible",
+            "gen_ai.request.model": self.model,
+            "llm.api_base": self.base_url or "",
+            "mal.role": role,
+        }
+        with obs.span("llm.call", **span_attrs):
+            try:
+                resp = client.chat.completions.create(
+                    model=self.model,
+                    messages=[
+                        {"role": "system", "content": system},
+                        {"role": "user", "content": prompt},
+                    ],
+                    max_tokens=self.max_tokens,
+                    temperature=self.temperature,
+                )
+                text = resp.choices[0].message.content.strip()
+                usage = getattr(resp, "usage", None)
+                if usage is not None:
+                    self._last_usage = {
+                        "prompt_tokens": getattr(usage, "prompt_tokens", 0) or 0,
+                        "completion_tokens": getattr(usage, "completion_tokens", 0) or 0,
+                        "total_tokens": getattr(usage, "total_tokens", 0) or 0,
+                    }
+                else:
+                    p, c = estimate_tokens(prompt), estimate_tokens(text)
+                    self._last_usage = {"prompt_tokens": p, "completion_tokens": c, "total_tokens": p + c}
+                obs.add_span_attrs(
+                    **{
+                        "gen_ai.usage.input_tokens": int(self._last_usage["prompt_tokens"]),
+                        "gen_ai.usage.output_tokens": int(self._last_usage["completion_tokens"]),
+                        "llm.prompt": prompt,
+                        "llm.completion": text,
+                    }
+                )
+                obs.record_llm_call(
+                    self.model,
+                    prompt_tokens=int(self._last_usage["prompt_tokens"]),
+                    completion_tokens=int(self._last_usage["completion_tokens"]),
+                )
+                obs.log(
+                    "llm.call",
+                    role=role,
+                    model=self.model,
+                    structured=False,
+                    prompt_tokens=int(self._last_usage["prompt_tokens"]),
+                    completion_tokens=int(self._last_usage["completion_tokens"]),
+                )
+                obs.log("llm.exchange", level="debug", role=role, model=self.model, prompt=prompt, completion=text)
+                return text
+            except Exception as exc:
+                self._last_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
+                obs.log("llm.error", level="warning", model=self.model, role=role, error=str(exc))
+                return model_error(exc)
 
     @staticmethod
     def _system_for_role(role: str) -> str:

src/models/provider.py

@@ -5,6 +5,8 @@ import json
 import re
 from dataclasses import dataclass, field
 
+from src import observability as obs
+
 
 def estimate_tokens(text: str) -> int:
     """Rough token estimate (~4 chars/token) for providers without usage data.
@@ -167,6 +169,10 @@ class DeterministicTinyModel(ModelProvider):
     _last_usage: dict[str, int] = field(default_factory=dict, init=False, repr=False)
 
     def complete(self, role: str, prompt: str) -> str:
+        with obs.span("llm.call", **{"gen_ai.system": "stub", "gen_ai.request.model": self.variant, "mal.role": role}):
+            return self._complete(role, prompt)
+
+    def _complete(self, role: str, prompt: str) -> str:
         digest = hashlib.sha256(f"{self.variant}:{role}:{prompt}".encode("utf-8")).hexdigest()
         choices = {
             "scene-whisperer": [
@@ -243,11 +249,33 @@ class DeterministicTinyModel(ModelProvider):
                     obj[name] = self._synth_field(name, role, digest)
                 out = json.dumps(obj, ensure_ascii=False)
 
+        prompt_tokens, completion_tokens = estimate_tokens(prompt), estimate_tokens(out)
         self._last_usage = {
-            "prompt_tokens": estimate_tokens(prompt),
-            "completion_tokens": estimate_tokens(out),
-            "total_tokens": estimate_tokens(prompt) + estimate_tokens(out),
+            "prompt_tokens": prompt_tokens,
+            "completion_tokens": completion_tokens,
+            "total_tokens": prompt_tokens + completion_tokens,
         }
+        obs.add_span_attrs(
+            **{
+                "gen_ai.usage.input_tokens": prompt_tokens,
+                "gen_ai.usage.output_tokens": completion_tokens,
+                "llm.prompt": prompt,
+                "llm.completion": out,
+            }
+        )
+        obs.record_llm_call(
+            self.variant, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, cost_usd=0.0
+        )
+        obs.log(
+            "llm.call",
+            role=role,
+            model=self.variant,
+            structured=False,
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            cost_usd=0.0,
+        )
+        obs.log("llm.exchange", level="debug", role=role, model=self.variant, prompt=prompt, completion=out)
         return out
 
     def _synth_field(self, name: str, role: str, digest: str) -> str:

src/models/router.py

@@ -23,6 +23,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 
+from src import observability as obs
 from src.core.manifest import ModelProfile, resolve_model
 from src.models.openai_compat import has_live_credentials
 from src.models.provider import DeterministicTinyModel, ModelProvider
@@ -86,12 +87,15 @@ class ModelRouter:
 
     def _build(self, profile: str) -> ModelProvider:
         if self.offline:
+            obs.log("router.resolve", profile=profile, mode="offline", model=f"stub:{profile}")
             return DeterministicTinyModel(variant=f"stub:{profile}")
         # Live transport is the LiteLLM gateway (ADR-0015).  Lazy-import keeps the
         # offline path free of the dependency.
         from src.models.litellm_provider import LiteLLMProvider
 
         spec = self._spec_for(profile)
+        # Resolution is logged WITHOUT the api key — only the model + endpoint.
+        obs.log("router.resolve", profile=profile, mode="live", model=spec.model, api_base=spec.base_url or "")
         return LiteLLMProvider(
             model=spec.model,
             api_base=spec.base_url,

tests/test_observability.py

@@ -45,14 +45,18 @@ def test_log_sanitises_reserved_field_names():
 
 def test_context_binding_stamps_logs():
     _fresh()
+    # `set_context` (used by the conductor for run/turn) persists by design, so the
+    # process context may be non-empty here; capture it and assert bind RESTORES it.
+    before = obs.current_context().get("run_id")
     with obs.bind(run_id="run-7", turn=2, agent="clue-gatherer"):
+        assert obs.current_context()["run_id"] == "run-7"
         obs.log("memory.recall", k=5)
     record = [r for r in obs.telemetry_store().recent_logs() if r.get("event") == "memory.recall"][-1]
     assert record["run_id"] == "run-7"
     assert record["turn"] == 2
     assert record["agent"] == "clue-gatherer"
-    # Binding is scoped — it clears on exit.
-    assert "run_id" not in obs.current_context()
+    # Binding is scoped — run_id returns to whatever it was before the block.
+    assert obs.current_context().get("run_id") == before
 
 
 def test_span_recorded_with_attributes_and_nesting():