feat(cost-diagnostics): per-call context ledger — prefix split, window trims, compaction events

backend/src/agents/main_agent/base_agent.py

@@ -14,6 +14,7 @@
 from agents.main_agent.session import SessionFactory
 from agents.main_agent.session.hooks import (
     AgentStatusHook,
+    ContextLedgerHook,
     ToolCensusHook,
     DisplayTextHook,
     SteeringHook,
@@ -350,6 +351,13 @@ def _create_hooks(self) -> List:
         self.tool_census_hook = ToolCensusHook()
         hooks.append(self.tool_census_hook)
 
+        # Per-model-call context ledger: the conversation window's cumulative
+        # trim count and the compaction decisions taken since the previous
+        # call. Same shape and lifecycle as the census — read per call at
+        # turn end, persisted on the cost row, off with the same kill switch.
+        self.context_ledger_hook = ContextLedgerHook()
+        hooks.append(self.context_ledger_hook)
+
         # Per-model-call prompt-cache prefix fingerprints (toolConfig /
         # system prompt / history hashes). Best-effort; the stream
         # coordinator persists them on each call's metadata row so avoidable

backend/src/agents/main_agent/session/hooks/__init__.py

@@ -2,6 +2,7 @@
 
 from agents.main_agent.session.hooks.agent_status import AgentStatusHook
 from agents.main_agent.session.hooks.context_attribution import ContextAttributionHook
+from agents.main_agent.session.hooks.context_ledger import ContextLedgerHook
 from agents.main_agent.session.hooks.display_text import DisplayTextHook
 from agents.main_agent.session.hooks.oauth_consent import OAuthConsentHook
 from agents.main_agent.session.hooks.prefix_fingerprint import PrefixFingerprintHook
@@ -13,6 +14,7 @@
 __all__ = [
     "AgentStatusHook",
     "ContextAttributionHook",
+    "ContextLedgerHook",
     "DisplayTextHook",
     "OAuthConsentHook",
     "PrefixFingerprintHook",

backend/src/agents/main_agent/session/hooks/context_attribution.py

@@ -31,7 +31,7 @@
 """
 
 import logging
-from typing import Any, Optional
+from typing import Any, Dict, Optional
 
 from strands.hooks import BeforeModelCallEvent, HookProvider, HookRegistry
 
@@ -50,6 +50,26 @@ def get_context_breakdown(agent: Any) -> Optional[dict]:
     return getattr(agent, _BREAKDOWN_ATTR, None)
 
 
+def get_prefix_token_split(agent: Any) -> Optional[Dict[str, int]]:
+    """The stable ``{"system": n, "tools": n}`` split for this agent, or ``None``.
+
+    Persisted on each call's cost row (as ``prefixTokens``) so the static
+    prefix a session carries — and which part of it is tool schemas — is a
+    stored fact rather than a scan-and-guess. Same numbers the SSE breakdown
+    reports; this just reads the cached split without re-counting.
+    """
+    split = getattr(agent, _SPLIT_ATTR, None)
+    if not isinstance(split, dict):
+        return None
+    try:
+        return {
+            "system": int(split.get("systemTokens") or 0),
+            "tools": int(split.get("toolTokens") or 0),
+        }
+    except (TypeError, ValueError):
+        return None
+
+
 class ContextAttributionHook(HookProvider):
     """Compute the system / tools / messages token breakdown each turn."""
 

backend/src/agents/main_agent/session/hooks/context_ledger.py

@@ -0,0 +1,117 @@
+"""Hook that records, per model call, what happened to the context *before* it.
+
+Two content-free facts the cost anatomy could not otherwise answer from stored
+data, both found missing during the 2026-09-15 prod cost audit:
+
+- ``windowRemovedMessages`` — the conversation manager's cumulative
+  ``removed_message_count`` at the moment of the call. A rise between two
+  consecutive rows means the message list was trimmed in between (Strands'
+  sliding window sliding, or a compaction slice), which re-writes the cached
+  prefix. Distinguishing "pure window slide" sessions from "compaction spiral"
+  sessions took an hour of fingerprint reading; with this field it is one
+  query.
+- ``compactionEvents`` — the compaction decisions the session manager made
+  since the previous call (restore-time slice applied, checkpoint advanced,
+  and whatever a future scheduling policy records: forced cut, floor
+  unreachable). Each carries the summary's token size at that moment, which
+  is what proves a summary cap shrank summaries without another scan.
+
+Per-turn, per-model-call, held on the agent wrapper exactly like the tool
+census: the stream coordinator reads ``ledger_for_call(idx)`` at turn end and
+persists it on that call's ``C#`` cost row. Gated by the same
+``COST_DIAGNOSTICS_ENABLED`` kill switch; while off the callbacks return
+immediately and nothing is written, so a row without the fields reads
+"not tracked", never "0".
+"""
+
+from __future__ import annotations
+
+import copy
+import logging
+from typing import Any, Dict, List, Optional
+
+from strands.hooks import BeforeInvocationEvent, BeforeModelCallEvent, HookProvider, HookRegistry
+
+from apis.shared.feature_flags import cost_diagnostics_enabled
+
+logger = logging.getLogger(__name__)
+
+#: A call never carries more events than this; a runaway recorder must not
+#: grow a cost row without bound.
+_MAX_EVENTS_PER_CALL = 8
+
+
+def _removed_message_count(agent: Any) -> Optional[int]:
+    manager = getattr(agent, "conversation_manager", None)
+    value = getattr(manager, "removed_message_count", None)
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _drain_compaction_events(agent: Any) -> List[Dict[str, Any]]:
+    """Take the session manager's pending compaction events, if it keeps any.
+
+    Strands stores the manager as ``agent._session_manager``; ours exposes
+    ``drain_compaction_events``. Anything else (tests, other managers) yields
+    an empty list.
+    """
+    manager = getattr(agent, "_session_manager", None)
+    drain = getattr(manager, "drain_compaction_events", None)
+    if not callable(drain):
+        return []
+    try:
+        events = drain()
+    except Exception as e:  # noqa: BLE001 - a ledger must never break a turn
+        logger.debug("Context ledger could not drain compaction events: %s", e)
+        return []
+    return list(events or [])[:_MAX_EVENTS_PER_CALL]
+
+
+class ContextLedgerHook(HookProvider):
+    """Per-turn, per-model-call context ledger: ``{cycle: {...}}``."""
+
+    def __init__(self) -> None:
+        self._cycle = 0
+        self._ledger: Dict[int, Dict[str, Any]] = {}
+
+    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
+        registry.add_callback(BeforeInvocationEvent, self._on_turn_start)
+        registry.add_callback(BeforeModelCallEvent, self._on_before_model_call)
+
+    def ledger_for_call(self, call_index: int) -> Optional[Dict[str, Any]]:
+        """The ledger entry for model call ``call_index`` (0-based), or
+        ``None`` when nothing was recorded — or the diagnostics are off.
+
+        Returns a copy so the caller can hand it to the persistence layer
+        without aliasing per-turn state.
+        """
+        if not cost_diagnostics_enabled():
+            return None
+        entry = self._ledger.get(call_index + 1)
+        return copy.deepcopy(entry) if entry else None
+
+    def _on_turn_start(self, event: BeforeInvocationEvent) -> None:
+        self._cycle = 0
+        self._ledger = {}
+
+    def _on_before_model_call(self, event: BeforeModelCallEvent) -> None:
+        self._cycle += 1
+        if not cost_diagnostics_enabled():
+            return
+        try:
+            agent = event.agent
+            entry: Dict[str, Any] = {}
+            removed = _removed_message_count(agent)
+            if removed is not None:
+                entry["windowRemovedMessages"] = removed
+            events = _drain_compaction_events(agent)
+            if events:
+                entry["compactionEvents"] = events
+            if entry:
+                self._ledger[self._cycle] = entry
+        except Exception as e:  # noqa: BLE001 - a ledger must never break a turn
+            logger.debug("Context ledger skipped a call: %s", e)

backend/src/agents/main_agent/session/turn_based_session_manager.py

@@ -49,6 +49,21 @@
 
 logger = logging.getLogger(__name__)
 
+#: Compaction decisions the per-call ledger will record. Reserved kinds exist
+#: so a scheduling policy can report them without a schema change.
+COMPACTION_EVENT_KINDS = frozenset({"applied", "checkpoint", "forced", "floor_unreachable"})
+_MAX_PENDING_COMPACTION_EVENTS = 8
+
+
+def _approx_tokens(text: Any) -> int:
+    """~4 chars/token; the same estimate the admin profile uses for summaries."""
+    if not text:
+        return 0
+    try:
+        return len(text) // 4
+    except TypeError:
+        return 0
+
 
 class TurnBasedSessionManager(AgentCoreMemorySessionManager):
     """
@@ -108,6 +123,9 @@ def __init__(
         # Cached data for checkpoint calculation
         self._valid_cutoff_indices: List[int] = []
         self._all_messages_for_summary: List[Dict] = []
+        # Compaction decisions taken since the last model call, drained by
+        # ``ContextLedgerHook`` onto that call's cost row. Bounded; content-free.
+        self._pending_compaction_events: List[Dict[str, Any]] = []
         self._total_message_count_at_init: int = 0
         # Absolute index (into the stored history) of ``agent.messages[0]``.
         # The restore slice sets it to the applied checkpoint; the persisted
@@ -368,6 +386,15 @@ def _apply_compaction(self, agent: "Agent") -> None:
 
         agent.messages = messages_to_process
 
+        if checkpoint > 0 and offset > 0:
+            self.record_compaction_event(
+                "applied",
+                checkpoint=checkpoint,
+                summaryTokens=_approx_tokens(self.compaction_state.summary),
+                retainedMessages=len(agent.messages),
+                truncatedToolResults=truncation_count,
+            )
+
         logger.info(
             f"Compaction initialized: stage={stage}, "
             f"original={self._total_message_count_at_init}, "
@@ -603,6 +630,48 @@ def _adopt_persisted_compaction_state(self) -> None:
         )
         self.compaction_state = persisted
 
+    # ------------------------------------------------------------------
+    # Compaction event ledger (content-free; persisted per model call)
+    # ------------------------------------------------------------------
+
+    def record_compaction_event(self, kind: str, **fields: Any) -> None:
+        """Queue a compaction decision for the next model call's cost row.
+
+        ``kind`` is one of the ``COMPACTION_EVENT_KINDS`` — ``applied`` (the
+        restore-time slice ran), ``checkpoint`` (a new checkpoint was cut
+        post-turn), ``forced`` and ``floor_unreachable`` (reserved for the
+        scheduling policy: a cut taken at the hard ceiling because the previous
+        one did not take, and a cut that could not reach its floor because the
+        protected tail alone exceeds it). ``fields`` are numbers only —
+        ``summaryTokens`` in particular is what proves a summary cap works
+        without another table scan. Anything else is dropped here so the
+        ledger can never carry content.
+
+        No-op when ``COST_DIAGNOSTICS_ENABLED`` is off, so a row without the
+        field reads "not tracked", never "0". Bounded so a runaway caller
+        cannot grow a cost row.
+        """
+        from apis.shared.feature_flags import cost_diagnostics_enabled
+
+        if not cost_diagnostics_enabled():
+            return
+        if kind not in COMPACTION_EVENT_KINDS:
+            logger.debug("Ignoring unknown compaction event kind %r", kind)
+            return
+        if len(self._pending_compaction_events) >= _MAX_PENDING_COMPACTION_EVENTS:
+            return
+        event: Dict[str, Any] = {"kind": kind}
+        for key, value in fields.items():
+            if isinstance(value, bool) or not isinstance(value, (int, float)):
+                continue
+            event[key] = int(value)
+        self._pending_compaction_events.append(event)
+
+    def drain_compaction_events(self) -> List[Dict[str, Any]]:
+        """Return and clear the queued events (called by ``ContextLedgerHook``)."""
+        events, self._pending_compaction_events = self._pending_compaction_events, []
+        return events
+
     def _save_compaction_state(self, state: CompactionState, record_event: bool = False) -> None:
         """Save compaction state to DynamoDB session metadata.
 
@@ -1011,6 +1080,17 @@ async def update_after_turn(
         # This save is the compaction event itself — count it.
         self._save_compaction_state(state, record_event=True)
         self._emit_compaction_metrics(state, policy, forced, retained_estimate, bounded, deferred)
+        # Routed through the defensive seam rather than calling the recorder
+        # directly: same no-op-without-a-ledger contract as every other cut
+        # decision. Queued when the cut is *decided* — when ``deferred`` the
+        # bytes do not move until ``apply_pending_compaction`` runs.
+        self._record_ledger_event(
+            "checkpoint",
+            checkpoint=new_checkpoint,
+            summaryTokens=_approx_tokens(summary),
+            summarizedTurns=summarized_turns,
+            inputTokens=input_tokens,
+        )
 
         logger.info(
             f"Compaction checkpoint set: {new_checkpoint}, "

backend/src/agents/main_agent/streaming/stream_coordinator.py

@@ -15,6 +15,8 @@
     get_prefix_fingerprint,
     reset_prefix_fingerprints,
 )
+from agents.main_agent.session.hooks.context_attribution import get_prefix_token_split
+from apis.shared.feature_flags import cost_diagnostics_enabled
 from apis.shared.errors import (
     ConversationalErrorEvent,
     ErrorCode,
@@ -1217,6 +1219,9 @@ async def stream_response(
                 # cost row carries the tools that call requested. None when the
                 # wrapper has no hook (tests, older agents) or the census is off.
                 tool_census_hook = getattr(main_agent_wrapper, "tool_census_hook", None)
+                # Same discipline for the context ledger (window trims +
+                # compaction decisions per call).
+                context_ledger_hook = getattr(main_agent_wrapper, "context_ledger_hook", None)
 
                 # Build list of metadata storage tasks for parallel execution
                 metadata_tasks = []
@@ -1273,6 +1278,10 @@ async def stream_response(
                                 tool_census_hook.tally_for_call(idx)
                                 if tool_census_hook is not None else None
                             ),
+                            context_ledger=(
+                                context_ledger_hook.ledger_for_call(idx)
+                                if context_ledger_hook is not None else None
+                            ),
                         )
                     )
 
@@ -2893,6 +2902,7 @@ async def _store_message_metadata(
         call_index: Optional[int] = None,
         turn_agent_id: Optional[str] = None,
         tool_calls: Optional[Dict[str, Dict[str, int]]] = None,
+        context_ledger: Optional[Dict[str, Any]] = None,
     ) -> None:
         """
         Store message-level metadata (token usage, latency, model info, citations)
@@ -3087,6 +3097,25 @@ async def _store_message_metadata(
                 if tool_calls:
                     metadata_kwargs["toolCalls"] = tool_calls
 
+                # Context ledger for this call: the conversation window's
+                # cumulative trim count (a rise between consecutive rows is a
+                # trim, i.e. a prefix re-write) and the compaction decisions
+                # taken since the previous call, each with the summary's
+                # token size. Plus the agent's stable prefix split (system /
+                # tools tokens) so "how big is the static prefix, and how much
+                # of it is tool schemas" is a stored fact. All numbers.
+                if context_ledger:
+                    removed = context_ledger.get("windowRemovedMessages")
+                    if removed is not None:
+                        metadata_kwargs["windowRemovedMessages"] = removed
+                    events = context_ledger.get("compactionEvents")
+                    if events:
+                        metadata_kwargs["compactionEvents"] = events
+                if strands_agent is not None and cost_diagnostics_enabled():
+                    prefix_tokens = get_prefix_token_split(strands_agent)
+                    if prefix_tokens:
+                        metadata_kwargs["prefixTokens"] = prefix_tokens
+
                 message_metadata = MessageMetadata(**metadata_kwargs)
 
                 # Store metadata