diff --git a/.agent/handoff/245.md b/.agent/handoff/245.md
new file mode 100644
index 0000000..4256aa6
--- /dev/null
+++ b/.agent/handoff/245.md
@@ -0,0 +1,21 @@
+# Handoff note: issue #245
+
+- issue: 245
+- started_at: 2026-10-14T14:44:28.136967Z
+- reason: token_soft_threshold
+- last_work_commit: 6bc8700486fd951d07958358bdc4cb001d02ba19
+
+## Summary
+
+Implemented context telemetry for issue #245 (uncommitted at handoff): `simple_coding_agent/model_execution.py` tracks the context window/threshold via the SDK `get_context_usage()` call at session start and after each compaction only (basis `sdk_assumed`, or `configured` when `CLAUDE_CODE_MAX_CONTEXT_TOKENS` is set; failures yield `unavailable` with a reason and never affect the attempt), compaction count/last (`at`, `trigger`, `pre_tokens`, `post_tokens`) from `compact_boundary` events with post filled from the follow-up usage total, and a subagent registry keyed by Task tool_use_id with running flags cleared on Task tool results (view capped at 20, running first, newest first). `simple_coding_agent/telemetry.py` builds `context.{main,main_peak,window,auto_compact_threshold,compactions,subagents}` with figure invariants, and the tracer carries an unavailable context section. New `tests/test_telemetry_context.py` (untracked) covers window basis with/without the override, failed usage, compaction count/last plus re-reading, and subagent ordering plus the 20-entry cap.
+
+## Remaining work
+
+1. Invoke the `code-review` skill on the diff and address findings.
+2. Commit the implementation (`model_execution.py`, `telemetry.py`, `tests/test_telemetry_context.py`) on `agent/issue-245`.
+3. Run the completion gate in order: (1) ruff check, (2) py3.11 pytest, (3) py3.13 pytest.
+
+## Evidence
+
+- Branch state at handoff: `M simple_coding_agent/model_execution.py`, `M simple_coding_agent/telemetry.py`, `?? tests/test_telemetry_context.py` (left uncommitted; sealing preserves them).
+- Prior turn results (pre-threshold): new context tests 11 passed; full py3.11 suite 1125 passed, 10 skipped; `ruff check` clean.
diff --git a/simple_coding_agent/model_execution.py b/simple_coding_agent/model_execution.py
index df82a37..b9e2819 100644
--- a/simple_coding_agent/model_execution.py
+++ b/simple_coding_agent/model_execution.py
@@ -8,6 +8,7 @@
 from datetime import UTC, datetime
 from enum import StrEnum
 import json
+import os
 from pathlib import Path
 import re
 import shlex
@@ -32,6 +33,11 @@
 )
 
 from simple_coding_agent.config import RuntimeConfig
