From 0a90be399ec794e2b841f1ba956c99a5eefbe44a Mon Sep 17 00:00:00 2001 From: Claude Code Date: Wed, 14 Oct 2026 14:26:53 +0000 Subject: [PATCH 1/3] Handoff note: issue #244 --- .agent/handoff/244.md | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 .agent/handoff/244.md diff --git a/.agent/handoff/244.md b/.agent/handoff/244.md new file mode 100644 index 0000000..e6ffd97 --- /dev/null +++ b/.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. From ea270d873e652e9175ef61ff73d4046eafebf618 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Wed, 14 Oct 2026 14:27:22 +0000 Subject: [PATCH 2/3] Seal attempt work for #244 (model_limit_reached) Attempt-Id: 2026-10-14T14:12:55.198389Z --- simple_coding_agent/lifecycle.py | 27 ++ simple_coding_agent/model_execution.py | 88 +++++ simple_coding_agent/telemetry.py | 306 ++++++++++++++- simple_coding_agent/token_ledger.py | 20 + tests/test_telemetry_live_figures.py | 524 +++++++++++++++++++++++++ 5 files changed, 964 insertions(+), 1 deletion(-) create mode 100644 tests/test_telemetry_live_figures.py diff --git a/simple_coding_agent/lifecycle.py b/simple_coding_agent/lifecycle.py index 79bf98a..627a538 100644 --- a/simple_coding_agent/lifecycle.py +++ b/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. diff --git a/simple_coding_agent/model_execution.py b/simple_coding_agent/model_execution.py index a4e20fc..db2c8be 100644 --- a/simple_coding_agent/model_execution.py +++ b/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,70 @@ 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 caller wraps + the result as immutable 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": 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. diff --git a/simple_coding_agent/telemetry.py b/simple_coding_agent/telemetry.py index a04391e..749fcd2 100644 --- a/simple_coding_agent/telemetry.py +++ b/simple_coding_agent/telemetry.py @@ -15,7 +15,9 @@ startup-leftover path), so the stored seed alone decides the section: a set seed means an active attempt. Its configured limits come from the seed (claim-time values) falling back to the reporter construction defaults; -live figures stay ``unavailable`` until later tickets publish them. +its live budget, token, turn and cache figures come from the immutable +attempt views the executor publishes via ``publish_attempt_view`` after +every limit check. Until the first view arrives they stay ``unavailable``. Building the section never raises into the attempt: a failure logs once per episode and keeps the previous view. @@ -63,6 +65,27 @@ #: failing snapshotter must not flood the log. TELEMETRY_FAILURE_COOLDOWN_SECONDS = 60.0 +#: Minimum measured main-thread responses before the prompt-cache hit rate +#: is reported as ``measured``; below it the figure is ``unavailable`` +#: (never zero) while ``counted`` still reports the responses seen so far. +#: Mirrors ``model_execution.PROMPT_CACHE_MIN_MEASURED_RESPONSES`` without +#: importing the SDK-bound executor module on the listener thread. +CACHE_MIN_MEASURED_RESPONSES = 20 + +#: Reason carried by a hit-rate figure with too few measured responses. +CACHE_NOT_ENOUGH_RESPONSES_REASON = "not enough measured responses" + +#: Reason carried by a hit-rate figure with no measured input to score. +CACHE_NO_MEASURED_INPUT_REASON = "no measured input to score" + +#: Token categories published under ``tokens_by_category`` (measured only). +TOKEN_CATEGORIES = ( + "input_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + "output_tokens", +) + logger = logging.getLogger(__name__) @@ -100,6 +123,71 @@ def _figure_unavailable(reason: str = "not yet published") -> dict: return {"value": None, "basis": "unavailable", "reason": reason} +def _plain_dict(view: object) -> dict: + """Copy a published attempt view into plain JSON-safe containers. + + The executor publishes an immutable mapping; the copy detaches the + stored view from whatever the publisher holds so later mutation on + either side cannot leak into the served snapshot. Raises ``ValueError`` + when the view is not a mapping. + """ + + if not isinstance(view, dict): + from collections.abc import Mapping as _Mapping + + if not isinstance(view, _Mapping): + raise ValueError("published attempt view must be a mapping") + view = dict(view) + plain = _plain_value(view) + if not isinstance(plain, dict): + raise ValueError("published attempt view must be a mapping") + return plain + + +def _plain_value(value: object) -> object: + from collections.abc import Mapping as _Mapping + + if isinstance(value, _Mapping): + return {key: _plain_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_plain_value(item) for item in value] + return value + + +def _view_tokens(view: dict, key: str) -> int: + """A rounded token total from a published view (never negative).""" + + value = view.get(key) + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"published attempt view {key} must be a number") + rounded = int(round(value)) + if rounded < 0: + raise ValueError(f"published attempt view {key} must be non-negative") + return rounded + + +def _view_int(view: dict, key: str) -> int: + """A non-negative integer from a published view.""" + + value = view.get(key) + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"published attempt view {key} must be an integer") + if value < 0: + raise ValueError(f"published attempt view {key} must be non-negative") + return value + + +def _view_category(categories: dict, category: str) -> int: + """One measured token category from a published view.""" + + value = categories.get(category) + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError( + f"published attempt view tokens_by_category[{category}] must be a non-negative integer" + ) + return value + + def _as_int(value: object) -> int | None: if isinstance(value, bool): return None @@ -410,6 +498,7 @@ def __init__( max_turns: int | None = None, model_timeout: int | None = None, max_auto_continuations: int | None = None, + cache_min_measured_responses: int = CACHE_MIN_MEASURED_RESPONSES, ) -> None: self._identity = { "repository": repository, @@ -434,6 +523,13 @@ def __init__( self._last_good_current: dict | None = None self._last_finished: dict | None = None self._build_failed_logged = False + self._cache_min_measured_responses = cache_min_measured_responses + # The latest immutable live attempt view published by the executor. + # Swapped by reference under ``_attempt_view_lock`` (held only for + # the swap); the listener thread serves this copy and never reads + # the token ledger. + self._attempt_view: dict | None = None + self._attempt_view_lock = threading.Lock() def begin_attempt(self, seed: dict) -> None: """Seed the ``current_attempt`` view at claim; never raises. @@ -453,6 +549,13 @@ def begin_attempt(self, seed: dict) -> None: with self._attempt_lock: self._current_seed = normalized self._build_failed_logged = False + # A new attempt starts with no live figures until the executor + # publishes its first view after a limit check. + try: + with self._attempt_view_lock: + self._attempt_view = None + except Exception as error: # pragma: no cover - lock failure + logger.warning("telemetry begin_attempt failed: %s", error) def finish_attempt(self, summary: dict | None = None) -> None: """Record the last finished attempt at finalisation; never raises. @@ -487,6 +590,35 @@ def finish_attempt(self, summary: dict | None = None) -> None: self._build_failed_logged = False except Exception as error: # pragma: no cover - lock failure logger.warning("telemetry finish_attempt failed: %s", error) + try: + with self._attempt_view_lock: + self._attempt_view = None + except Exception as error: # pragma: no cover - lock failure + logger.warning("telemetry finish_attempt failed: %s", error) + + def publish_attempt_view(self, view: object) -> None: + """Store one immutable live attempt view from the executor; never raises. + + The view carries the executor's raw live readings (budget totals, + threshold state, measured categories, turns, cache and elapsed + figures); this method stamps ``observed_at`` and swaps the stored + reference. Only the swap holds ``_attempt_view_lock``: the copy and + the timestamp are prepared outside it, and the listener thread later + serves the stored copy without ever reading the token ledger. + After model execution ends no further view arrives, so the served + figures — including ``observed_at`` — stay frozen until + :meth:`finish_attempt`. + """ + + try: + staged = _plain_dict(view) + staged["observed_at"] = _format_timestamp(self._clock()) + with self._attempt_view_lock: + self._attempt_view = staged + except Exception as error: + logger.warning( + "telemetry publish_attempt_view failed: %s; keeping previous view", error + ) def snapshot(self, control: TelemetryControl) -> dict: """Build one schema-v1 snapshot; raise when the control section is missing. @@ -544,6 +676,13 @@ def _build_current_attempt(self) -> dict | None: configured = dict(self._configured) if seed is None: return None + with self._attempt_view_lock: + view = copy.deepcopy(self._attempt_view) if self._attempt_view is not None else None + if view is None: + return self._tracer_attempt(seed, configured) + return self._live_attempt(seed, configured, view) + + def _tracer_attempt(self, seed: dict, configured: dict) -> dict: continuation_index = seed.get("continuation_index") continuation_max = _seed_or_configured(seed, configured, "continuation_max") hard_ceiling = _seed_or_configured(seed, configured, "hard_ceiling") @@ -614,6 +753,171 @@ def _build_current_attempt(self) -> dict | None: attempt["issue_title"] = issue_title return attempt + def _live_attempt(self, seed: dict, configured: dict, view: dict) -> dict: + """Build the ``current_attempt`` section from a published live view. + + The seed still owns identity and configured limits; the view owns + every live figure and ``observed_at`` (stamped at publication, so a + frozen view stays frozen). Raises ``ValueError`` on a malformed + view so the caller keeps the last good section. + """ + + continuation_index = seed.get("continuation_index") + continuation_max = _seed_or_configured(seed, configured, "continuation_max") + hard_ceiling = _seed_or_configured(seed, configured, "hard_ceiling") + turns_max = _seed_or_configured(seed, configured, "turns_max") + timeout_seconds = _seed_or_configured(seed, configured, "timeout_seconds") + for name, value in ( + ("continuation_max", continuation_max), + ("hard_ceiling", hard_ceiling), + ("turns_max", turns_max), + ("timeout_seconds", timeout_seconds), + ): + if value is not None and ( + isinstance(value, bool) or not isinstance(value, int) or value < 0 + ): + raise ValueError(f"stored {name} is invalid") + + observed_at = view.get("observed_at") + if not isinstance(observed_at, str) or not observed_at: + raise ValueError("published attempt view must carry observed_at") + used = _view_tokens(view, "budget_tokens") + measured = _view_tokens(view, "measured_tokens") + estimated = _view_tokens(view, "estimated_tokens") + hard_ceiling_reached = view.get("hard_ceiling_reached") + if not isinstance(hard_ceiling_reached, bool): + raise ValueError("published attempt view hard_ceiling_reached must be a boolean") + + soft_value = _view_int(view, "soft_threshold_value") + rule = view.get("soft_threshold_rule") + if rule not in ("percentage", "reserve"): + raise ValueError("published attempt view soft_threshold_rule must be percentage or reserve") + percentage = view.get("soft_threshold_percentage") + if isinstance(percentage, bool) or not isinstance(percentage, (int, float)): + raise ValueError("published attempt view soft_threshold_percentage must be a number") + reserve_turns = _view_int(view, "soft_threshold_reserve_turns") + crossed = view.get("soft_threshold_crossed") + if not isinstance(crossed, bool): + raise ValueError("published attempt view soft_threshold_crossed must be a boolean") + crossed_at_tokens = view.get("soft_threshold_crossed_at_tokens") + if crossed_at_tokens is not None and ( + isinstance(crossed_at_tokens, bool) or not isinstance(crossed_at_tokens, int) + ): + raise ValueError("published attempt view soft_threshold_crossed_at_tokens must be an integer or None") + crossed_at = view.get("soft_threshold_crossed_at") + if crossed_at is not None and not isinstance(crossed_at, str): + raise ValueError("published attempt view soft_threshold_crossed_at must be a string or None") + main_basis = view.get("main_context_basis") + if main_basis not in ("measured", "estimated"): + main_basis = "estimated" + soft_basis = "configured" if rule == "percentage" else main_basis + + categories = view.get("tokens_by_category") + if not isinstance(categories, dict): + raise ValueError("published attempt view tokens_by_category must be a mapping") + tokens_by_category = {} + for category in TOKEN_CATEGORIES: + tokens_by_category[category] = { + "value": _view_category(categories, category), + "basis": "measured", + } + + turns = _view_int(view, "turns") + hit_rate = view.get("cache_hit_rate") + if hit_rate is not None and ( + isinstance(hit_rate, bool) or not isinstance(hit_rate, (int, float)) + ): + raise ValueError("published attempt view cache_hit_rate must be a number or None") + counted = _view_int(view, "cache_counted") + elapsed_seconds = _view_int(view, "elapsed_seconds") + if counted < 0 or elapsed_seconds < 0 or turns < 0: + raise ValueError("published attempt view counts must be non-negative") + + if counted >= self._cache_min_measured_responses and hit_rate is not None: + main_hit_rate: dict = { + "value": float(hit_rate), + "basis": "measured", + "counted": counted, + } + elif hit_rate is None and counted >= self._cache_min_measured_responses: + main_hit_rate = { + "value": None, + "basis": "unavailable", + "reason": CACHE_NO_MEASURED_INPUT_REASON, + "counted": counted, + } + else: + main_hit_rate = { + "value": None, + "basis": "unavailable", + "reason": CACHE_NOT_ENOUGH_RESPONSES_REASON, + "counted": counted, + } + + attempt: dict = { + "attempt_id": seed["attempt_id"], + "issue_number": seed["issue_number"], + "observed_at": observed_at, + "budget": { + "used": { + "value": used, + "basis": "measured" if estimated == 0 else "estimated", + }, + "measured": {"value": measured, "basis": "measured"}, + "estimated": {"value": estimated, "basis": "estimated"}, + "hard_ceiling": ( + _figure_configured(hard_ceiling) + if hard_ceiling is not None + else _figure_unavailable() + ), + "soft_threshold": { + "value": soft_value, + "basis": soft_basis, + "rule": rule, + "percentage": percentage, + "reserve_turns": reserve_turns, + "crossed": crossed, + "crossed_at_tokens": crossed_at_tokens, + "crossed_at": crossed_at, + }, + "hard_ceiling_reached": hard_ceiling_reached, + }, + "tokens_by_category": tokens_by_category, + "turns": { + "value": turns, + "basis": "measured", + "max": ( + _figure_configured(turns_max) + if turns_max is not None + else _figure_unavailable() + ), + }, + "continuation": { + "index": ( + _figure_configured(continuation_index) + if continuation_index is not None + else _figure_unavailable() + ), + "max": ( + _figure_configured(continuation_max) + if continuation_max is not None + else _figure_unavailable() + ), + }, + "cache": {"main_hit_rate": main_hit_rate}, + "provider_rate_limit": _figure_unavailable("not exposed through the gateway"), + "elapsed_seconds": {"value": elapsed_seconds, "basis": "measured"}, + "timeout_seconds": ( + _figure_configured(timeout_seconds) + if timeout_seconds is not None + else _figure_unavailable() + ), + } + issue_title = seed.get("issue_title") + if issue_title is not None: + attempt["issue_title"] = issue_title + return attempt + def apply_size_cap(snapshot: dict) -> dict: """Trim an assembled snapshot to the serialised size bound, in place. diff --git a/simple_coding_agent/token_ledger.py b/simple_coding_agent/token_ledger.py index 7b226f4..c07aca3 100644 --- a/simple_coding_agent/token_ledger.py +++ b/simple_coding_agent/token_ledger.py @@ -577,6 +577,26 @@ def response_cache_metrics( miss_tokens=miss, ) + @property + def main_measured_response_count(self) -> int: + """Number of measured main-thread responses (hit-rate denominator count).""" + + return len(self._measured_main_responses(None)) + + def main_context_basis(self) -> str: + """Basis of the main thread's current context for telemetry. + + ``"measured"`` when the main thread's latest response reported usage + (the context is anchored to that measurement); ``"estimated"`` + otherwise (no main thread yet, no response yet, or the latest + response is still an estimate). + """ + + thread = self._threads.get("main") + if thread is None or not thread.responses: + return "estimated" + return "measured" if thread.responses[-1].usage is not None else "estimated" + @property def main_measured_settled_count(self) -> int: """Settled measured responses on the main thread (warning trigger).""" diff --git a/tests/test_telemetry_live_figures.py b/tests/test_telemetry_live_figures.py new file mode 100644 index 0000000..b7b7804 --- /dev/null +++ b/tests/test_telemetry_live_figures.py @@ -0,0 +1,524 @@ +"""Live attempt figures published during model execution (telemetry). + +During model execution the executor publishes an immutable attempt view +after every limit check; the reporter swaps the reference under a lock +held only for the swap, and the listener thread serves the stored copy +without ever reading the token ledger. When model execution ends the +served figures — including ``observed_at`` — stay frozen until +finalisation. +""" + +from __future__ import annotations + +import asyncio +import json +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import MappingProxyType, SimpleNamespace + +from claude_agent_sdk import AssistantMessage + +from simple_coding_agent.config import RuntimeConfig +from simple_coding_agent.lifecycle import AgentLifecycle, ModelAttemptRunner +from simple_coding_agent.model_execution import ModelExecutionStatus, ModelExecutor +from simple_coding_agent.telemetry import TelemetryReporter +from simple_coding_agent.attempt_state import AttemptStateStore +from simple_coding_agent.control import ControlStore +from tests.fakes import FakePublisher, FakeTracker, InMemoryWorkspace + + +EXAMPLES = Path(__file__).parent.parent / "examples" / "telemetry" + + +def load_example(name: str) -> dict: + return json.loads((EXAMPLES / name).read_text()) + + +def scripted_clock(start: datetime = datetime(2026, 10, 14, 12, 0, tzinfo=UTC)): + """A clock that advances one second per call (deterministic publications).""" + + state = {"now": start} + + def tick() -> datetime: + state["now"] = state["now"] + timedelta(seconds=1) + return state["now"] + + return tick + + +def make_reporter(**overrides: object) -> TelemetryReporter: + values: dict[str, object] = { + "repository": "owner/repo", + "agent_version": "0.0-test", + "sdk_version": "0.2.163-test", + "cli_version": "2.1.286-test", + "process_started_at": "2026-01-01T00:00:00Z", + "hard_ceiling": 4_000_000, + "turns_max": 60, + "timeout_seconds": 3600, + "continuation_max": 0, + "clock": scripted_clock(), + } + values.update(overrides) + return TelemetryReporter(**values) # type: ignore[arg-type] + + +def claim_seed(**overrides: object) -> dict: + seed: dict[str, object] = { + "attempt_id": "2026-10-14T12:00:00-attempt-0977", + "issue_number": 977, + "issue_title": "Example: add a synthetic widget panel", + "continuation_index": 0, + } + seed.update(overrides) + return seed + + +def raw_view(**overrides: object) -> dict: + """One executor-shaped raw live view (pre-basis readings).""" + + view: dict[str, object] = { + "budget_tokens": 640000.0, + "measured_tokens": 512000.0, + "estimated_tokens": 128000.0, + "hard_ceiling_reached": False, + "soft_threshold_value": 3200000, + "soft_threshold_rule": "percentage", + "soft_threshold_percentage": 0.2, + "soft_threshold_reserve_turns": 6, + "soft_threshold_crossed": False, + "soft_threshold_crossed_at_tokens": None, + "soft_threshold_crossed_at": None, + "main_context_basis": "measured", + "tokens_by_category": { + "input_tokens": 120000, + "cache_read_input_tokens": 350000, + "cache_creation_input_tokens": 0, + "output_tokens": 42000, + }, + "turns": 18, + "cache_hit_rate": 0.86, + "cache_counted": 22, + "elapsed_seconds": 2700, + } + view.update(overrides) + return view + + +class StubControl: + def __init__(self) -> None: + self.section = {"control_observed_at": "2026-10-14T12:00:00Z"} + + def telemetry_control_snapshot(self) -> dict: + return dict(self.section) + + +def executor_config(tmp_path: Path, **overrides: object) -> RuntimeConfig: + values: dict[str, object] = { + "github_token": "github-secret", + "meta_api_key": "meta-secret", + "target_repo": "octo/example", + "data_dir": tmp_path, + "clone_dir": tmp_path / "repo", + "poll_interval": 60, + "log_level": "INFO", + "model_timeout": 60, + "publish_timeout": 120, + "max_retries": 3, + "max_consecutive_errors": 3, + "agent_trust_project_settings": False, + "review_blocking_severities": frozenset({"must-fix"}), + } + values.update(overrides) + return RuntimeConfig(**values) # type: ignore[arg-type] + + +def usage_message(index: int, input_tokens: int) -> AssistantMessage: + return AssistantMessage( + content=[], + model="muse-spark-1.3-contributor", + usage={ + "input_tokens": input_tokens, + "cache_read_input_tokens": 0, + "cache_creation_input_tokens": 0, + "output_tokens": 10, + }, + message_id=f"r{index}", + ) + + +class FakeClient: + def __init__(self, options: object, messages: list[object]) -> None: + self.options = options + self._messages = messages + self.queried_prompts: list[str] = [] + + async def __aenter__(self) -> FakeClient: + return self + + async def __aexit__(self, *unused: object) -> None: + return None + + async def connect(self, prompt: str) -> None: + pass + + async def interrupt(self) -> None: + pass + + async def query(self, prompt: str) -> None: + self.queried_prompts.append(prompt) + + async def receive_response(self): + if self.queried_prompts: + return + for message in self._messages: + yield message + + +def terminal_result() -> object: + return SimpleNamespace( + is_error=False, + stop_reason="end_turn", + model_usage={"muse-spark-1.3-contributor": {"input_tokens": 12}}, + ) + + +# --------------------------------------------------------------------------- +# shape conformance with the contract examples +# --------------------------------------------------------------------------- + + +def test_live_figures_match_the_attempt_in_progress_example_shape() -> None: + example = load_example("snapshot-attempt-in-progress.json")["current_attempt"] + reporter = make_reporter() + reporter.begin_attempt(claim_seed()) + reporter.publish_attempt_view(raw_view()) + + current = reporter.snapshot(StubControl())["current_attempt"] + + assert current is not None + assert set(current["budget"]) == set(example["budget"]) + assert current["budget"]["used"] == {"value": 640000, "basis": "estimated"} + assert current["budget"]["measured"] == {"value": 512000, "basis": "measured"} + assert current["budget"]["estimated"] == {"value": 128000, "basis": "estimated"} + assert current["budget"]["hard_ceiling"] == {"value": 4000000, "basis": "configured"} + assert current["budget"]["hard_ceiling_reached"] is False + assert set(current["budget"]["soft_threshold"]) == set(example["budget"]["soft_threshold"]) + assert current["budget"]["soft_threshold"]["basis"] == "configured" + assert current["budget"]["soft_threshold"]["rule"] == "percentage" + assert current["budget"]["soft_threshold"]["crossed"] is False + assert current["budget"]["soft_threshold"]["crossed_at_tokens"] is None + assert current["budget"]["soft_threshold"]["crossed_at"] is None + assert set(current["tokens_by_category"]) == set(example["tokens_by_category"]) + for figure in current["tokens_by_category"].values(): + assert figure["basis"] == "measured" + assert current["tokens_by_category"]["input_tokens"] == { + "value": 120000, + "basis": "measured", + } + assert set(current["turns"]) == set(example["turns"]) + assert current["turns"]["value"] == 18 + assert current["turns"]["basis"] == "measured" + assert set(current["cache"]["main_hit_rate"]) == set(example["cache"]["main_hit_rate"]) + assert current["cache"]["main_hit_rate"]["basis"] == "measured" + assert current["cache"]["main_hit_rate"]["counted"] == 22 + assert current["provider_rate_limit"] == example["provider_rate_limit"] + assert set(current["elapsed_seconds"]) == set(example["elapsed_seconds"]) + assert current["elapsed_seconds"] == {"value": 2700, "basis": "measured"} + assert isinstance(current["observed_at"], str) and current["observed_at"] + assert "usd" not in json.dumps(current).lower() + + +def test_soft_threshold_crossed_matches_the_handoff_example_shape() -> None: + example = load_example("snapshot-soft-threshold-handoff.json")["current_attempt"] + reporter = make_reporter() + reporter.begin_attempt(claim_seed()) + reporter.publish_attempt_view( + raw_view( + budget_tokens=3050000.0, + measured_tokens=2900000.0, + estimated_tokens=150000.0, + soft_threshold_value=2950000, + soft_threshold_rule="reserve", + soft_threshold_crossed=True, + soft_threshold_crossed_at_tokens=2960000, + soft_threshold_crossed_at="2026-09-30T11:55:00Z", + main_context_basis="estimated", + turns=47, + cache_hit_rate=0.91, + cache_counted=47, + elapsed_seconds=10500, + ) + ) + + current = reporter.snapshot(StubControl())["current_attempt"] + + assert current is not None + assert set(current["budget"]["soft_threshold"]) == set( + example["budget"]["soft_threshold"] + ) + soft = current["budget"]["soft_threshold"] + # Under the reserve rule the threshold inherits the main context's basis. + assert soft["basis"] == "estimated" + assert soft["rule"] == "reserve" + assert soft["crossed"] is True + assert soft["crossed_at_tokens"] == 2960000 + assert soft["crossed_at"] == "2026-09-30T11:55:00Z" + assert current["budget"]["used"] == {"value": 3050000, "basis": "estimated"} + assert current["turns"]["value"] == 47 + assert "usd" not in json.dumps(current).lower() + + +def test_used_basis_is_measured_when_nothing_is_estimated() -> None: + reporter = make_reporter() + reporter.begin_attempt(claim_seed()) + reporter.publish_attempt_view( + raw_view( + budget_tokens=512000.0, + measured_tokens=512000.0, + estimated_tokens=0.0, + ) + ) + + current = reporter.snapshot(StubControl())["current_attempt"] + + assert current is not None + assert current["budget"]["used"] == {"value": 512000, "basis": "measured"} + + +def test_hit_rate_is_unavailable_below_the_minimum_never_zero() -> None: + reporter = make_reporter() + reporter.begin_attempt(claim_seed()) + reporter.publish_attempt_view(raw_view(cache_hit_rate=0.9, cache_counted=5)) + + current = reporter.snapshot(StubControl())["current_attempt"] + + assert current is not None + hit_rate = current["cache"]["main_hit_rate"] + assert hit_rate["value"] is None + assert hit_rate["basis"] == "unavailable" + assert isinstance(hit_rate["reason"], str) and hit_rate["reason"] + assert hit_rate["counted"] == 5 + + +def test_observed_at_and_sequence_advance_with_every_publication() -> None: + reporter = make_reporter() + reporter.begin_attempt(claim_seed()) + reporter.publish_attempt_view(raw_view(turns=1)) + + first = reporter.snapshot(StubControl()) + + reporter.publish_attempt_view(raw_view(turns=2)) + second = reporter.snapshot(StubControl()) + + assert first["current_attempt"] is not None + assert second["current_attempt"] is not None + assert first["current_attempt"]["observed_at"] != second["current_attempt"]["observed_at"] + assert second["sequence"] == first["sequence"] + 1 + assert second["current_attempt"]["turns"]["value"] == 2 + + +# --------------------------------------------------------------------------- +# executor publication behaviour +# --------------------------------------------------------------------------- + + +def test_executor_publishes_an_immutable_view_after_every_limit_check( + tmp_path: Path, +) -> None: + published: list[object] = [] + executor = ModelExecutor( + executor_config(tmp_path), + attempt_publisher=published.append, + ) + + executor._observe_assistant_message(usage_message(0, 1000)) + executor._observe_assistant_message(usage_message(1, 2000)) + + assert len(published) == 2 + for view in published: + assert isinstance(view, MappingProxyType) + first = dict(published[0]) # type: ignore[arg-type] + second = dict(published[1]) # type: ignore[arg-type] + # The second publication carries the later readings; the first is untouched. + assert second["budget_tokens"] > first["budget_tokens"] + assert second["turns"] == 2 + assert first["turns"] == 1 + assert first["tokens_by_category"]["input_tokens"] == 1000 + assert second["tokens_by_category"]["input_tokens"] == 3000 + + +def test_failing_publish_does_not_affect_the_attempt(tmp_path: Path) -> None: + events: list[tuple] = [] + + def exploding_publisher(view: object) -> None: + raise RuntimeError("telemetry sink is down") + + executor = ModelExecutor( + executor_config(tmp_path, max_budget_tokens=1000), + attempt_publisher=exploding_publisher, + event_log=lambda event, detail="", level="INFO", issue_number=None: events.append( + (event, level) + ), + ) + + # A raising publisher must not escape the limit check, and the + # soft-threshold latch still engages on the same check. + executor._observe_assistant_message(usage_message(0, 1000)) + + assert executor._soft_threshold_crossed is True # noqa: SLF001 + assert ("telemetry_publish_failed", "WARNING") in events + + execution = asyncio.run( + ModelExecutor( + executor_config(tmp_path), + client_factory=lambda options: FakeClient( + options, [usage_message(0, 1000), terminal_result()] + ), + attempt_publisher=exploding_publisher, + ).execute(issue_body="Fix the widget.", working_directory=tmp_path / "repo") + ) + + assert execution.status is ModelExecutionStatus.SUCCEEDED + + +def test_views_freeze_after_execution_ends(tmp_path: Path) -> None: + reporter = make_reporter() + reporter.begin_attempt(claim_seed()) + publications: list[object] = [] + + def counting_publisher(view: object) -> None: + publications.append(view) + reporter.publish_attempt_view(view) + + executor = ModelExecutor( + executor_config(tmp_path), + client_factory=lambda options: FakeClient( + options, + [ + usage_message(0, 1000), + usage_message(1, 2000), + terminal_result(), + ], + ), + attempt_publisher=counting_publisher, + ) + + execution = asyncio.run( + executor.execute(issue_body="Fix the widget.", working_directory=tmp_path / "repo") + ) + + assert execution.status is ModelExecutionStatus.SUCCEEDED + frozen_count = len(publications) + assert frozen_count >= 2 + + first = reporter.snapshot(StubControl()) + second = reporter.snapshot(StubControl()) + + # Serving never publishes: the count is stable and the attempt section + # — including observed_at — is identical while the top-level sequence + # still advances per served snapshot. + assert len(publications) == frozen_count + assert first["current_attempt"] == second["current_attempt"] + assert second["sequence"] == first["sequence"] + 1 + assert first["current_attempt"]["turns"] == {"value": 2, "basis": "measured", "max": {"value": 60, "basis": "configured"}} + + +def test_listener_serves_a_detached_copy_without_ledger_access() -> None: + reporter = make_reporter() + assert not hasattr(reporter, "_ledger") + reporter.begin_attempt(claim_seed()) + + source = raw_view() + reporter.publish_attempt_view(source) + source["turns"] = 999 + assert isinstance(source["tokens_by_category"], dict) + source["tokens_by_category"]["input_tokens"] = 999999 + + current = reporter.snapshot(StubControl())["current_attempt"] + + assert current is not None + assert current["turns"]["value"] == 18 + assert current["tokens_by_category"]["input_tokens"]["value"] == 120000 + + +def test_begin_and_finish_clear_the_published_view() -> None: + reporter = make_reporter() + reporter.begin_attempt(claim_seed()) + reporter.publish_attempt_view(raw_view()) + + assert reporter.snapshot(StubControl())["current_attempt"]["turns"]["value"] == 18 + + reporter.begin_attempt(claim_seed()) + reseeded = reporter.snapshot(StubControl())["current_attempt"] + assert reseeded is not None + assert reseeded["turns"]["value"] is None + assert reseeded["budget"]["used"]["basis"] == "unavailable" + + reporter.publish_attempt_view(raw_view()) + assert reporter.snapshot(StubControl())["current_attempt"] is not None + reporter.finish_attempt() + assert reporter.snapshot(StubControl())["current_attempt"] is None + + +# --------------------------------------------------------------------------- +# ledger helpers and lifecycle wiring +# --------------------------------------------------------------------------- + + +def test_ledger_main_basis_and_measured_count(tmp_path: Path) -> None: + executor = ModelExecutor(executor_config(tmp_path)) + + assert executor._ledger.main_context_basis() == "estimated" # noqa: SLF001 + assert executor._ledger.main_measured_response_count == 0 # noqa: SLF001 + + executor._observe_assistant_message(usage_message(0, 1000)) + + assert executor._ledger.main_context_basis() == "measured" # noqa: SLF001 + assert executor._ledger.main_measured_response_count == 1 # noqa: SLF001 + + +def test_lifecycle_wires_the_reporter_into_the_executor(tmp_path: Path) -> None: + reporter = make_reporter() + executor = ModelExecutor(executor_config(tmp_path)) + lifecycle = AgentLifecycle( + tracker=FakeTracker(), + attempt_state=AttemptStateStore(tmp_path), + workspace=InMemoryWorkspace(tmp_path / "repo"), + profile_loader=lambda _: type("Profile", (), {"base_branch": "main"})(), + publisher=FakePublisher(), + attempt_runner=ModelAttemptRunner( + attempt_state=AttemptStateStore(tmp_path), + workspace=InMemoryWorkspace(tmp_path / "repo"), + verifier=None, # type: ignore[arg-type] + evaluator=None, # type: ignore[arg-type] + model_executor=executor, + ), + control_store=ControlStore(tmp_path), + repository="owner/repo", + telemetry=reporter, + sleeper=lambda seconds: None, + ) + + lifecycle._wire_attempt_telemetry_publisher() # noqa: SLF001 + + assert executor._attempt_publisher == reporter.publish_attempt_view # noqa: SLF001 + + +def test_lifecycle_wiring_leaves_plain_function_runners_alone(tmp_path: Path) -> None: + lifecycle = AgentLifecycle( + tracker=FakeTracker(), + attempt_state=AttemptStateStore(tmp_path), + workspace=InMemoryWorkspace(tmp_path / "repo"), + profile_loader=lambda _: type("Profile", (), {"base_branch": "main"})(), + publisher=FakePublisher(), + attempt_runner=lambda claim, profile, prepared: (_ for _ in ()).throw( + AssertionError("must not run") + ), + control_store=ControlStore(tmp_path), + repository="owner/repo", + telemetry=make_reporter(), + sleeper=lambda seconds: None, + ) + + lifecycle._wire_attempt_telemetry_publisher() # noqa: SLF001 From 694afe270eebeac8b5939b3865127619956ede76 Mon Sep 17 00:00:00 2001 From: Claude Code Date: Wed, 14 Oct 2026 14:39:43 +0000 Subject: [PATCH 3/3] Address code-review findings for #244 (deep-freeze published view, wire cache threshold) - Freeze the nested tokens_by_category mapping in the published attempt view so the executor's publication is deeply immutable, and extend the per-limit-check test to cover it. - Pass PROMPT_CACHE_MIN_MEASURED_RESPONSES explicitly when constructing the telemetry reporter so the served hit-rate threshold shares its single source of truth with the executor's cache warning. - The context section stays out of scope: issue #245 owns context-window, compaction and subagent figures. Co-Authored-By: Claude Code --- simple_coding_agent/__main__.py | 9 ++++++++- simple_coding_agent/model_execution.py | 8 +++++--- tests/test_telemetry_live_figures.py | 8 ++++++++ 3 files changed, 21 insertions(+), 4 deletions(-) diff --git a/simple_coding_agent/__main__.py b/simple_coding_agent/__main__.py index ab95abf..580e627 100644 --- a/simple_coding_agent/__main__.py +++ b/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, diff --git a/simple_coding_agent/model_execution.py b/simple_coding_agent/model_execution.py index db2c8be..df82a37 100644 --- a/simple_coding_agent/model_execution.py +++ b/simple_coding_agent/model_execution.py @@ -1995,8 +1995,10 @@ def _live_attempt_view(self) -> dict[str, Any]: 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 caller wraps - the result as immutable before publishing. + 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() @@ -2020,7 +2022,7 @@ def _live_attempt_view(self) -> dict[str, Any]: None if crossed_at is None else _format_timestamp(crossed_at) ), "main_context_basis": self._ledger.main_context_basis(), - "tokens_by_category": dict(self._ledger.measured_by_category), + "tokens_by_category": MappingProxyType(dict(self._ledger.measured_by_category)), "turns": ( self._ledger.response_count("main") if self._ledger.has_thread("main") diff --git a/tests/test_telemetry_live_figures.py b/tests/test_telemetry_live_figures.py index b7b7804..1ebbebd 100644 --- a/tests/test_telemetry_live_figures.py +++ b/tests/test_telemetry_live_figures.py @@ -338,6 +338,14 @@ def test_executor_publishes_an_immutable_view_after_every_limit_check( assert len(published) == 2 for view in published: assert isinstance(view, MappingProxyType) + # The nested category mapping is frozen too, not just the top level. + assert isinstance(view["tokens_by_category"], MappingProxyType) # type: ignore[index] + try: + view["tokens_by_category"]["input_tokens"] = 0 # type: ignore[index] + except TypeError: + pass + else: + raise AssertionError("nested tokens_by_category must be immutable") first = dict(published[0]) # type: ignore[arg-type] second = dict(published[1]) # type: ignore[arg-type] # The second publication carries the later readings; the first is untouched.