diff --git a/backend/src/agents/main_agent/base_agent.py b/backend/src/agents/main_agent/base_agent.py index 281d45c2..87ba396b 100644 --- a/backend/src/agents/main_agent/base_agent.py +++ b/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 diff --git a/backend/src/agents/main_agent/session/hooks/__init__.py b/backend/src/agents/main_agent/session/hooks/__init__.py index 1080adc5..d5754524 100644 --- a/backend/src/agents/main_agent/session/hooks/__init__.py +++ b/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", diff --git a/backend/src/agents/main_agent/session/hooks/context_attribution.py b/backend/src/agents/main_agent/session/hooks/context_attribution.py index 8edd5575..c997ec4b 100644 --- a/backend/src/agents/main_agent/session/hooks/context_attribution.py +++ b/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.""" diff --git a/backend/src/agents/main_agent/session/hooks/context_ledger.py b/backend/src/agents/main_agent/session/hooks/context_ledger.py new file mode 100644 index 00000000..263d0bf9 --- /dev/null +++ b/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) diff --git a/backend/src/agents/main_agent/session/turn_based_session_manager.py b/backend/src/agents/main_agent/session/turn_based_session_manager.py index 81572d82..7520bf3e 100644 --- a/backend/src/agents/main_agent/session/turn_based_session_manager.py +++ b/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}, " diff --git a/backend/src/agents/main_agent/streaming/stream_coordinator.py b/backend/src/agents/main_agent/streaming/stream_coordinator.py index 838a376d..e807c1c9 100644 --- a/backend/src/agents/main_agent/streaming/stream_coordinator.py +++ b/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 diff --git a/backend/src/apis/app_api/admin/costs/models.py b/backend/src/apis/app_api/admin/costs/models.py index be72e451..1754da8a 100644 --- a/backend/src/apis/app_api/admin/costs/models.py +++ b/backend/src/apis/app_api/admin/costs/models.py @@ -106,6 +106,33 @@ class PrefixFingerprints(BaseModel): message_count: Optional[int] = Field(None, alias="messageCount") +class PrefixTokens(BaseModel): + """The agent's stable static prefix, split: system prompt vs tool schemas.""" + model_config = ConfigDict(populate_by_name=True) + + system: int = 0 + tools: int = 0 + + +class CompactionEvent(BaseModel): + """One compaction decision recorded before a model call (numbers only). + + ``kind``: ``applied`` (restore-time slice ran), ``checkpoint`` (a new + checkpoint was cut after the previous turn), ``forced`` / ``floor_unreachable`` + (reserved for the scheduling policy). ``summaryTokens`` is the summary's + size at that moment — the number a summary cap has to move. + """ + model_config = ConfigDict(populate_by_name=True, extra="allow") + + kind: str + checkpoint: Optional[int] = None + summary_tokens: Optional[int] = Field(None, alias="summaryTokens") + summarized_turns: Optional[int] = Field(None, alias="summarizedTurns") + retained_messages: Optional[int] = Field(None, alias="retainedMessages") + truncated_tool_results: Optional[int] = Field(None, alias="truncatedToolResults") + input_tokens: Optional[int] = Field(None, alias="inputTokens") + + class SessionCallRow(BaseModel): """One model call within a session's cost anatomy.""" model_config = ConfigDict(populate_by_name=True) @@ -144,6 +171,16 @@ class SessionCallRow(BaseModel): prefix_fingerprints: Optional[PrefixFingerprints] = Field( None, alias="prefixFingerprints" ) + # Context ledger (optional; absent on rows written before it shipped or + # while COST_DIAGNOSTICS_ENABLED=false). + prefix_tokens: Optional[PrefixTokens] = Field(None, alias="prefixTokens") + # The conversation window's cumulative trimmed-message count at this call. + window_removed_messages: Optional[int] = Field(None, alias="windowRemovedMessages") + # Messages trimmed since the previous ledger-bearing call — derived, so a + # reader does not have to diff consecutive rows. A positive value means + # the prefix changed before this call. + window_trimmed: Optional[int] = Field(None, alias="windowTrimmed") + compaction_events: Optional[List[CompactionEvent]] = Field(None, alias="compactionEvents") class SessionCostAnatomy(BaseModel): @@ -306,6 +343,11 @@ class UserSessionSummary(BaseModel): tool_call_count: Optional[int] = Field(None, alias="toolCallCount") tool_error_count: Optional[int] = Field(None, alias="toolErrorCount") compaction_count: Optional[int] = Field(None, alias="compactionCount") + compaction_applied_count: Optional[int] = Field(None, alias="compactionAppliedCount") + compaction_forced_count: Optional[int] = Field(None, alias="compactionForcedCount") + compaction_floor_unreachable_count: Optional[int] = Field( + None, alias="compactionFloorUnreachableCount" + ) diagnosis_count: int = Field(0, alias="diagnosisCount") top_diagnosis_severity: Optional[str] = Field(None, alias="topDiagnosisSeverity") @@ -356,6 +398,10 @@ class ContextTrajectoryPoint(BaseModel): cost: Optional[float] = None # Per-call tool census when recorded (PR-3): tool name -> calls tool_calls: Optional[Dict[str, int]] = Field(None, alias="toolCalls") + # Context ledger when recorded: messages trimmed before this call, and the + # kinds of compaction decision taken before it. + window_trimmed: Optional[int] = Field(None, alias="windowTrimmed") + compaction: Optional[List[str]] = None class FingerprintChanges(BaseModel): @@ -384,6 +430,9 @@ class DataCoverage(BaseModel): compaction_count: bool = Field(False, alias="compactionCount") fingerprints: bool = False cost: bool = False + prefix_tokens: bool = Field(False, alias="prefixTokens") + window_trim: bool = Field(False, alias="windowTrim") + compaction_events: bool = Field(False, alias="compactionEvents") class SessionProfile(BaseModel): @@ -414,3 +463,15 @@ class SessionProfile(BaseModel): diagnoses: List[SessionDiagnosis] = Field(default_factory=list) data_coverage: DataCoverage = Field(default_factory=DataCoverage, alias="dataCoverage") + # Latest recorded static prefix split (system prompt vs tool schemas). + prefix_tokens: Optional[PrefixTokens] = Field(None, alias="prefixTokens") + # How many calls in this session were preceded by a window trim, and the + # messages the window has removed in total (last ledger-bearing call). + window_trim_calls: int = Field(0, alias="windowTrimCalls") + window_removed_messages: Optional[int] = Field(None, alias="windowRemovedMessages") + # Compaction decisions by kind across the session's calls. + compaction_event_counts: Dict[str, int] = Field( + default_factory=dict, alias="compactionEventCounts" + ) + # The summary's token size at the most recent compaction decision. + last_summary_tokens: Optional[int] = Field(None, alias="lastSummaryTokens") diff --git a/backend/src/apis/app_api/admin/costs/service.py b/backend/src/apis/app_api/admin/costs/service.py index 4e266b5d..d355306c 100644 --- a/backend/src/apis/app_api/admin/costs/service.py +++ b/backend/src/apis/app_api/admin/costs/service.py @@ -20,6 +20,8 @@ top_severity, ) from .models import ( + CompactionEvent, + PrefixTokens, AttachmentProfile, ContextTrajectoryPoint, DataCoverage, @@ -86,6 +88,53 @@ def _context_tokens(record: Dict[str, Any]) -> int: ) +from dataclasses import dataclass as _dataclass, field as _field + + +@_dataclass +class _CallLedger: + """The context-ledger fields of one cost row, decoded and diffed.""" + + prefix_tokens: Optional[PrefixTokens] = None + removed: Optional[int] = None + trimmed: Optional[int] = None + events: List[CompactionEvent] = _field(default_factory=list) + + +def _call_ledger(record: Dict[str, Any], previous_removed: Optional[int]) -> _CallLedger: + """Decode a cost row's ``prefixTokens`` / ``windowRemovedMessages`` / + ``compactionEvents`` and derive ``trimmed`` (messages removed since the + previous ledger-bearing row). Absent fields stay ``None`` — "not tracked", + never 0 — and malformed ones are ignored rather than raised. + """ + ledger = _CallLedger() + raw_prefix = record.get("prefixTokens") + if isinstance(raw_prefix, dict): + try: + ledger.prefix_tokens = PrefixTokens( + system=int(raw_prefix.get("system") or 0), + tools=int(raw_prefix.get("tools") or 0), + ) + except (TypeError, ValueError): + ledger.prefix_tokens = None + removed = _as_int(record.get("windowRemovedMessages")) + if removed is not None: + ledger.removed = removed + ledger.trimmed = ( + max(removed - previous_removed, 0) if previous_removed is not None else 0 + ) + raw_events = record.get("compactionEvents") + if isinstance(raw_events, list): + for entry in raw_events: + if not isinstance(entry, dict) or not entry.get("kind"): + continue + try: + ledger.events.append(CompactionEvent(**entry)) + except (TypeError, ValueError): + continue + return ledger + + class AdminCostService: """Service for admin cost dashboard operations.""" @@ -614,11 +663,15 @@ async def get_session_cost_anatomy(self, session_id: str) -> SessionCostAnatomy: wasted_usd = 0.0 agent_switch_misses = 0 agent_switch_usd = 0.0 + previous_removed: Optional[int] = None for record in records: token_usage = record.get("tokenUsage") or {} model_info = record.get("modelInfo") or {} fingerprints_raw = record.get("prefixFingerprints") + ledger = _call_ledger(record, previous_removed) + if ledger.removed is not None: + previous_removed = ledger.removed # cost is a breakdown dict ({"total": ...}) on the streaming path # or a bare float on the legacy path. @@ -675,6 +728,10 @@ async def get_session_cost_anatomy(self, session_id: str) -> SessionCostAnatomy: PrefixFingerprints(**fingerprints_raw) if isinstance(fingerprints_raw, dict) else None ), + prefix_tokens=ledger.prefix_tokens, + window_removed_messages=ledger.removed, + window_trimmed=ledger.trimmed, + compaction_events=ledger.events or None, )) cache_traffic = total_cache_read + total_cache_write @@ -800,6 +857,11 @@ def _session_summary( tool_call_count=_as_int(row.get("toolCallCount")), tool_error_count=_as_int(row.get("toolErrorCount")), compaction_count=_as_int(row.get("compactionCount")), + compaction_applied_count=_as_int(row.get("compactionAppliedCount")), + compaction_forced_count=_as_int(row.get("compactionForcedCount")), + compaction_floor_unreachable_count=_as_int( + row.get("compactionFloorUnreachableCount") + ), diagnosis_count=len(findings), top_diagnosis_severity=top_severity(findings), ) @@ -936,6 +998,13 @@ async def get_session_profile(self, session_id: str) -> Optional[SessionProfile] read_total = write_total = 0 any_fingerprints = any_census = False previous_fp: Optional[Dict[str, Any]] = None + any_prefix_tokens = any_window = any_compaction_events = False + prefix_tokens: Optional[PrefixTokens] = None + previous_removed: Optional[int] = None + last_removed: Optional[int] = None + window_trim_calls = 0 + compaction_event_counts: Counter = Counter() + last_summary_tokens: Optional[int] = None for index, record in enumerate(records): usage = record.get("tokenUsage") or {} @@ -981,6 +1050,21 @@ async def get_session_profile(self, session_id: str) -> Optional[SessionProfile] slot.calls += calls slot.errors += errors + ledger = _call_ledger(record, previous_removed) + if ledger.prefix_tokens is not None: + any_prefix_tokens = True + prefix_tokens = ledger.prefix_tokens + if ledger.removed is not None: + any_window = True + previous_removed = last_removed = ledger.removed + if ledger.trimmed: + window_trim_calls += 1 + if ledger.events: + any_compaction_events = True + for event in ledger.events: + compaction_event_counts[event.kind] += 1 + if event.summary_tokens is not None: + last_summary_tokens = event.summary_tokens trajectory.append(ContextTrajectoryPoint( call_index=index, timestamp=record.get("timestamp", ""), @@ -989,6 +1073,8 @@ async def get_session_profile(self, session_id: str) -> Optional[SessionProfile] model_id=model_id, cost=_record_cost(record), tool_calls=point_tool_calls, + window_trimmed=ledger.trimmed, + compaction=[e.kind for e in ledger.events] or None, )) # Cache totals: the rows are authoritative when present, else the @@ -1051,7 +1137,17 @@ async def get_session_profile(self, session_id: str) -> Optional[SessionProfile] compaction_count=row.get("compactionCount") is not None, fingerprints=any_fingerprints, cost=total_cost is not None, + prefix_tokens=any_prefix_tokens, + window_trim=any_window, + compaction_events=( + any_compaction_events or row.get("compactionAppliedCount") is not None + ), ), + prefix_tokens=prefix_tokens, + window_trim_calls=window_trim_calls, + window_removed_messages=last_removed, + compaction_event_counts=dict(compaction_event_counts), + last_summary_tokens=last_summary_tokens, ) async def get_dashboard( diff --git a/backend/src/apis/shared/observability/content_policy.py b/backend/src/apis/shared/observability/content_policy.py index 795485d2..5867f121 100644 --- a/backend/src/apis/shared/observability/content_policy.py +++ b/backend/src/apis/shared/observability/content_policy.py @@ -127,6 +127,11 @@ def is_content_bearing(path: str) -> bool: "toolCallCount", "toolErrorCount", "compactionCount", + # Compaction decisions by kind (per-call ledger rolled up; see + # `TurnBasedSessionManager.record_compaction_event`) + "compactionAppliedCount", + "compactionForcedCount", + "compactionFloorUnreachableCount", ) #: C# rows for the cost anatomy and the session profile's trajectory. @@ -146,6 +151,13 @@ def is_content_bearing(path: str) -> bool: "prefixFingerprints", "contextWindow", "toolCalls", # per-call census, optional (PR-3) + # Context ledger, optional: the agent's stable prefix split + # ({system, tools} tokens), the conversation window's cumulative trim + # count, and the compaction decisions taken before this call — all + # numbers, never text. + "prefixTokens", + "windowRemovedMessages", + "compactionEvents", ) #: FILE# rows for the session profile's attachment summary. diff --git a/backend/src/apis/shared/sessions/metadata.py b/backend/src/apis/shared/sessions/metadata.py index 8d336ae8..f58a0486 100644 --- a/backend/src/apis/shared/sessions/metadata.py +++ b/backend/src/apis/shared/sessions/metadata.py @@ -1742,6 +1742,14 @@ async def _bump_session_aggregates( update_parts_add.append("toolErrorCount :toolErrors") values[":toolCalls"] = tool_calls_total values[":toolErrors"] = tool_errors_total + # Compaction decisions, counted per kind from the call's + # `compactionEvents` ledger. `checkpoint` is deliberately not + # here — `_save_compaction_state(record_event=True)` already + # bumps `compactionCount` for it, and two counters for one event + # would disagree under concurrency. + for kind, attr in _COMPACTION_EVENT_COUNTERS.items(): + update_parts_add.append(f"{attr} :{attr}") + values[f":{attr}"] = _compaction_event_count(message_metadata, kind) update_expression = ( "ADD " + ", ".join(update_parts_add) + " SET " + ", ".join(update_parts_set) @@ -1764,6 +1772,29 @@ async def _bump_session_aggregates( logger.debug("bump_session_aggregates failed (will be backfilled on read): %s", e) +#: Session-row counter per compaction event kind (see +#: ``TurnBasedSessionManager.record_compaction_event``). Written as 0 while +#: the diagnostics are on so the attribute exists from the first call. +_COMPACTION_EVENT_COUNTERS = { + "applied": "compactionAppliedCount", + "forced": "compactionForcedCount", + "floor_unreachable": "compactionFloorUnreachableCount", +} + + +def _compaction_event_count(message_metadata: Any, kind: str) -> int: + """How many events of ``kind`` the call's ``compactionEvents`` extra carries. + + Malformed entries count as zero rather than raising — the aggregate bump + must never fail on them. + """ + extra = getattr(message_metadata, "model_extra", None) + events = extra.get("compactionEvents") if isinstance(extra, dict) else None + if not isinstance(events, list): + return 0 + return sum(1 for e in events if isinstance(e, dict) and e.get("kind") == kind) + + def _tool_census_totals(message_metadata: Any) -> tuple[int, int]: """``(calls, errors)`` summed over the call's ``toolCalls`` extra field. diff --git a/backend/tests/agents/main_agent/session/test_compaction_event_ledger.py b/backend/tests/agents/main_agent/session/test_compaction_event_ledger.py new file mode 100644 index 00000000..1a240b7e --- /dev/null +++ b/backend/tests/agents/main_agent/session/test_compaction_event_ledger.py @@ -0,0 +1,45 @@ +"""`TurnBasedSessionManager.record_compaction_event` / `drain_compaction_events` +— the queue the `ContextLedgerHook` drains onto the next model call's cost row.""" + +import pytest + +from agents.main_agent.session.turn_based_session_manager import ( + _MAX_PENDING_COMPACTION_EVENTS, + COMPACTION_EVENT_KINDS, +) + + +@pytest.fixture(autouse=True) +def diagnostics_enabled(monkeypatch): + monkeypatch.delenv("COST_DIAGNOSTICS_ENABLED", raising=False) + + +def test_events_carry_numbers_only_and_drain_once(make_session_manager): + manager = make_session_manager() + manager.record_compaction_event( + "applied", checkpoint=12, summaryTokens=900.7, retainedMessages=30, + summary="never persisted", flag=True, + ) + drained = manager.drain_compaction_events() + assert drained == [{"kind": "applied", "checkpoint": 12, "summaryTokens": 900, "retainedMessages": 30}] + assert manager.drain_compaction_events() == [] + + +def test_unknown_kinds_are_ignored_and_the_queue_is_bounded(make_session_manager): + manager = make_session_manager() + manager.record_compaction_event("mystery", checkpoint=1) + assert manager.drain_compaction_events() == [] + for _ in range(_MAX_PENDING_COMPACTION_EVENTS + 3): + manager.record_compaction_event("forced") + assert len(manager.drain_compaction_events()) == _MAX_PENDING_COMPACTION_EVENTS + + +def test_reserved_kinds_exist_for_the_scheduling_policy(): + assert {"applied", "checkpoint", "forced", "floor_unreachable"} <= COMPACTION_EVENT_KINDS + + +def test_kill_switch_records_nothing(make_session_manager, monkeypatch): + monkeypatch.setenv("COST_DIAGNOSTICS_ENABLED", "false") + manager = make_session_manager() + manager.record_compaction_event("applied", checkpoint=3) + assert manager.drain_compaction_events() == [] diff --git a/backend/tests/agents/main_agent/session/test_compaction_policy.py b/backend/tests/agents/main_agent/session/test_compaction_policy.py index eb8b5aea..11b8d73b 100644 --- a/backend/tests/agents/main_agent/session/test_compaction_policy.py +++ b/backend/tests/agents/main_agent/session/test_compaction_policy.py @@ -310,7 +310,11 @@ async def test_forced_and_floor_unreachable_are_recorded(self, make_session_mana assert floor_call.kwargs["retainedTokens"] > 250 @pytest.mark.asyncio - async def test_no_ledger_is_a_noop(self, make_session_manager, compaction_config): + async def test_no_ledger_is_a_noop(self, make_session_manager, compaction_config, monkeypatch): + # The cost-diagnostics ledger now ships on this class, so absence has to + # be simulated: strip the recorder and prove the cut still lands rather + # than raising through ``_record_ledger_event``'s getattr seam. mgr = _armed_manager(make_session_manager, compaction_config) + monkeypatch.delattr(type(mgr), "record_compaction_event") assert not hasattr(mgr, "record_compaction_event") assert await mgr.update_after_turn(1200) is not None # no AttributeError diff --git a/backend/tests/agents/main_agent/session/test_context_ledger_hook.py b/backend/tests/agents/main_agent/session/test_context_ledger_hook.py new file mode 100644 index 00000000..039ac6b9 --- /dev/null +++ b/backend/tests/agents/main_agent/session/test_context_ledger_hook.py @@ -0,0 +1,109 @@ +"""`ContextLedgerHook` — per model call, the window's trimmed-message count and +the compaction decisions taken since the previous call. + +Same lifecycle contract as the tool census: keyed by the turn's Nth model +call, reset per turn, read (never drained) at turn end, off with the same +kill switch, and never able to break a turn. +""" + +from unittest.mock import MagicMock + +import pytest + +from agents.main_agent.session.hooks.context_ledger import _MAX_EVENTS_PER_CALL, ContextLedgerHook + + +@pytest.fixture(autouse=True) +def diagnostics_enabled(monkeypatch): + monkeypatch.delenv("COST_DIAGNOSTICS_ENABLED", raising=False) + + +class _Manager: + def __init__(self, events=None): + self._events = list(events or []) + + def drain_compaction_events(self): + events, self._events = self._events, [] + return events + + +def _agent(removed=None, manager=None): + agent = MagicMock() + agent.conversation_manager.removed_message_count = removed + agent._session_manager = manager + return agent + + +def _call(hook, agent): + hook._on_before_model_call(MagicMock(agent=agent)) + + +def test_records_the_window_count_and_drains_compaction_events_per_call(): + hook = ContextLedgerHook() + manager = _Manager([{"kind": "applied", "checkpoint": 12, "summaryTokens": 900}]) + hook._on_turn_start(MagicMock()) + _call(hook, _agent(removed=0, manager=manager)) + _call(hook, _agent(removed=4, manager=manager)) # nothing left to drain + + assert hook.ledger_for_call(0) == { + "windowRemovedMessages": 0, + "compactionEvents": [{"kind": "applied", "checkpoint": 12, "summaryTokens": 900}], + } + assert hook.ledger_for_call(1) == {"windowRemovedMessages": 4} + assert hook.ledger_for_call(2) is None + + +def test_a_new_turn_forgets_the_previous_one(): + hook = ContextLedgerHook() + hook._on_turn_start(MagicMock()) + _call(hook, _agent(removed=7)) + hook._on_turn_start(MagicMock()) + assert hook.ledger_for_call(0) is None + + +def test_reads_do_not_drain_and_do_not_alias(): + hook = ContextLedgerHook() + hook._on_turn_start(MagicMock()) + _call(hook, _agent(removed=1, manager=_Manager([{"kind": "checkpoint"}]))) + first = hook.ledger_for_call(0) + first["compactionEvents"].append({"kind": "tampered"}) + assert hook.ledger_for_call(0)["compactionEvents"] == [{"kind": "checkpoint"}] + + +def test_no_window_manager_and_no_session_manager_records_nothing(): + hook = ContextLedgerHook() + hook._on_turn_start(MagicMock()) + agent = MagicMock() + agent.conversation_manager = None + agent._session_manager = None + _call(hook, agent) + assert hook.ledger_for_call(0) is None + + +def test_kill_switch_records_nothing_and_reads_none(monkeypatch): + monkeypatch.setenv("COST_DIAGNOSTICS_ENABLED", "false") + hook = ContextLedgerHook() + hook._on_turn_start(MagicMock()) + _call(hook, _agent(removed=3, manager=_Manager([{"kind": "applied"}]))) + assert hook.ledger_for_call(0) is None + + +def test_event_list_is_bounded_and_malformed_counts_never_raise(): + hook = ContextLedgerHook() + hook._on_turn_start(MagicMock()) + many = [{"kind": "applied"}] * (_MAX_EVENTS_PER_CALL + 5) + _call(hook, _agent(removed="not-a-number", manager=_Manager(many))) + entry = hook.ledger_for_call(0) + assert "windowRemovedMessages" not in entry + assert len(entry["compactionEvents"]) == _MAX_EVENTS_PER_CALL + + +def test_a_raising_manager_is_swallowed(): + class Broken: + def drain_compaction_events(self): + raise RuntimeError("boom") + + hook = ContextLedgerHook() + hook._on_turn_start(MagicMock()) + _call(hook, _agent(removed=2, manager=Broken())) + assert hook.ledger_for_call(0) == {"windowRemovedMessages": 2} diff --git a/backend/tests/agents/main_agent/streaming/test_context_ledger_attach.py b/backend/tests/agents/main_agent/streaming/test_context_ledger_attach.py new file mode 100644 index 00000000..9b98a00a --- /dev/null +++ b/backend/tests/agents/main_agent/streaming/test_context_ledger_attach.py @@ -0,0 +1,95 @@ +"""The stream coordinator persists the context ledger and the prefix split on +the call's cost row, as extra fields next to `toolCalls`.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from agents.main_agent.session.hooks.context_attribution import _SPLIT_ATTR +from agents.main_agent.streaming.stream_coordinator import StreamCoordinator + + +def _coordinator() -> StreamCoordinator: + return object.__new__(StreamCoordinator) + + +def _usage_metadata(): + return {"usage": {"inputTokens": 100, "outputTokens": 20, "totalTokens": 120}} + + +def _wrapper(strands_agent): + # A plain namespace, not a MagicMock: the coordinator reads + # `model_config.model_id` into a pydantic model and a mock is not a str. + return SimpleNamespace( + agent=strands_agent, + model_config=SimpleNamespace(model_id="claude-haiku-4-5", model_name="Claude Haiku 4.5"), + ) + + +def _wrapper_with_split(system=12_000, tools=48_000): + strands_agent = SimpleNamespace() + setattr(strands_agent, _SPLIT_ATTR, {"systemTokens": system, "toolTokens": tools}) + return _wrapper(strands_agent) + + +async def _store(monkeypatch, **kwargs): + monkeypatch.delenv("COST_DIAGNOSTICS_ENABLED", raising=False) + store = AsyncMock() + with patch("apis.shared.sessions.metadata.store_message_metadata", store), \ + patch("agents.main_agent.streaming.stream_coordinator.get_prefix_fingerprint", return_value=None): + await _coordinator()._store_message_metadata( + session_id="s1", user_id="u1", message_id=3, + accumulated_metadata=_usage_metadata(), + stream_start_time=0.0, stream_end_time=1.0, first_token_time=0.5, + call_index=0, **kwargs, + ) + store.assert_awaited_once() + return store.await_args.kwargs["message_metadata"] + + +@pytest.mark.asyncio +async def test_ledger_and_prefix_split_are_attached(monkeypatch): + stored = await _store( + monkeypatch, + agent=_wrapper_with_split(), + context_ledger={ + "windowRemovedMessages": 6, + "compactionEvents": [{"kind": "applied", "summaryTokens": 1200}], + }, + ) + extra = stored.model_extra + assert extra["windowRemovedMessages"] == 6 + assert extra["compactionEvents"] == [{"kind": "applied", "summaryTokens": 1200}] + assert extra["prefixTokens"] == {"system": 12_000, "tools": 48_000} + dumped = stored.model_dump(by_alias=True) + assert dumped["prefixTokens"]["tools"] == 48_000 + + +@pytest.mark.asyncio +async def test_absent_ledger_and_split_leave_no_fields(monkeypatch): + stored = await _store(monkeypatch, agent=_wrapper(SimpleNamespace()), context_ledger=None) + for key in ("windowRemovedMessages", "compactionEvents", "prefixTokens"): + assert key not in (stored.model_extra or {}) + + +@pytest.mark.asyncio +async def test_kill_switch_drops_the_prefix_split_too(monkeypatch): + monkeypatch.setenv("COST_DIAGNOSTICS_ENABLED", "false") + store = AsyncMock() + with patch("apis.shared.sessions.metadata.store_message_metadata", store), \ + patch("agents.main_agent.streaming.stream_coordinator.get_prefix_fingerprint", return_value=None): + await _coordinator()._store_message_metadata( + session_id="s1", user_id="u1", message_id=3, + accumulated_metadata=_usage_metadata(), + stream_start_time=0.0, stream_end_time=1.0, first_token_time=0.5, + agent=_wrapper_with_split(), call_index=0, context_ledger=None, + ) + stored = store.await_args.kwargs["message_metadata"] + assert "prefixTokens" not in (stored.model_extra or {}) + + +@pytest.mark.asyncio +async def test_the_argument_is_optional_for_the_interrupt_path(monkeypatch): + stored = await _store(monkeypatch, agent=None) + assert "windowRemovedMessages" not in (stored.model_extra or {}) diff --git a/backend/tests/apis/app_api/admin/costs/test_drilldown_routes.py b/backend/tests/apis/app_api/admin/costs/test_drilldown_routes.py index 7f05d079..0d5d9f56 100644 --- a/backend/tests/apis/app_api/admin/costs/test_drilldown_routes.py +++ b/backend/tests/apis/app_api/admin/costs/test_drilldown_routes.py @@ -102,7 +102,10 @@ def test_session_profile_returns_200(): assert resp.status_code == 200 body = resp.json() assert body["sessionId"] == "s1" and body["compactionThreshold"] == 100_000 - assert body["dataCoverage"] == {"toolCensus": False, "compactionCount": False, "fingerprints": False, "cost": False} + assert body["dataCoverage"] == { + "toolCensus": False, "compactionCount": False, "fingerprints": False, "cost": False, + "prefixTokens": False, "windowTrim": False, "compactionEvents": False, + } service.get_session_profile.assert_awaited_once_with("s1") diff --git a/backend/tests/apis/app_api/admin/costs/test_session_profile_service.py b/backend/tests/apis/app_api/admin/costs/test_session_profile_service.py index 83a10ab1..ea7f6ebf 100644 --- a/backend/tests/apis/app_api/admin/costs/test_session_profile_service.py +++ b/backend/tests/apis/app_api/admin/costs/test_session_profile_service.py @@ -206,3 +206,51 @@ async def test_the_whole_profile_serializes_without_content_bearing_keys(): files = [{"uploadId": "a", "mimeType": "application/pdf", "sizeBytes": 1}] p = await _service(_row(), [_call(0, tool_calls={"t": {"calls": 1, "errors": 0}})], files=files).get_session_profile("s1") assert content_bearing_paths(p.model_dump(by_alias=True)) == [] + + +@pytest.mark.asyncio +async def test_context_ledger_is_decoded_diffed_and_covered(): + records = [ + _call(0, read=10_000), + _call(1, read=10_000), + _call(2, read=10_000), + _call(3, read=10_000), + ] + # Rows carry the ledger from call 1 on: a stable prefix split, a window + # count that rises before call 3 (a trim), and one compaction decision + # with the summary's size at that moment. + records[1]["prefixTokens"] = {"system": 12_000, "tools": 48_000} + records[1]["windowRemovedMessages"] = 0 + records[2]["windowRemovedMessages"] = 0 + records[3]["windowRemovedMessages"] = 8 + records[3]["compactionEvents"] = [{"kind": "applied", "checkpoint": 12, "summaryTokens": 2_300}] + + p = await _service(_row(), records).get_session_profile("s1") + + assert p.prefix_tokens.system == 12_000 and p.prefix_tokens.tools == 48_000 + assert p.window_trim_calls == 1 + assert p.window_removed_messages == 8 + assert p.compaction_event_counts == {"applied": 1} + assert p.last_summary_tokens == 2_300 + assert p.data_coverage.prefix_tokens and p.data_coverage.window_trim and p.data_coverage.compaction_events + trimmed = [pt.window_trimmed for pt in p.context_trajectory] + assert trimmed == [None, 0, 0, 8] + assert p.context_trajectory[3].compaction == ["applied"] + + +@pytest.mark.asyncio +async def test_rows_without_a_ledger_read_not_tracked(): + p = await _service(_row(), [_call(0), _call(1)]).get_session_profile("s1") + assert p.prefix_tokens is None + assert p.window_removed_messages is None and p.window_trim_calls == 0 + assert p.compaction_event_counts == {} and p.last_summary_tokens is None + assert not p.data_coverage.prefix_tokens and not p.data_coverage.window_trim + assert not p.data_coverage.compaction_events + assert all(pt.window_trimmed is None and pt.compaction is None for pt in p.context_trajectory) + + +@pytest.mark.asyncio +async def test_session_row_counters_alone_mark_compaction_events_as_tracked(): + p = await _service(_row(compactionAppliedCount=0), [_call(0)]).get_session_profile("s1") + assert p.data_coverage.compaction_events + assert p.session.compaction_applied_count == 0 diff --git a/backend/tests/shared/test_context_ledger_persistence.py b/backend/tests/shared/test_context_ledger_persistence.py new file mode 100644 index 00000000..1ceb884c --- /dev/null +++ b/backend/tests/shared/test_context_ledger_persistence.py @@ -0,0 +1,102 @@ +"""Context-ledger extras land on the `C#` row and compaction decisions roll up +by kind on the session row (moto).""" + +from decimal import Decimal + +import pytest + +from apis.shared.sessions.models import MessageMetadata, ModelInfo, TokenUsage + + +def _meta(**extra): + return MessageMetadata( + token_usage=TokenUsage(inputTokens=100, outputTokens=50, totalTokens=150), + model_info=ModelInfo(modelId="claude-haiku-4-5", modelName="Claude Haiku 4.5"), + cost=0.01, + **extra, + ) + + +def _seed_session(table, session_id="s1", user_id="u1"): + table.put_item(Item={ + "PK": f"USER#{user_id}", "SK": f"S#{session_id}", + "GSI_PK": f"SESSION#{session_id}", "GSI_SK": "META", + "sessionId": session_id, "userId": user_id, "status": "active", + "createdAt": "2026-09-01T00:00:00Z", "lastMessageAt": "2026-09-01T00:00:00Z", + "messageCount": Decimal(0), + }) + + +def _session_row(table, session_id="s1", user_id="u1"): + return table.get_item(Key={"PK": f"USER#{user_id}", "SK": f"S#{session_id}"})["Item"] + + +@pytest.mark.asyncio +async def test_ledger_lands_on_the_cost_row_and_events_roll_up_by_kind(sessions_metadata_table, monkeypatch): + monkeypatch.delenv("COST_DIAGNOSTICS_ENABLED", raising=False) + from apis.shared.sessions.metadata import store_message_metadata + + _seed_session(sessions_metadata_table) + await store_message_metadata( + session_id="s1", user_id="u1", message_id=1, + message_metadata=_meta( + prefixTokens={"system": 12_000, "tools": 48_000}, + windowRemovedMessages=0, + compactionEvents=[ + {"kind": "applied", "checkpoint": 12, "summaryTokens": 900}, + {"kind": "checkpoint", "checkpoint": 12, "summaryTokens": 900}, + ], + ), + ) + await store_message_metadata( + session_id="s1", user_id="u1", message_id=2, + message_metadata=_meta(windowRemovedMessages=4, compactionEvents=[{"kind": "forced"}]), + ) + + cost_rows = sorted( + (i for i in sessions_metadata_table.scan()["Items"] if i["SK"].startswith("C#")), + key=lambda r: r["SK"], + ) + assert len(cost_rows) == 2 + first = [r for r in cost_rows if "prefixTokens" in r][0] + assert first["prefixTokens"] == {"system": Decimal(12_000), "tools": Decimal(48_000)} + assert first["windowRemovedMessages"] == Decimal(0) + assert first["compactionEvents"][0]["kind"] == "applied" + assert [r["windowRemovedMessages"] for r in cost_rows if "prefixTokens" not in r] == [Decimal(4)] + + row = _session_row(sessions_metadata_table) + assert row["compactionAppliedCount"] == Decimal(1) + assert row["compactionForcedCount"] == Decimal(1) + assert row["compactionFloorUnreachableCount"] == Decimal(0) + # `checkpoint` is counted by `_save_compaction_state(record_event=True)`, + # never here — two counters for one event would disagree. + assert "compactionCheckpointCount" not in row + + +@pytest.mark.asyncio +async def test_kill_switch_writes_no_counters(sessions_metadata_table, monkeypatch): + monkeypatch.setenv("COST_DIAGNOSTICS_ENABLED", "false") + from apis.shared.sessions.metadata import store_message_metadata + + _seed_session(sessions_metadata_table) + await store_message_metadata( + session_id="s1", user_id="u1", message_id=1, + message_metadata=_meta(compactionEvents=[{"kind": "applied"}]), + ) + row = _session_row(sessions_metadata_table) + for attr in ("compactionAppliedCount", "compactionForcedCount", "compactionFloorUnreachableCount"): + assert attr not in row + + +@pytest.mark.asyncio +async def test_malformed_events_count_as_zero_not_as_a_failure(sessions_metadata_table, monkeypatch): + monkeypatch.delenv("COST_DIAGNOSTICS_ENABLED", raising=False) + from apis.shared.sessions.metadata import store_message_metadata + + _seed_session(sessions_metadata_table) + await store_message_metadata( + session_id="s1", user_id="u1", message_id=1, + message_metadata=_meta(compactionEvents=["applied", {"nokind": 1}, {"kind": "applied"}]), + ) + row = _session_row(sessions_metadata_table) + assert row["compactionAppliedCount"] == Decimal(1) diff --git a/frontend/ai.client/src/app/admin/costs/models/admin-cost.models.ts b/frontend/ai.client/src/app/admin/costs/models/admin-cost.models.ts index b5c17906..0f500c9e 100644 --- a/frontend/ai.client/src/app/admin/costs/models/admin-cost.models.ts +++ b/frontend/ai.client/src/app/admin/costs/models/admin-cost.models.ts @@ -118,6 +118,28 @@ export interface PrefixFingerprints { messageCount?: number | null; } +/** The agent's stable static prefix, split: system prompt vs tool schemas. */ +export interface PrefixTokens { + system: number; + tools: number; +} + +/** + * One compaction decision recorded before a model call (numbers only). + * `kind`: `applied` (restore-time slice ran), `checkpoint` (a new checkpoint + * was cut after the previous turn), `forced` / `floor_unreachable` (the + * scheduling policy). `summaryTokens` is the summary's size at that moment. + */ +export interface CompactionEvent { + kind: string; + checkpoint?: number | null; + summaryTokens?: number | null; + summarizedTurns?: number | null; + retainedMessages?: number | null; + truncatedToolResults?: number | null; + inputTokens?: number | null; +} + /** One model call within a session's cost anatomy. */ export interface SessionCallRow { timestamp: string; @@ -152,6 +174,13 @@ export interface SessionCallRow { turnAgentId?: string | null; agentSwitched?: boolean; prefixFingerprints?: PrefixFingerprints | null; + /** Context ledger — absent on rows written before it shipped or with diagnostics off. */ + prefixTokens?: PrefixTokens | null; + /** The conversation window's cumulative trimmed-message count at this call. */ + windowRemovedMessages?: number | null; + /** Messages trimmed since the previous ledger-bearing call; > 0 means the prefix changed before this call. */ + windowTrimmed?: number | null; + compactionEvents?: CompactionEvent[] | null; } /** Per-call cost anatomy for one session (admin cache-miss forensics). */ @@ -268,6 +297,9 @@ export interface UserSessionSummary { toolCallCount?: number | null; toolErrorCount?: number | null; compactionCount?: number | null; + compactionAppliedCount?: number | null; + compactionForcedCount?: number | null; + compactionFloorUnreachableCount?: number | null; diagnosisCount: number; topDiagnosisSeverity?: DiagnosisSeverity | null; } @@ -318,6 +350,10 @@ export interface ContextTrajectoryPoint { cost?: number | null; /** Per-call tool census when recorded: tool name → calls. */ toolCalls?: Record | null; + /** Messages trimmed before this call, when the ledger recorded it. */ + windowTrimmed?: number | null; + /** Kinds of compaction decision taken before this call. */ + compaction?: string[] | null; } export interface FingerprintChanges { @@ -338,6 +374,9 @@ export interface DataCoverage { compactionCount: boolean; fingerprints: boolean; cost: boolean; + prefixTokens?: boolean; + windowTrim?: boolean; + compactionEvents?: boolean; } /** The content-free diagnostic profile of one conversation. */ @@ -357,6 +396,15 @@ export interface SessionProfile { enabledToolIds: string[]; diagnoses: SessionDiagnosis[]; dataCoverage: DataCoverage; + /** Latest recorded static prefix split (system prompt vs tool schemas). */ + prefixTokens?: PrefixTokens | null; + /** Calls preceded by a window trim, and the messages the window has removed in total. */ + windowTrimCalls?: number; + windowRemovedMessages?: number | null; + /** Compaction decisions by kind across the session's calls. */ + compactionEventCounts?: Record; + /** The summary's token size at the most recent compaction decision. */ + lastSummaryTokens?: number | null; } // ========== API Request Options ========== diff --git a/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.spec.ts b/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.spec.ts index ad3de553..78572a1f 100644 --- a/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.spec.ts +++ b/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.spec.ts @@ -242,6 +242,31 @@ describe('SessionCostAnatomyPage', () => { expect(page.profileNotFound()).toBe(false); }); + it('summarises compaction decisions by kind with the latest summary size', async () => { + const profile = vi.fn().mockReturnValue( + of({ + ...MOCK_PROFILE, + dataCoverage: { ...MOCK_PROFILE.dataCoverage, compactionCount: true, compactionEvents: true, windowTrim: true, prefixTokens: true }, + compactionEventCounts: { forced: 1, applied: 3 }, + lastSummaryTokens: 2_300, + prefixTokens: { system: 12_000, tools: 48_000 }, + windowTrimCalls: 4, + windowRemovedMessages: 16, + }), + ); + const page = setup(vi.fn().mockReturnValue(of(MOCK_ANATOMY)), profile).componentInstance; + await vi.waitFor(() => expect(page.profileResource.hasValue()).toBe(true)); + expect(page.compactionEventsLine()).toBe('3 applied · 1 forced · summary 2.3K'); + }); + + it('describes one compaction event from its numbers only', () => { + const page = setup(vi.fn().mockReturnValue(of(MOCK_ANATOMY))).componentInstance; + expect( + page.compactionEventTitle({ kind: 'floor_unreachable', checkpoint: 12, summaryTokens: 900, retainedMessages: 30 }), + ).toBe('floor unreachable · checkpoint 12 · summary 900 · 30 messages retained'); + expect(page.compactionEventTitle({ kind: 'checkpoint' })).toBe('checkpoint'); + }); + it('survives a missing profile without touching the anatomy', async () => { const page = setup( vi.fn().mockReturnValue(of(MOCK_ANATOMY)), diff --git a/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.ts b/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.ts index 18e290b5..63e37c00 100644 --- a/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.ts +++ b/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.ts @@ -21,7 +21,9 @@ import { import { AdminCostHttpService } from '../services/admin-cost-http.service'; import { SpinnerComponent } from '../../../components/spinner/spinner.component'; import { ContextTrajectoryChartComponent } from '../components/context-trajectory-chart.component'; -import { CacheStatus, DiagnosisSeverity, SessionDiagnosis } from '../models'; +import { CacheStatus, DiagnosisSeverity, SessionDiagnosis, + CompactionEvent, +} from '../models'; import { AnatomyRow, FINGERPRINT_KEYS, @@ -112,7 +114,7 @@ import { @let profile = profileResource.value();