+from simple_coding_agent.telemetry import CONTEXT_WINDOW_BASES
+from simple_coding_agent.telemetry import (
+    MAX_CONTEXT_SUBAGENTS as _MAX_CONTEXT_SUBAGENTS,
+)
+from simple_coding_agent.telemetry import _as_int as _telemetry_as_int
 from simple_coding_agent.telemetry import format_timestamp as _format_timestamp
 from simple_coding_agent.token_ledger import (
     USAGE_CATEGORIES,
@@ -61,6 +67,34 @@
 _OPERATOR_HANDOFF_DELIVERY_WINDOW_SECONDS = 60
 _OPERATOR_HANDOFF_MODEL_DEADLINE_SECONDS = 240
 
+#: Operator override that marks the telemetry context window as configured.
+#: When set, the CLI uses it as the context window, so the SDK reading's
+#: basis is ``configured`` instead of ``sdk_assumed``.
+CONTEXT_WINDOW_OVERRIDE_ENV_VAR = "CLAUDE_CODE_MAX_CONTEXT_TOKENS"
+
+#: Upper bound of telemetry subagent entries in one published view, shared
+#: with the reporter so both caps stay a single source of truth.
+MAX_TELEMETRY_SUBAGENTS = _MAX_CONTEXT_SUBAGENTS
+
+
+def _context_window_configured() -> bool:
+    """Whether the operator overrode the context window for this process."""
+
+    return bool(os.environ.get(CONTEXT_WINDOW_OVERRIDE_ENV_VAR, "").strip())
+
+
+def _context_window_basis() -> str:
+    """Basis for a successful context-usage window reading."""
+
+    return "configured" if _context_window_configured() else "sdk_assumed"
+
+
+def _as_non_negative_int(value: object) -> int | None:
+    """An int figure from an SDK payload, or ``None`` when untrustworthy."""
+
+    coerced = _telemetry_as_int(value)
+    return coerced if coerced is not None and coerced >= 0 else None
+
 def _handoff_target(issue_number: int | None) -> str:
     """Name the issue and the exact note path every handoff instruction must carry."""
 
@@ -577,6 +611,25 @@ def _reset_ledger_state(self) -> None:
         self._degraded_logged = False
         self._cache_warning_evaluated = False
         self._auto_response_seq = 0
+        self._reset_context_state()
+
+    def _reset_context_state(self) -> None:
+        """Start fresh per-attempt telemetry context figures.
+
+        The window/threshold stay ``unavailable`` until the session-start
+        context-usage reading lands; compactions and subagents start empty.
+        """
+
+        self._context_window: int | None = None
+        self._context_window_basis: str = "unavailable"
+        self._context_window_reason: str | None = "not yet read"
+        self._auto_compact_threshold: int | None = None
+        self._auto_compact_threshold_basis: str = "unavailable"
+        self._auto_compact_threshold_reason: str | None = "not yet read"
+        self._compaction_count = 0
+        self._compaction_last: dict[str, Any] | None = None
+        self._subagent_info: dict[str, dict[str, Any]] = {}
+        self._subagent_order: list[str] = []
 
     def set_operator_handoff_provider(
         self, provider: OperatorHandoffProvider | None
@@ -649,7 +702,10 @@ def _log_context_compacted(self, message: SystemMessage) -> None:
         The budget is left alone; the ledger only learns of the compaction so
         the next response is excluded from cache-miss accounting. SDK 0.2.163
         has no typed message for it and ``compact_metadata`` is unverified,
-        so ``trigger`` and ``pre_tokens`` are read best-effort.
+        so ``trigger`` and ``pre_tokens`` are read best-effort. The telemetry
+        compaction count and last-compaction record are updated here; the
+        post-compaction tokens are filled from the next context-usage
+        reading (or from ``post_tokens`` metadata when the CLI provides it).
         """
 
         data = getattr(message, "data", None)
@@ -663,6 +719,89 @@ def _log_context_compacted(self, message: SystemMessage) -> None:
             f"thread={thread}; trigger={metadata.get('trigger')}; "
             f"pre_tokens={metadata.get('pre_tokens')}",
         )
+        trigger = metadata.get("trigger")
+        trigger = trigger if isinstance(trigger, str) and trigger else "unknown"
+        pre_tokens = _as_non_negative_int(metadata.get("pre_tokens"))
+        post_tokens = _as_non_negative_int(metadata.get("post_tokens"))
+        self._compaction_count += 1
+        self._compaction_last = {
+            "at": _format_timestamp(self._clock()),
+            "trigger": trigger,
+            "pre_tokens": pre_tokens,
+            "post_tokens": post_tokens,
+        }
+
+    async def _refresh_context_usage(self, client: Any, *, reason: str) -> None:
+        """Refresh the telemetry window from the SDK's context-usage call.
+
+        Runs only at session start and after each compaction, on the
+        attempt's event loop — never per turn. A failure marks the window
+        and threshold ``unavailable`` with a reason and never affects the
+        attempt. A successful reading also fills a pending compaction
+        post-tokens value from the live total.
+        """
+
+        read = getattr(client, "get_context_usage", None)
+        if not callable(read):
+            self._mark_context_usage_unavailable("context usage not supported")
+            return
+        try:
+            usage = await read()
+        except Exception as error:
+            self._mark_context_usage_unavailable(f"context usage failed: {type(error).__name__}")
+            self._log(
+                "telemetry_context_usage_failed",
+                f"reason={reason}; error={type(error).__name__}",
+                level="WARNING",
+            )
+            return
+        if not isinstance(usage, Mapping):
+            self._mark_context_usage_unavailable("context usage failed: bad response")
+            return
+        basis = _context_window_basis()
+        raw_max = _as_non_negative_int(usage.get("rawMaxTokens"))
+        max_tokens = _as_non_negative_int(usage.get("maxTokens"))
+        window = raw_max if raw_max is not None else max_tokens
+        threshold = _as_non_negative_int(usage.get("autoCompactThreshold"))
+        if window is None:
+            self._context_window = None
+            self._context_window_basis = "unavailable"
+            self._context_window_reason = "context usage missing window"
+        else:
+            self._context_window = window
+            self._context_window_basis = basis
+            self._context_window_reason = None
+        if threshold is None:
+            self._auto_compact_threshold = None
+            self._auto_compact_threshold_basis = "unavailable"
+            self._auto_compact_threshold_reason = "context usage missing threshold"
+        else:
+            self._auto_compact_threshold = threshold
+            self._auto_compact_threshold_basis = basis
+            self._auto_compact_threshold_reason = None
+        total = _as_non_negative_int(usage.get("totalTokens"))
+        if (
+            total is not None
+            and self._compaction_last is not None
+            and self._compaction_last.get("post_tokens") is None
+        ):
+            self._compaction_last["post_tokens"] = total
+
+    def _mark_context_usage_unavailable(self, reason: str) -> None:
+        """Mark the window and threshold unavailable without touching the attempt.
+
+        A previous successful reading is kept: a failed re-read after a
+        compaction must not discard the window the session start observed.
+        """
+
+        if self._context_window_basis not in CONTEXT_WINDOW_BASES:
+            self._context_window = None
+            self._context_window_basis = "unavailable"
+            self._context_window_reason = reason
+        if self._auto_compact_threshold_basis not in CONTEXT_WINDOW_BASES:
+            self._auto_compact_threshold = None
+            self._auto_compact_threshold_basis = "unavailable"
+            self._auto_compact_threshold_reason = reason
 
     def _record_cli_session_id(self, message: object) -> None:
         """Latch the CLI session id from the init message or any sessioned message.
@@ -927,6 +1066,17 @@ def _start_launch_thread(self, hook_input: Any, tool_use_id: str | None) -> None
         self._ensure_ledger_thread(
             tool_use_id, subagent_type=subagent_type, prompt_chars=len(prompt)
         )
+        info = self._subagent_info.get(tool_use_id)
+        if info is None:
+            self._subagent_info[tool_use_id] = {
+                "type": subagent_type if isinstance(subagent_type, str) and subagent_type else "unknown",
+                "running": True,
+            }
+            self._subagent_order.append(tool_use_id)
+        else:
+            info["running"] = True
+            if isinstance(subagent_type, str) and subagent_type:
+                info["type"] = subagent_type
 
     def _track_subagent_seen(self, hook_input: Any) -> None:
         """Remember a subagent id seen before the soft-threshold crossing.
@@ -1049,6 +1199,7 @@ async def execute(
         try:
             async with client:
                 await client.connect(f"/implement\n\n{issue_body}")
+                await self._refresh_context_usage(client, reason="session_start")
                 try:
                     async with asyncio.timeout(self._config.model_timeout):
                         terminal, observed_models = await self._receive_terminal(client)
@@ -1550,6 +1701,8 @@ async def _receive_terminal(
                 if isinstance(message, SystemMessage) and message.subtype == "compact_boundary":
                     self._log_context_compacted(message)
                     self._arm_handoff_reinject_after_compaction()
+                    await self._refresh_context_usage(client, reason="compaction")
+                    self._publish_attempt_view()
                 had_pending_background = bool(background_pending)
                 if self._observe_background_task(
                     message, background_pending, background_finished
@@ -1806,6 +1959,9 @@ def _observe_user_message(self, message: UserMessage) -> None:
             # with its own reading.
             if tool_use_id != thread_id and self._ledger.has_thread(tool_use_id):
                 self._settle_thread_and_log(tool_use_id)
+            info = self._subagent_info.get(tool_use_id)
+            if info is not None:
+                info["running"] = False
         self._ledger.observe_input(thread_id, _content_chars(getattr(message, "content", "")))
         self._after_ledger_update(thread_id)
 
@@ -1995,9 +2151,9 @@ 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 result is
-        deeply immutable: the nested ``tokens_by_category`` mapping is
-        frozen too, so the caller only needs the top-level
+        context-window/compaction/subagent figures, and elapsed seconds.
+        No USD figure is included. The result is deeply immutable: nested
+        mappings are frozen too, so the caller only needs the top-level
         ``MappingProxyType`` wrap before publishing.
         """
 
@@ -2022,6 +2178,9 @@ 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(),
+            "main_context_tokens": self._latest_main_context_tokens(),
+            "main_peak_context_tokens": float(self._ledger.peak_main_context_tokens),
+            "main_peak_context_basis": self._ledger.peak_main_context_basis(),
             "tokens_by_category": MappingProxyType(dict(self._ledger.measured_by_category)),
             "turns": (
                 self._ledger.response_count("main")
@@ -2031,8 +2190,53 @@ def _live_attempt_view(self) -> dict[str, Any]:
             "cache_hit_rate": self._ledger.main_hit_rate(),
             "cache_counted": self._ledger.main_measured_response_count,
             "elapsed_seconds": self._elapsed_seconds(),
+            "context_window": self._context_window,
+            "context_window_basis": self._context_window_basis,
+            "context_window_reason": self._context_window_reason,
+            "auto_compact_threshold": self._auto_compact_threshold,
+            "auto_compact_threshold_basis": self._auto_compact_threshold_basis,
+            "auto_compact_threshold_reason": self._auto_compact_threshold_reason,
+            "compaction_count": self._compaction_count,
+            "compaction_last": (
+                None
+                if self._compaction_last is None
+                else MappingProxyType(dict(self._compaction_last))
+            ),
+            "subagents": tuple(
+                MappingProxyType(dict(entry)) for entry in self._telemetry_subagents()
+            ),
         }
 
+    def _telemetry_subagents(self) -> list[dict[str, Any]]:
+        """Live subagent readings for one telemetry publication, running first.
+
+        Each entry carries the ledger thread's current context with its
+        running flag; the reporter tags the context ``estimated`` and caps
+        the list at 20. Newest launches come first within each running
+        group so a fresh running subagent is never trimmed before an old
+        finished one.
+        """
+
+        try:
+            contexts = self._ledger.thread_contexts()
+        except Exception:
+            contexts = {}
+        running: list[dict[str, Any]] = []
+        finished: list[dict[str, Any]] = []
+        for thread_id in reversed(self._subagent_order):
+            info = self._subagent_info.get(thread_id)
+            if info is None:
+                continue
+            entry = {
+                "id": thread_id,
+                "type": info.get("type", "unknown"),
+                "context": float(contexts.get(thread_id, 0.0)),
+                "running": bool(info.get("running", False)),
+            }
+            (running if entry["running"] else finished).append(entry)
+        ordered = running + finished
+        return ordered[:MAX_TELEMETRY_SUBAGENTS]
+
     def _publish_attempt_view(self) -> None:
         """Publish one immutable attempt view; a failure never affects the attempt.
 
diff --git a/simple_coding_agent/telemetry.py b/simple_coding_agent/telemetry.py
index 749fcd2..deaa2c2 100644
--- a/simple_coding_agent/telemetry.py
+++ b/simple_coding_agent/telemetry.py
@@ -86,6 +86,16 @@
     "output_tokens",
 )
 
+#: Bases a successful context-usage window reading may carry. Anything else
+#: on the view is treated as ``unavailable`` so a malformed publisher can
+#: never smuggle an unknown basis into the snapshot.
+CONTEXT_WINDOW_BASES = ("sdk_assumed", "configured")
+
+#: Maximum telemetry subagent entries in one snapshot section (running
+#: first). The executor caps its views at the same size; the reporter
+#: re-enforces it so an oversized view never reaches the socket.
+MAX_CONTEXT_SUBAGENTS = 20
+
 
 logger = logging.getLogger(__name__)
 
@@ -188,6 +198,28 @@ def _view_category(categories: dict, category: str) -> int:
     return value
 
 
+def _view_main_context(view: dict) -> int:
+    """Main-thread context from a published view (ledger reading, rounded).
+
+    Missing keys (views published before context landed) read as zero so
+    older publishers still build; a malformed value raises ``ValueError``.
+    """
+
+    if "main_context_tokens" not in view:
+        return 0
+    return _view_tokens(view, "main_context_tokens")
+
+
+def _view_main_peak_context(view: dict) -> int:
+    """Peak main-thread context from a published view (rounded)."""
+
+    if "main_peak_context_tokens" not in view:
+        # Views published before the peak landed carry no peak; fall back to
+        # the current main reading so the section stays consistent.
+        return _view_main_context(view)
+    return _view_tokens(view, "main_peak_context_tokens")
+
+
 def _as_int(value: object) -> int | None:
     if isinstance(value, bool):
         return None
@@ -364,6 +396,136 @@ def _measured_figure(candidate: object) -> dict:
     return {"value": value, "basis": "measured"}
 
 
+def _context_window_figure(view: dict, prefix: str) -> dict:
+    """One window/threshold figure from a published view.
+
+    A present non-negative integer with an ``sdk_assumed``/``configured``
+    basis passes through; anything else (missing value, unknown basis, or
+    an explicit ``unavailable`` basis) is ``unavailable`` with the view's
+    reason (or a default). Raises ``ValueError`` on a malformed basis or
+    reason so the caller keeps the last good section.
+    """
+
+    value = view.get(prefix)
+    basis = view.get(f"{prefix}_basis", "unavailable")
+    reason = view.get(f"{prefix}_reason")
+    if basis in CONTEXT_WINDOW_BASES:
+        value = _as_int(value)
+        if value is None or value < 0:
+            raise ValueError(f"published attempt view {prefix} must be a non-negative integer")
+        return {"value": value, "basis": basis}
+    if basis != "unavailable":
+        raise ValueError(f"published attempt view {prefix}_basis must be sdk_assumed, configured or unavailable")
+    if reason is not None and not isinstance(reason, str):
+        raise ValueError(f"published attempt view {prefix}_reason must be a string or None")
+    return {
+        "value": None,
+        "basis": "unavailable",
+        "reason": reason if isinstance(reason, str) and reason else "not yet published",
+    }
+
+
+def _compaction_token_figure(value: object) -> dict:
+    """One compaction pre/post figure (measured when present)."""
+
+    value = _as_int(value)
+    if value is not None and value >= 0:
+        return {"value": value, "basis": "measured"}
+    return {"value": None, "basis": "unavailable", "reason": "not reported"}
+
+
+def _build_compactions(view: dict) -> dict:
+    """The ``context.compactions`` section from a published view.
+
+    Missing keys (views published before context landed) read as zero
+    compactions; a malformed count or entry raises ``ValueError``.
+    """
+
+    count = view.get("compaction_count", 0)
+    if isinstance(count, bool) or not isinstance(count, int) or count < 0:
+        raise ValueError("published attempt view compaction_count must be a non-negative integer")
+    raw_last = view.get("compaction_last")
+    if raw_last is None:
+        return {"count": count, "last": None}
+    if not isinstance(raw_last, dict):
+        raise ValueError("published attempt view compaction_last must be a mapping or None")
+    at = raw_last.get("at")
+    trigger = raw_last.get("trigger")
+    if not isinstance(at, str) or not at:
+        raise ValueError("published attempt view compaction_last.at must be a non-empty string")
+    if not isinstance(trigger, str) or not trigger:
+        raise ValueError("published attempt view compaction_last.trigger must be a non-empty string")
+    return {
+        "count": count,
+        "last": {
+            "at": at,
+            "trigger": trigger,
+            "pre_tokens": _compaction_token_figure(raw_last.get("pre_tokens")),
+            "post_tokens": _compaction_token_figure(raw_last.get("post_tokens")),
+        },
+    }
+
+
+def _view_peak_basis(view: dict, main_basis: str) -> str:
+    """Basis of the peak main-thread context from a published view.
+
+    Views published before the peak basis landed fall back to the current
+    main basis; a present-but-unknown basis raises ``ValueError`` so the
+    caller keeps the last good section.
+    """
+
+    peak_basis = view.get("main_peak_context_basis", main_basis)
+    if peak_basis not in ("measured", "estimated"):
+        raise ValueError("published attempt view main_peak_context_basis must be measured or estimated")
+    return peak_basis
+
+
+def _build_subagents(view: dict) -> list[dict]:
+    """The ``context.subagents`` list from a published view, running first.
+
+    Each entry is ``{id, type, context (estimated), running}``; the list is
+    capped at 20 with running entries first (stable within each group).
+    Missing views read as empty; malformed entries raise ``ValueError``.
+    """
+
+    raw = view.get("subagents", [])
+    if raw is None:
+        return []
+    if not isinstance(raw, (list, tuple)):
+        raise ValueError("published attempt view subagents must be a list")
+    entries: list[dict] = []
+    for item in raw:
+        if not isinstance(item, dict):
+            raise ValueError("published attempt view subagents entries must be mappings")
+        agent_id = item.get("id")
+        agent_type = item.get("type", "unknown")
+        running = item.get("running")
+        if not isinstance(agent_id, str) or not agent_id:
+            raise ValueError("published attempt view subagents id must be a non-empty string")
+        if not isinstance(agent_type, str) or not agent_type:
+            agent_type = "unknown"
+        if not isinstance(running, bool):
+            raise ValueError("published attempt view subagents running must be a boolean")
+        context_value = item.get("context", 0)
+        if isinstance(context_value, bool):
+            raise ValueError("published attempt view subagents context must be a number")
+        if isinstance(context_value, float):
+            context_value = int(round(context_value))
+        if not isinstance(context_value, int) or context_value < 0:
+            raise ValueError("published attempt view subagents context must be non-negative")
+        entries.append(
+            {
+                "id": agent_id,
+                "type": agent_type,
+                "context": {"value": context_value, "basis": "estimated"},
+                "running": running,
+            }
+        )
+    running_first = [entry for entry in entries if entry["running"]]
+    finished = [entry for entry in entries if not entry["running"]]
+    return (running_first + finished)[:MAX_CONTEXT_SUBAGENTS]
+
+
 def _normalize_pull_request(candidate: object) -> dict | None:
     """A ``{number, url}`` pull request, or ``None`` when there is none."""
 
@@ -746,6 +908,14 @@ def _tracer_attempt(self, seed: dict, configured: dict) -> dict:
                 else _figure_unavailable()
             ),
             "elapsed_seconds": _figure_unavailable(),
+            "context": {
+                "main": _figure_unavailable(),
+                "main_peak": _figure_unavailable(),
+                "window": _figure_unavailable(),
+                "auto_compact_threshold": _figure_unavailable(),
+                "compactions": {"count": 0, "last": None},
+                "subagents": [],
+            },
             "provider_rate_limit": _figure_unavailable("not exposed through the gateway"),
         }
         issue_title = seed.get("issue_title")
@@ -905,6 +1075,22 @@ def _live_attempt(self, seed: dict, configured: dict, view: dict) -> dict:
                 ),
             },
             "cache": {"main_hit_rate": main_hit_rate},
+            "context": {
+                "main": {
+                    "value": _view_main_context(view),
+                    "basis": main_basis,
+                },
+                "main_peak": {
+                    "value": _view_main_peak_context(view),
+                    "basis": _view_peak_basis(view, main_basis),
+                },
+                "window": _context_window_figure(view, "context_window"),
+                "auto_compact_threshold": _context_window_figure(
+                    view, "auto_compact_threshold"
+                ),
+                "compactions": _build_compactions(view),
+                "subagents": _build_subagents(view),
+            },
             "provider_rate_limit": _figure_unavailable("not exposed through the gateway"),
             "elapsed_seconds": {"value": elapsed_seconds, "basis": "measured"},
             "timeout_seconds": (
@@ -923,7 +1109,8 @@ def apply_size_cap(snapshot: dict) -> dict:
     """Trim an assembled snapshot to the serialised size bound, in place.
 
     Over the cap the snapshot drops ``recent_events`` oldest first, then
-    ``current_attempt.context.subagents`` oldest first, and reports
+    ``current_attempt.context.subagents`` from the tail (the list is
+    running-first, so the tail is the oldest finished entry), and reports
     ``truncated: true``. Under the cap the snapshot is returned untouched.
     """
 
@@ -947,7 +1134,7 @@ def apply_size_cap(snapshot: dict) -> dict:
         subagents = context.get("subagents")
         if not isinstance(subagents, list) or not subagents:
             break
-        subagents.pop(0)
+        subagents.pop()
     return snapshot
 
 
diff --git a/simple_coding_agent/token_ledger.py b/simple_coding_agent/token_ledger.py
index c07aca3..29d98fd 100644
--- a/simple_coding_agent/token_ledger.py
+++ b/simple_coding_agent/token_ledger.py
@@ -264,6 +264,7 @@ def __init__(
         self._first_settled_without_usage: DegradedNotice | None = None
         self._settled_readings: dict[tuple[str, str], SettledReading] = {}
         self._peak_main_context_tokens = 0.0
+        self._peak_main_context_basis = "estimated"
         self._compaction_pending: set[str] = set()
 
     # -- events ----------------------------------------------------------
@@ -353,7 +354,7 @@ def observe_usage(
             + response.usage["cache_creation_input_tokens"]
             + response.usage["output_tokens"]
         )
-        self._track_peak(thread)
+        self._track_peak(thread, basis="measured")
 
     def observe_input(self, thread_id: str, chars: int) -> None:
         """Grow a thread's context with a tool result or user message."""
@@ -624,6 +625,16 @@ def peak_main_context_tokens(self) -> float:
 
         return self._peak_main_context_tokens
 
+    def peak_main_context_basis(self) -> str:
+        """Basis of the peak main-thread context for telemetry.
+
+        ``"measured"`` when the response holding the peak reported usage;
+        ``"estimated"`` otherwise. Tracked at peak-set time because the
+        current reading's basis may differ from the peak's.
+        """
+
+        return self._peak_main_context_basis
+
     @property
     def first_settled_without_usage(self) -> DegradedNotice | None:
         """First response that settled without reported usage, if any."""
@@ -757,6 +768,22 @@ def _main_miss_at(responses: list[_Response], index: int) -> int | None:
             return None
         return max(0, _total_input_of(previous.usage) - response.usage["cache_read_input_tokens"])
 
-    def _track_peak(self, thread: _Thread) -> None:
+    def _track_peak(self, thread: _Thread, *, basis: str | None = None) -> None:
+        """Record a new main-thread peak with the basis of the reading behind it.
+
+        Usage merges pass ``"measured"`` explicitly: the merged response may
+        not be the latest one (late usage for an older response), so the
+        latest response's basis can differ from the peak's. Other call sites
+        derive the basis from the latest response.
+        """
+
         if thread.kind == "main" and thread.context > self._peak_main_context_tokens:
             self._peak_main_context_tokens = thread.context
+            if basis is None:
+                latest = thread.responses[-1] if thread.responses else None
+                basis = (
+                    "measured"
+                    if latest is not None and latest.usage is not None
+                    else "estimated"
+                )
+            self._peak_main_context_basis = basis
diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py
index 7fd3015..da33ec0 100644
--- a/tests/test_telemetry.py
+++ b/tests/test_telemetry.py
@@ -796,8 +796,10 @@ def test_size_cap_drops_subagents_after_events_are_empty() -> None:
     assert len(json.dumps(trimmed).encode("utf-8")) <= SNAPSHOT_SIZE_LIMIT_BYTES
     remaining = trimmed["current_attempt"]["context"]["subagents"]
     assert remaining
-    assert remaining[0]["id"] != "agent-000"
-    assert remaining[-1]["id"] == "agent-059"
+    # The list is running-first, so trimming drops the tail (oldest
+    # finished) and preserves the head (newest running).
+    assert remaining[0]["id"] == "agent-000"
+    assert remaining[-1]["id"] != "agent-059"
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/test_telemetry_context.py b/tests/test_telemetry_context.py
new file mode 100644
index 0000000..61c6d6f
--- /dev/null
+++ b/tests/test_telemetry_context.py
@@ -0,0 +1,512 @@
+"""Context figures in telemetry (issue #245).
+
+``context.main``/``main_peak`` come from the ledger, ``window`` and
+``auto_compact_threshold`` from the SDK context-usage call (session start
+and after each compaction only), ``compactions`` from compaction boundary
+events, and ``subagents`` from ledger threads with running flags.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from types import MappingProxyType
+
+from claude_agent_sdk import AssistantMessage, SystemMessage
+
+from simple_coding_agent.config import RuntimeConfig
+from simple_coding_agent.model_execution import ModelExecutor
+from simple_coding_agent.telemetry import SNAPSHOT_SIZE_LIMIT_BYTES, apply_size_cap
+from simple_coding_agent.token_ledger import TokenLedger
+from tests.test_telemetry_live_figures import (
+    claim_seed,
+    make_reporter,
+    raw_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}",
+    )
+
+
+def terminal_result() -> object:
+    from types import SimpleNamespace
+
+    return SimpleNamespace(
+        is_error=False,
+        stop_reason="end_turn",
+        model_usage={"muse-spark-1.3-contributor": {"input_tokens": 12}},
+    )
+
+
+class FakeClient:
+    """SDK fake with a scripted context-usage response."""
+
+    instances: list["FakeClient"] = []
+
+    def __init__(
+        self,
+        options: object,
+        messages: list[object],
+        usage: dict | Exception | None = None,
+    ) -> None:
+        self.options = options
+        self._messages = messages
+        self._usage = (
+            {
+                "rawMaxTokens": 200000,
+                "maxTokens": 200000,
+                "autoCompactThreshold": 190000,
+                "totalTokens": 50000,
+            }
+            if usage is None
+            else usage
+        )
+        self.usage_calls = 0
+        FakeClient.instances.append(self)
+
+    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:
+        pass
+
+    async def get_context_usage(self) -> dict:
+        self.usage_calls += 1
+        if isinstance(self._usage, Exception):
+            raise self._usage
+        self._usage["totalTokens"] = 50000 + self.usage_calls * 1000
+        return dict(self._usage)
+
+    async def receive_response(self):
+        for message in self._messages:
+            yield message
+
+
+def run_execute(tmp_path: Path, messages: list[object], usage=None) -> tuple[ModelExecutor, FakeClient]:
+    FakeClient.instances.clear()
+    published: list[object] = []
+    executor = ModelExecutor(
+        executor_config(tmp_path),
+        client_factory=lambda options: FakeClient(options, messages, usage),
+        attempt_publisher=published.append,
+    )
+    asyncio.run(executor.execute(issue_body="Fix it.", working_directory=tmp_path / "repo"))
+    assert FakeClient.instances
+    return executor, FakeClient.instances[0]
+
+
+def context_view(**overrides: object) -> dict:
+    view = raw_view()
+    view.update(
+        {
+            "main_context_tokens": 148000.0,
+            "main_peak_context_tokens": 151200.0,
+            "main_peak_context_basis": "measured",
+            "context_window": 200000,
+            "context_window_basis": "sdk_assumed",
+            "context_window_reason": None,
+            "auto_compact_threshold": 190000,
+            "auto_compact_threshold_basis": "sdk_assumed",
+            "auto_compact_threshold_reason": None,
+            "compaction_count": 0,
+            "compaction_last": None,
+            "subagents": (),
+        }
+    )
+    view.update(overrides)
+    return view
+
+
+# ---------------------------------------------------------------------------
+# reporter shape
+# ---------------------------------------------------------------------------
+
+
+def test_reporter_context_mains_use_ledger_basis() -> None:
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    reporter.publish_attempt_view(context_view(main_context_basis="measured"))
+
+    current = reporter.snapshot(StubControl())["current_attempt"]
+
+    assert current is not None
+    assert current["context"]["main"] == {"value": 148000, "basis": "measured"}
+    assert current["context"]["main_peak"] == {"value": 151200, "basis": "measured"}
+    assert current["context"]["window"] == {"value": 200000, "basis": "sdk_assumed"}
+    assert current["context"]["auto_compact_threshold"] == {
+        "value": 190000,
+        "basis": "sdk_assumed",
+    }
+    assert current["context"]["compactions"] == {"count": 0, "last": None}
+    assert current["context"]["subagents"] == []
+
+
+def test_reporter_window_configured_basis() -> None:
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    reporter.publish_attempt_view(
+        context_view(
+            context_window_basis="configured",
+            auto_compact_threshold_basis="configured",
+        )
+    )
+
+    current = reporter.snapshot(StubControl())["current_attempt"]
+
+    assert current is not None
+    assert current["context"]["window"] == {"value": 200000, "basis": "configured"}
+    assert current["context"]["auto_compact_threshold"] == {
+        "value": 190000,
+        "basis": "configured",
+    }
+
+
+def test_reporter_window_unavailable_carries_reason() -> None:
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    reporter.publish_attempt_view(
+        context_view(
+            context_window=None,
+            context_window_basis="unavailable",
+            context_window_reason="context usage failed: RuntimeError",
+            auto_compact_threshold=None,
+            auto_compact_threshold_basis="unavailable",
+            auto_compact_threshold_reason="context usage failed: RuntimeError",
+        )
+    )
+
+    current = reporter.snapshot(StubControl())["current_attempt"]
+
+    assert current is not None
+    assert current["context"]["window"] == {
+        "value": None,
+        "basis": "unavailable",
+        "reason": "context usage failed: RuntimeError",
+    }
+
+
+def test_reporter_subagents_running_first_and_capped() -> None:
+    subagents = tuple(
+        {
+            "id": f"sub-{index:02d}",
+            "type": "Explore",
+            "context": 1000.0 + index,
+            "running": index % 2 == 0,
+        }
+        for index in range(25)
+    )
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    reporter.publish_attempt_view(context_view(subagents=subagents))
+
+    current = reporter.snapshot(StubControl())["current_attempt"]
+
+    assert current is not None
+    listed = current["context"]["subagents"]
+    assert len(listed) == 20
+    # Running entries come first.
+    seen_finished = False
+    for entry in listed:
+        if not entry["running"]:
+            seen_finished = True
+        else:
+            assert not seen_finished
+        assert entry["context"]["basis"] == "estimated"
+        assert isinstance(entry["context"]["value"], int)
+
+
+def test_reporter_tracer_context_is_unavailable() -> None:
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+
+    current = reporter.snapshot(StubControl())["current_attempt"]
+
+    assert current is not None
+    assert current["context"]["main"]["basis"] == "unavailable"
+    assert current["context"]["window"]["basis"] == "unavailable"
+    assert current["context"]["compactions"] == {"count": 0, "last": None}
+    assert current["context"]["subagents"] == []
+
+
+# ---------------------------------------------------------------------------
+# executor behaviour
+# ---------------------------------------------------------------------------
+
+
+def test_window_basis_sdk_assumed_without_override(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.delenv("CLAUDE_CODE_MAX_CONTEXT_TOKENS", raising=False)
+    executor, client = run_execute(tmp_path, [usage_message(0, 1000), terminal_result()])
+
+    assert client.usage_calls == 1
+    view = dict(executor._live_attempt_view())  # noqa: SLF001
+    assert view["context_window"] == 200000
+    assert view["context_window_basis"] == "sdk_assumed"
+    assert view["auto_compact_threshold"] == 190000
+    assert view["auto_compact_threshold_basis"] == "sdk_assumed"
+
+
+def test_window_basis_configured_with_override(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv("CLAUDE_CODE_MAX_CONTEXT_TOKENS", "200000")
+    executor, client = run_execute(tmp_path, [usage_message(0, 1000), terminal_result()])
+
+    assert client.usage_calls == 1
+    view = dict(executor._live_attempt_view())  # noqa: SLF001
+    assert view["context_window_basis"] == "configured"
+    assert view["auto_compact_threshold_basis"] == "configured"
+
+
+def test_failed_context_usage_yields_unavailable(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.delenv("CLAUDE_CODE_MAX_CONTEXT_TOKENS", raising=False)
+    executor, client = run_execute(
+        tmp_path,
+        [usage_message(0, 1000), terminal_result()],
+        usage=RuntimeError("boom"),
+    )
+
+    # The attempt still succeeds; only the figures are unavailable.
+    assert client.usage_calls == 1
+    view = dict(executor._live_attempt_view())  # noqa: SLF001
+    assert view["context_window"] is None
+    assert view["context_window_basis"] == "unavailable"
+    assert isinstance(view["context_window_reason"], str)
+
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    reporter.publish_attempt_view(MappingProxyType(dict(view)))
+    current = reporter.snapshot(StubControl())["current_attempt"]
+    assert current is not None
+    assert current["context"]["window"]["basis"] == "unavailable"
+    assert current["context"]["window"]["value"] is None
+
+
+def test_compaction_updates_count_last_and_triggers_reading(
+    tmp_path: Path, monkeypatch
+) -> None:
+    monkeypatch.delenv("CLAUDE_CODE_MAX_CONTEXT_TOKENS", raising=False)
+    executor, client = run_execute(
+        tmp_path,
+        [
+            usage_message(0, 1000),
+            SystemMessage(
+                subtype="compact_boundary",
+                data={"compact_metadata": {"trigger": "auto", "pre_tokens": 150000}},
+            ),
+            usage_message(1, 2000),
+            terminal_result(),
+        ],
+    )
+
+    # Session start + one compaction.
+    assert client.usage_calls == 2
+    view = dict(executor._live_attempt_view())  # noqa: SLF001
+    assert view["compaction_count"] == 1
+    last = dict(view["compaction_last"])
+    assert last["trigger"] == "auto"
+    assert last["pre_tokens"] == 150000
+    # Post tokens filled from the post-compaction context-usage total.
+    assert isinstance(last["post_tokens"], int) and last["post_tokens"] > 0
+    assert isinstance(last["at"], str) and last["at"]
+
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    reporter.publish_attempt_view(MappingProxyType(dict(view)))
+    current = reporter.snapshot(StubControl())["current_attempt"]
+    assert current is not None
+    assert current["context"]["compactions"]["count"] == 1
+    snapshot_last = current["context"]["compactions"]["last"]
+    assert snapshot_last["trigger"] == "auto"
+    assert snapshot_last["pre_tokens"] == {"value": 150000, "basis": "measured"}
+    assert snapshot_last["post_tokens"]["basis"] == "measured"
+
+
+def test_subagent_running_flag_and_ordering(tmp_path: Path) -> None:
+    executor = ModelExecutor(executor_config(tmp_path))
+    executor._reset_ledger_state()  # noqa: SLF001
+    executor._ensure_ledger_thread("main", prompt_chars=10)  # noqa: SLF001
+
+    # Three launches: sub-1 finishes, sub-2 and sub-3 stay running.
+    for tool_id, sub_type in [("tool-1", "Explore"), ("tool-2", "Plan"), ("tool-3", "Explore")]:
+        executor._ensure_ledger_thread(tool_id, subagent_type=sub_type, prompt_chars=10)  # noqa: SLF001
+        executor._subagent_info[tool_id] = {"type": sub_type, "running": True}  # noqa: SLF001
+        executor._subagent_order.append(tool_id)  # noqa: SLF001
+    executor._subagent_info["tool-1"]["running"] = False  # noqa: SLF001
+
+    ordered = executor._telemetry_subagents()  # noqa: SLF001
+    assert [entry["id"] for entry in ordered] == ["tool-3", "tool-2", "tool-1"]
+    assert [entry["running"] for entry in ordered] == [True, True, False]
+
+    view = dict(executor._live_attempt_view())  # noqa: SLF001
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    reporter.publish_attempt_view(MappingProxyType(dict(view)))
+    current = reporter.snapshot(StubControl())["current_attempt"]
+    assert current is not None
+    listed = current["context"]["subagents"]
+    assert [entry["id"] for entry in listed] == ["tool-3", "tool-2", "tool-1"]
+
+
+def test_subagent_view_capped_at_20(tmp_path: Path) -> None:
+    executor = ModelExecutor(executor_config(tmp_path))
+    executor._reset_ledger_state()  # noqa: SLF001
+    executor._ensure_ledger_thread("main", prompt_chars=10)  # noqa: SLF001
+    for index in range(25):
+        tool_id = f"tool-{index:02d}"
+        executor._ensure_ledger_thread(tool_id, subagent_type="Explore", prompt_chars=10)  # noqa: SLF001
+        executor._subagent_info[tool_id] = {"type": "Explore", "running": True}  # noqa: SLF001
+        executor._subagent_order.append(tool_id)  # noqa: SLF001
+
+    ordered = executor._telemetry_subagents()  # noqa: SLF001
+    assert len(ordered) == 20
+
+
+def test_reporter_peak_uses_own_basis() -> None:
+    """A measured peak keeps its basis when the current reading is estimated."""
+
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    reporter.publish_attempt_view(
+        context_view(main_context_basis="estimated", main_peak_context_basis="measured")
+    )
+
+    current = reporter.snapshot(StubControl())["current_attempt"]
+
+    assert current is not None
+    assert current["context"]["main"] == {"value": 148000, "basis": "estimated"}
+    assert current["context"]["main_peak"] == {"value": 151200, "basis": "measured"}
+
+
+def test_reporter_peak_basis_falls_back_to_main_basis() -> None:
+    """Views published before the peak basis landed reuse the main basis."""
+
+    reporter = make_reporter()
+    reporter.begin_attempt(claim_seed())
+    view = context_view(main_context_basis="estimated")
+    del view["main_peak_context_basis"]
+    reporter.publish_attempt_view(view)
+
+    current = reporter.snapshot(StubControl())["current_attempt"]
+
+    assert current is not None
+    assert current["context"]["main_peak"] == {"value": 151200, "basis": "estimated"}
+
+
+def test_ledger_peak_basis_tracks_peak_response() -> None:
+    """Late usage for an older response sets a measured peak under an estimated current."""
+
+    ledger = TokenLedger()
+    ledger.start_thread("main", kind="main", prompt_chars=0)
+    assert ledger.peak_main_context_basis() == "estimated"
+    ledger.observe_response("main", "r1", visible_chars=300)
+    ledger.observe_response("main", "r2", visible_chars=300)
+    assert ledger.main_context_basis() == "estimated"
+
+    ledger.observe_usage(
+        "main",
+        "r1",
+        {"input_tokens": 200_000, "cache_read_input_tokens": 0, "output_tokens": 100},
+    )
+
+    assert ledger.peak_main_context_tokens == 200_100
+    assert ledger.peak_main_context_basis() == "measured"
+    assert ledger.main_context_basis() == "estimated"
+
+
+def test_failed_reread_keeps_last_good_window(tmp_path: Path, monkeypatch) -> None:
+    """A failed post-compaction re-read keeps the session-start window."""
+
+    monkeypatch.delenv("CLAUDE_CODE_MAX_CONTEXT_TOKENS", raising=False)
+    executor, _client = run_execute(tmp_path, [usage_message(0, 1000), terminal_result()])
+
+    class FailingUsageClient:
+        async def get_context_usage(self) -> dict:
+            raise RuntimeError("boom")
+
+    asyncio.run(executor._refresh_context_usage(FailingUsageClient(), reason="compaction"))  # noqa: SLF001
+
+    view = dict(executor._live_attempt_view())  # noqa: SLF001
+    assert view["context_window"] == 200000
+    assert view["context_window_basis"] == "sdk_assumed"
+    assert view["context_window_reason"] is None
+    assert view["auto_compact_threshold"] == 190000
+    assert view["auto_compact_threshold_basis"] == "sdk_assumed"
+
+
+def test_size_cap_drops_tail_subagents_and_keeps_running() -> None:
+    """Over the cap, trimming drops the oldest finished entries, never running ones."""
+
+    subagents = [
+        {"id": "running-new", "type": "Explore", "context": {"value": 10, "basis": "estimated"}, "running": True},
+        {"id": "running-old", "type": "Plan", "context": {"value": 20, "basis": "estimated"}, "running": True},
+    ] + [
+        {
+            "id": f"finished-{index:03d}",
+            "type": "Explore",
+            "context": {"value": 30, "basis": "estimated"},
+            "running": False,
+            "pad": "z" * 5000,
+        }
+        for index in range(60)
+    ]
+    snapshot: dict = {
+        "recent_events": [],
+        "current_attempt": {"context": {"subagents": subagents}},
+        "truncated": False,
+    }
+
+    trimmed = apply_size_cap(snapshot)
+
+    assert trimmed["truncated"] is True
+    assert len(json.dumps(trimmed).encode("utf-8")) <= SNAPSHOT_SIZE_LIMIT_BYTES
+    remaining = trimmed["current_attempt"]["context"]["subagents"]
+    assert [entry["id"] for entry in remaining[:2]] == ["running-new", "running-old"]
+    assert all(entry["running"] or entry["id"].startswith("finished-") for entry in remaining)
+    assert remaining[-1]["id"] != "finished-059"