Publish live budget, token, turn and cache figures in telemetry

.agent/handoff/244.md

@@ -0,0 +1,30 @@
+# Handoff note: issue #244
+
+- issue: 244
+- started_at: 2026-10-14T14:12:55.198389Z
+- reason: token_soft_threshold
+- last_work_commit: 24d85f3236730f040cec81b2646e6aba687da873
+
+## Summary
+
+Implemented live attempt-figure publication for telemetry (issue #244, "publish the attempt's budget/token/turn/cache figures as they change"). All code changes are written but **uncommitted** (sealing will commit them):
+
+- `simple_coding_agent/token_ledger.py`: added `main_measured_response_count` and `main_context_basis()` (measured when the main thread's latest response reported usage, else estimated).
+- `simple_coding_agent/telemetry.py`: `TelemetryReporter.publish_attempt_view()` stamps `observed_at` and swaps the stored view under `_attempt_view_lock` (held only for the swap); `_build_current_attempt` serves the published view merged with the claim seed (full shape: budget used/measured/estimated + hard ceiling + soft threshold with rule/percentage/reserve_turns/crossed/crossed_at_tokens/crossed_at, tokens_by_category, turns, cache.main_hit_rate with counted, elapsed, provider_rate_limit, no USD). Without a published view it keeps the old tracer shape. `begin_attempt`/`finish_attempt` clear the view.
+- `simple_coding_agent/model_execution.py`: `ModelExecutor` takes injected `attempt_publisher` (ctor param + `set_attempt_publisher`, like `event_log`), publishes an immutable `MappingProxyType` view at the end of every `_check_limits`, records soft-threshold crossing wall-time, logs `telemetry_publish_failed` without affecting the attempt on publisher failure.
+- `simple_coding_agent/lifecycle.py`: `_wire_attempt_telemetry_publisher()` wires `reporter.publish_attempt_view` into the executor before dispatch; called alongside `_wire_attempt_runner_handoff()`.
+- `tests/test_telemetry_live_figures.py` (new, 13 tests): example-shape conformance for attempt-in-progress and soft-threshold-crossed, used-basis rule, hit-rate unavailability below minimum, observed_at/sequence advance, immutable per-check publication, failing-publish safety, frozen-after-execution, detached listener copy, begin/finish clearing, ledger helpers, lifecycle wiring.
+
+## Remaining work
+
+1. Invoke the `code-review` skill on the changes (required by the completion gate; an attempt with commits but no `code-review` invocation counts as incomplete).
+2. Run the completion gate in order: `ruff check`, full `pytest` on Python 3.12, full `pytest` on Python 3.13 (env vars redacted in the issue; PATH/HOME/UV_PYTHON_INSTALL_DIR/UV_PYTHON_DOWNLOADS/UV_NO_PROGRESS are set by the gate).
+3. Address any review/gate findings, then commit to `agent/issue-244`.
+
+## Evidence
+
+- New tests: `UV_PROJECT_ENVIRONMENT=/tmp/venv-py3-check uv run --python 3.12 --extra dev pytest tests/test_telemetry_live_figures.py` → 13 passed.
+- Related suites (`test_telemetry`, `test_telemetry_examples`, `test_model_execution`, `test_prompt_cache`, `test_lifecycle`) → 206 passed.
+- Full suite on 3.12 → 1099 passed, 10 skipped (one pre-existing `PytestUnhandledThreadExceptionWarning` about recovery, unrelated).
+- `ruff check` → all checks passed.
+- NOT yet run: `code-review` skill, full `pytest` on Python 3.13, gate with the official env.

simple_coding_agent/__main__.py

@@ -20,7 +20,10 @@
 from simple_coding_agent.git_workspace import GitWorkspace
 from simple_coding_agent.github_tracker import GitHubGraphQLTransport, GitHubTracker
 from simple_coding_agent.lifecycle import AgentLifecycle, AttemptInterruptionHandler, ModelAttemptRunner
-from simple_coding_agent.model_execution import ModelExecutor
+from simple_coding_agent.model_execution import (
+    PROMPT_CACHE_MIN_MEASURED_RESPONSES,
+    ModelExecutor,
+)
 from simple_coding_agent.observability import AttemptArchive, JsonEventLogger
 from simple_coding_agent.operating import ConsecutiveErrorStore
 from simple_coding_agent.publication import Publisher
@@ -110,6 +113,10 @@ def _attempt_sink(issue_number: int) -> AttemptArchive | None:
         turns_max=config.max_turns,
         timeout_seconds=config.model_timeout,
         continuation_max=config.max_auto_continuations,
+        # Single source of truth with the executor's prompt-cache warning:
+        # the served hit rate stays unavailable below this many measured
+        # main-thread responses.
+        cache_min_measured_responses=PROMPT_CACHE_MIN_MEASURED_RESPONSES,
     )
     lifecycle = AgentLifecycle(
         tracker=tracker,

simple_coding_agent/lifecycle.py

@@ -3379,6 +3379,7 @@ def run_once(self) -> LifecycleResult:
                 comment_posted = getattr(published, "comment_posted", True)
             else:
                 self._wire_attempt_runner_handoff()
+                self._wire_attempt_telemetry_publisher()
                 evidence = self._attempt_runner(claim, profile, prepared)
                 publication_started_at = self._clock()
                 published = self._publish_bounded(
@@ -3867,6 +3868,32 @@ def _wire_attempt_runner_handoff(self) -> None:
             return
         setter(_ControlOperatorHandoff(self._control_store, self._event_log))
 
+    def _wire_attempt_telemetry_publisher(self) -> None:
+        """Attach the telemetry reporter to the executor that publishes live figures.
+
+        The executor publishes an immutable attempt view after every limit
+        check; the reporter swaps the reference under a lock held only for
+        the swap, so the listener thread serves live figures without ever
+        reading the token ledger. Runners and executors without the seam
+        (older test doubles) are left alone. Never raises.
+        """
+
+        try:
+            reporter = self._telemetry
+            if reporter is None:
+                return
+            executor = getattr(self._attempt_runner, "_model_executor", None)
+            setter = getattr(executor, "set_attempt_publisher", None)
+            if not callable(setter):
+                return
+            setter(reporter.publish_attempt_view)
+        except Exception as error:
+            self._event_log(
+                "telemetry_publish_failed",
+                f"telemetry publisher wiring failed: {error}; continuing without live figures",
+                level="WARNING",
+            )
+
     def _handoff_publish_budget(self) -> float | None:
         """Bound one publication call by the pending handoff's remaining budget.
 

simple_coding_agent/model_execution.py

@@ -12,6 +12,7 @@
 import re
 import shlex
 import subprocess
+from types import MappingProxyType
 from typing import Any, Protocol
 
 from claude_agent_sdk import (
@@ -31,6 +32,7 @@
 )
 
 from simple_coding_agent.config import RuntimeConfig
+from simple_coding_agent.telemetry import format_timestamp as _format_timestamp
 from simple_coding_agent.token_ledger import (
     USAGE_CATEGORIES,
     TokenLedger,
@@ -455,6 +457,13 @@ def write_text(self, name: str, content: str) -> Path: ...
 Clock = Callable[[], datetime]
 EventLog = Callable[..., None]
 
+#: Live attempt figures published for telemetry after every limit check.
+#: The mapping is immutable; a raising publisher must not affect the
+#: attempt, so the executor logs ``telemetry_publish_failed`` and carries
+#: on. The driving process wires this to the telemetry reporter before
+#: dispatch (like ``event_log``); ``None`` disables publication.
+AttemptPublisher = Callable[[Mapping[str, Any]], None]
+
 _MAX_LOG_DETAIL = 2000
 # Bounded in-memory activity record for the emergency handoff note: the
 # recent tool trail keeps the last 15 calls, Bash history keeps the last 50
@@ -502,12 +511,14 @@ def __init__(
         clock: Clock = lambda: datetime.now(UTC),
         event_log: EventLog = lambda event, detail="", level="INFO", issue_number=None: None,
         cli_path: str | Path | None = None,
+        attempt_publisher: AttemptPublisher | None = None,
     ) -> None:
         self._config = config
         self._cli_path = Path(cli_path).expanduser() if cli_path is not None else None
         self._client_factory = client_factory
         self._clock = clock
         self._event_log = event_log
+        self._attempt_publisher = attempt_publisher
         self._skill_events: list[SkillEvent] = []
         self._review_count = 0
         self._review_nudge_sent = False
@@ -519,6 +530,7 @@ def __init__(
         self._started_at = self._clock()
         self._soft_threshold_crossed = False
         self._soft_threshold_crossed_at_tokens: float | None = None
+        self._soft_threshold_crossed_at: datetime | None = None
         self._soft_threshold_rule: str | None = None
         self._soft_threshold_effective_tokens: int | None = None
         self._total_cost_usd: float | None = None
@@ -589,6 +601,16 @@ def set_new_commits_provider(self, provider: NewCommitsProvider | None) -> None:
 
         self._new_commits_provider = provider
 
+    def set_attempt_publisher(self, publisher: AttemptPublisher | None) -> None:
+        """Publish immutable live attempt views for telemetry after every limit check.
+
+        The driving process wires this before dispatch (and only real
+        executors carry the seam; older fakes are left alone). ``None``
+        clears the publisher and disables publication.
+        """
+
+        self._attempt_publisher = publisher
+
     @property
     def review_nudge_sent(self) -> bool:
         """Whether the once-per-run review nudge was already delivered."""
@@ -975,6 +997,7 @@ async def execute(
         self._started_at = self._clock()
         self._soft_threshold_crossed = False
         self._soft_threshold_crossed_at_tokens = None
+        self._soft_threshold_crossed_at = None
         self._soft_threshold_rule = None
         self._soft_threshold_effective_tokens = None
         self._total_cost_usd = None
@@ -1939,6 +1962,7 @@ def _check_limits(self, thread_id: str = "main") -> None:
         if not self._soft_threshold_crossed and budget >= threshold:
             self._soft_threshold_crossed = True
             self._soft_threshold_crossed_at_tokens = budget
+            self._soft_threshold_crossed_at = self._clock()
             self._soft_threshold_rule = rule
             self._soft_threshold_effective_tokens = threshold
             self._log(
@@ -1962,6 +1986,72 @@ def _check_limits(self, thread_id: str = "main") -> None:
                 f"in_flight_estimated_tokens={self._ledger.estimated_tokens:.0f}",
                 level="WARNING",
             )
+        self._publish_attempt_view()
+
+    def _live_attempt_view(self) -> dict[str, Any]:
+        """Raw live readings for one telemetry publication (JSON-safe).
+
+        Built from the token ledger without mutating it: budget totals,
+        the effective soft threshold at the main thread's latest context,
+        measured per-category tokens over all threads, main-thread turns,
+        the main-thread cache hit rate with its measured-response count,
+        and elapsed seconds. No USD figure is included. The result is
+        deeply immutable: the nested ``tokens_by_category`` mapping is
+        frozen too, so the caller only needs the top-level
+        ``MappingProxyType`` wrap before publishing.
+        """
+
+        threshold, rule, _reserve_tokens = self._effective_threshold()
+        crossed_at = self._soft_threshold_crossed_at
+        return {
+            "budget_tokens": self._ledger.budget_tokens,
+            "measured_tokens": self._ledger.measured_tokens,
+            "estimated_tokens": self._ledger.estimated_tokens,
+            "hard_ceiling_reached": self._token_hard_ceiling_reached,
+            "soft_threshold_value": threshold,
+            "soft_threshold_rule": rule,
+            "soft_threshold_percentage": self._config.soft_threshold_percentage,
+            "soft_threshold_reserve_turns": self._config.handoff_reserve_turns,
+            "soft_threshold_crossed": self._soft_threshold_crossed,
+            "soft_threshold_crossed_at_tokens": (
+                None
+                if self._soft_threshold_crossed_at_tokens is None
+                else int(round(self._soft_threshold_crossed_at_tokens))
+            ),
+            "soft_threshold_crossed_at": (
+                None if crossed_at is None else _format_timestamp(crossed_at)
+            ),
+            "main_context_basis": self._ledger.main_context_basis(),
+            "tokens_by_category": MappingProxyType(dict(self._ledger.measured_by_category)),
+            "turns": (
+                self._ledger.response_count("main")
+                if self._ledger.has_thread("main")
+                else 0
+            ),
+            "cache_hit_rate": self._ledger.main_hit_rate(),
+            "cache_counted": self._ledger.main_measured_response_count,
+            "elapsed_seconds": self._elapsed_seconds(),
+        }
+
+    def _publish_attempt_view(self) -> None:
+        """Publish one immutable attempt view; a failure never affects the attempt.
+
+        Only the executor thread builds the view (the listener thread never
+        reads the ledger); the reporter swaps the reference under a lock
+        held just for the swap.
+        """
+
+        publisher = self._attempt_publisher
+        if publisher is None:
+            return
+        try:
+            publisher(MappingProxyType(self._live_attempt_view()))
+        except Exception as error:
+            self._log(
+                "telemetry_publish_failed",
+                f"attempt view publish failed: {error}; continuing without telemetry",
+                level="WARNING",
+            )
 
     async def _drain(self, client: SDKClient) -> ResultMessage | None:
         """Consume buffered messages after interrupting before client teardown.