Conversation profile

-
+

Messages

{{ profile.session.messageCount ?? '—' }}

@@ -147,6 +149,9 @@ import {

Compactions

@if (profile.dataCoverage.compactionCount) {

{{ profile.session.compactionCount ?? 0 }}

+ @if (compactionEventsLine(); as events) { +

{{ events }}

+ } } @else {

—

@@ -170,6 +175,35 @@ import {

of {{ formatTokens(profile.session.contextWindow) }} window

}
+
+

Static prefix

+ @if (profile.dataCoverage.prefixTokens && profile.prefixTokens; as prefix) { + +

{{ formatTokens(prefix.system + prefix.tools) }}

+

+ {{ formatTokens(prefix.system) }} system · {{ formatTokens(prefix.tools) }} tools +

+ } @else { +

—

+

not tracked

+ } +
+
+

Window trims

+ @if (profile.dataCoverage.windowTrim) { + +

{{ profile.windowTrimCalls ?? 0 }}

+

+ {{ profile.windowRemovedMessages ?? 0 }} messages removed +

+ } @else { +

—

+

not tracked

+ } +

Write : read

— } + @if (row.call.windowTrimmed; as trimmed) { + + trim −{{ trimmed }} + } + @for (event of row.call.compactionEvents ?? []; track $index) { + {{ event.kind }} + } {{ formatGap(row.call.cacheGapSeconds) }} @@ -540,6 +589,43 @@ import { {{ row.call.prefixFingerprints?.messageCount ?? '—' }}

+
+
+ Window Removed +
+
+ {{ row.call.windowRemovedMessages ?? '—' }} + @if (row.call.windowTrimmed) { + (+{{ row.call.windowTrimmed }} before this call) + } +
+
+
+
+ Static Prefix +
+
+ @if (row.call.prefixTokens; as prefix) { + {{ formatTokens(prefix.system) }} system · {{ formatTokens(prefix.tools) }} tools + } @else { + — + } +
+
+
+
+ Compaction +
+
+ @if (row.call.compactionEvents?.length) { + @for (event of row.call.compactionEvents; track $index) { +
{{ compactionEventTitle(event) }}
+ } + } @else { + — + } +
+
@for (key of fingerprintKeys; track key) {
@@ -688,6 +774,28 @@ export class SessionCostAnatomyPage { * money but are not a regression. The backend reports the explained subset rather than * deducting it, so the page does the subtraction where a reader can see both halves. */ + /** "3 applied · 1 forced · summary 2.3K" — the compaction decisions by kind. */ + readonly compactionEventsLine = computed(() => { + if (!this.profileResource.hasValue()) return ''; + const p = this.profileResource.value(); + const counts = p.compactionEventCounts ?? {}; + const parts = Object.keys(counts) + .sort() + .map((kind) => `${counts[kind]} ${kind.replace('_', ' ')}`); + if (p.lastSummaryTokens != null) parts.push(`summary ${this.formatTokens(p.lastSummaryTokens)}`); + return parts.join(' · '); + }); + + compactionEventTitle(event: CompactionEvent): string { + const parts = [event.kind.replace('_', ' ')]; + if (event.checkpoint != null) parts.push(`checkpoint ${event.checkpoint}`); + if (event.summaryTokens != null) parts.push(`summary ${this.formatTokens(event.summaryTokens)}`); + if (event.summarizedTurns != null) parts.push(`${event.summarizedTurns} turns summarized`); + if (event.retainedMessages != null) parts.push(`${event.retainedMessages} messages retained`); + if (event.truncatedToolResults) parts.push(`${event.truncatedToolResults} tool results truncated`); + return parts.join(' · '); + } + readonly unexplainedMisses = computed(() => { const anatomy = this.anatomyResource.value(); if (!anatomy) return 0;