diff --git a/backend/src/agents/main_agent/config/constants.py b/backend/src/agents/main_agent/config/constants.py index d19cf0a88..d9ed82a7e 100644 --- a/backend/src/agents/main_agent/config/constants.py +++ b/backend/src/agents/main_agent/config/constants.py @@ -28,6 +28,21 @@ class EnvVars: COMPACTION_PROTECTED_TURNS = "AGENTCORE_MEMORY_COMPACTION_PROTECTED_TURNS" COMPACTION_MAX_TOOL_CONTENT_LENGTH = "AGENTCORE_MEMORY_COMPACTION_MAX_TOOL_CONTENT_LENGTH" COMPACTION_CACHE_TTL_SECONDS = "AGENTCORE_MEMORY_COMPACTION_CACHE_TTL_SECONDS" + # Model-relative thresholds (docs/specs/compaction-model-relative-thresholds.md). + # The kill switch reverts to the fixed TOKEN_THRESHOLD and the legacy + # turn-count cut; the ratios/caps shape the per-model ceiling and floor. + COMPACTION_MODEL_RELATIVE_ENABLED = "AGENTCORE_MEMORY_COMPACTION_MODEL_RELATIVE_ENABLED" + COMPACTION_CEILING_RATIO = "AGENTCORE_MEMORY_COMPACTION_CEILING_RATIO" + COMPACTION_CEILING_CAP_TOKENS = "AGENTCORE_MEMORY_COMPACTION_CEILING_CAP_TOKENS" + COMPACTION_FLOOR_RATIO = "AGENTCORE_MEMORY_COMPACTION_FLOOR_RATIO" + COMPACTION_HARD_CEILING_RATIO = "AGENTCORE_MEMORY_COMPACTION_HARD_CEILING_RATIO" + COMPACTION_HARD_CEILING_MULTIPLIER = "AGENTCORE_MEMORY_COMPACTION_HARD_CEILING_MULTIPLIER" + # Strands conversation-manager window (messages). Our compaction owns + # history size; the SDK's default 40-message SlidingWindowConversationManager + # would otherwise slide the front of the list every turn past 40 messages + # (a prefix re-write per turn, and it moves the coordinates the compaction + # checkpoint is expressed in). Setting this to 40 restores the SDK default. + CONVERSATION_WINDOW_MESSAGES = "AGENTCORE_CONVERSATION_WINDOW_MESSAGES" # --- Restored-history repair --- # Kill switch for the restore-time tool-pairing/alternation repair @@ -112,6 +127,27 @@ class Defaults: COMPACTION_MAX_TOOL_CONTENT_LENGTH = 500 # Bedrock prompt-cache TTL (seconds); see CompactionConfig.cache_ttl_seconds COMPACTION_CACHE_TTL_SECONDS = 300 + # Model-relative compaction policy — see + # docs/specs/compaction-model-relative-thresholds.md §3.1 for the table + # these produce. ceiling = min(window * CEILING_RATIO, CEILING_CAP_TOKENS); + # floor = ceiling * FLOOR_RATIO; hard = min(window * HARD_CEILING_RATIO, + # ceiling * HARD_CEILING_MULTIPLIER). COMPACTION_TOKEN_THRESHOLD above is + # the ceiling used when the model's window is unknown. + COMPACTION_MODEL_RELATIVE_ENABLED = True + COMPACTION_CEILING_RATIO = 0.5 + # 100k, not 200k: the 2026-09-15 replay of 20 heavy Sonnet 5 sessions + # priced a 200k/50k policy 43% above 100k/25k on the input side, because + # 36% of cache-write dollars are cold re-writes after a >5 min pause and + # their size is the context at the pause. Raise only on evidence from + # compaction_forced + the cost anatomy (spec §3.1). + COMPACTION_CEILING_CAP_TOKENS = 100_000 + COMPACTION_FLOOR_RATIO = 0.25 + COMPACTION_HARD_CEILING_RATIO = 0.7 + COMPACTION_HARD_CEILING_MULTIPLIER = 1.5 + # Effectively "never trim proactively" — compaction decides what leaves the + # prompt. Overflow recovery (reduce_context on ContextWindowOverflow) still + # works at any window size. + CONVERSATION_WINDOW_MESSAGES = 2000 # --- DynamoDB Tables --- DYNAMODB_QUOTA_TABLE = "UserQuotas" diff --git a/backend/src/agents/main_agent/core/agent_factory.py b/backend/src/agents/main_agent/core/agent_factory.py index 5a0b04e67..e95511546 100644 --- a/backend/src/agents/main_agent/core/agent_factory.py +++ b/backend/src/agents/main_agent/core/agent_factory.py @@ -5,13 +5,14 @@ import logging from typing import List, Optional, Any from strands import Agent +from strands.agent.conversation_manager import SlidingWindowConversationManager from strands.models import BedrockModel from strands.models.openai import OpenAIModel from strands.models.gemini import GeminiModel from strands.tools.executors import SequentialToolExecutor from agents.main_agent.core.bedrock_count_tokens import CountTokensBedrockModel from agents.main_agent.core.model_config import ModelConfig, ModelProvider -from agents.main_agent.config.constants import EnvVars +from agents.main_agent.config.constants import EnvVars, Defaults from apis.shared.models.bedrock_responses import build_bedrock_responses_model from apis.shared.models.mantle import build_mantle_model from apis.shared.models.usage_normalization import usage_normalized @@ -318,9 +319,38 @@ def create_agent( tools=tools, tool_executor=SequentialToolExecutor(), session_manager=session_manager, + conversation_manager=AgentFactory.build_conversation_manager(), hooks=hooks if hooks else None, plugins=plugins if plugins else None, retry_strategy=retry_strategy, ) return agent + + @staticmethod + def build_conversation_manager() -> SlidingWindowConversationManager: + """The Strands conversation manager for the chat agent. + + Left unset, Strands installs ``SlidingWindowConversationManager()`` with + a **40-message** window and runs it after every event-loop cycle. Past + 40 messages that slides the front of ``agent.messages`` every turn, + which (a) re-writes the whole cached prefix each turn — the 2026-09-15 + prod cost audit saw fingerprint ``messageCount`` pinned at 39–41 with + every turn reading only tools+system — and (b) moves the list our + compaction checkpoint is expressed in (spiral-spec D3, ANCHOR_MISMATCH + on 14 of 20 audited sessions). History size is ``TurnBasedSessionManager``'s + job (docs/specs/compaction-model-relative-thresholds.md), so the window + is set large enough never to trim on its own. The manager is kept + (rather than ``NullConversationManager``) because its ``reduce_context`` + is the only ``ContextWindowOverflowException`` recovery in the stack, + and that path does not depend on the window size. + + ``AGENTCORE_CONVERSATION_WINDOW_MESSAGES=40`` restores the SDK default. + """ + raw = os.environ.get(EnvVars.CONVERSATION_WINDOW_MESSAGES, "").strip() + try: + window = int(raw) if raw else Defaults.CONVERSATION_WINDOW_MESSAGES + except ValueError: + window = Defaults.CONVERSATION_WINDOW_MESSAGES + window = max(2, window) + return SlidingWindowConversationManager(window_size=window, should_truncate_results=True) diff --git a/backend/src/agents/main_agent/session/__init__.py b/backend/src/agents/main_agent/session/__init__.py index 527a5b4ec..71386ccea 100644 --- a/backend/src/agents/main_agent/session/__init__.py +++ b/backend/src/agents/main_agent/session/__init__.py @@ -1,6 +1,7 @@ """Session management modules for Strands Agent""" from .session_factory import SessionFactory -from .compaction_models import CompactionState, CompactionConfig +from .compaction_models import CompactionState, CompactionConfig, CompactionResult +from .compaction_policy import CompactionPolicy from .turn_based_session_manager import TurnBasedSessionManager from .preview_session_manager import PreviewSessionManager, is_preview_session @@ -8,6 +9,8 @@ "SessionFactory", "CompactionState", "CompactionConfig", + "CompactionResult", + "CompactionPolicy", "TurnBasedSessionManager", "PreviewSessionManager", "is_preview_session", diff --git a/backend/src/agents/main_agent/session/compaction_models.py b/backend/src/agents/main_agent/session/compaction_models.py index 3758094f3..733106de5 100644 --- a/backend/src/agents/main_agent/session/compaction_models.py +++ b/backend/src/agents/main_agent/session/compaction_models.py @@ -3,6 +3,9 @@ These models define the state and configuration for automatic context window compaction, which helps manage token usage in long conversations. + +Thresholds are model-relative — see ``compaction_policy.py`` and +docs/specs/compaction-model-relative-thresholds.md. """ from dataclasses import dataclass @@ -21,7 +24,7 @@ class CompactionState: a separate DynamoDB item. This simplifies storage and ensures atomic updates with session data. """ - checkpoint: int = 0 # Message index to load from (0 = load all) + checkpoint: int = 0 # Absolute message index to load from (0 = load all) summary: Optional[str] = None # Pre-computed summary for skipped messages last_input_tokens: int = 0 # Input tokens from last turn updated_at: Optional[str] = None # ISO timestamp of last update @@ -37,6 +40,14 @@ class CompactionState: # turns (the re-write is free then). It must never be derived from a # per-restore sliding window. truncation_anchor: int = 0 + # Hysteresis (spec §3.3). A cut disarms the trigger; a turn at or below + # the ceiling re-arms it. While disarmed, only the hard ceiling can force + # another cut. Legacy rows predate the field and default to armed. + armed: bool = True + # Snapshot of the policy the last cut was made under (window, ceiling, + # floor, hard ceiling, whether it was forced) — what the admin session + # profile reads to explain a compaction after the fact. + policy: Optional[Dict[str, Any]] = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for DynamoDB storage.""" @@ -47,6 +58,8 @@ def to_dict(self) -> Dict[str, Any]: "updatedAt": self.updated_at, "totalSummarizedTurns": self.total_summarized_turns, "truncationAnchor": self.truncation_anchor, + "armed": self.armed, + "policy": self.policy, } @classmethod @@ -55,6 +68,8 @@ def from_dict(cls, data: Optional[Dict[str, Any]]) -> "CompactionState": if not data: return cls() checkpoint = int(data.get("checkpoint", 0)) + armed = data.get("armed", True) + policy = data.get("policy") return cls( checkpoint=checkpoint, summary=data.get("summary"), @@ -65,6 +80,8 @@ def from_dict(cls, data: Optional[Dict[str, Any]]) -> "CompactionState": # so nothing retained by the slice is truncated (byte-stable from # the first restore under the anchor design). truncation_anchor=int(data.get("truncationAnchor", checkpoint)), + armed=bool(armed) if armed is not None else True, + policy=dict(policy) if isinstance(policy, dict) else None, ) @@ -72,9 +89,10 @@ def from_dict(cls, data: Optional[Dict[str, Any]]) -> "CompactionState": class CompactionResult: """ Returned by ``TurnBasedSessionManager.update_after_turn`` when a turn - crosses the token threshold and the checkpoint advances. Carries the + crosses the ceiling and the checkpoint advances. Carries the information the frontend needs to render an inline "earlier messages - summarized" divider in the conversation. + summarized" divider in the conversation, plus the policy the cut was + made under (additive fields on the ``compaction`` SSE payload). ``summarized_turns`` is the *delta* count of turns rolled into the summary at this compaction event (not the cumulative total across @@ -84,6 +102,19 @@ class CompactionResult: new_checkpoint: int summarized_turns: int input_tokens: int + context_window: Optional[int] = None + ceiling: Optional[int] = None + floor: Optional[int] = None + hard_ceiling: Optional[int] = None + # True when the cut ran while disarmed because input reached the hard + # ceiling — the signal that the previous cut did not take. + forced: bool = False + retained_tokens_estimate: Optional[int] = None + + +def _env_flag_default_on(name: str) -> bool: + """House-style kill switch: unset/empty → on; only the literal "false" is off.""" + return os.environ.get(name, "").strip().lower() != "false" @dataclass @@ -94,7 +125,9 @@ class CompactionConfig: Can be loaded from environment variables or passed directly. """ enabled: bool = True - token_threshold: int = 100_000 # Trigger checkpoint when exceeded + # Ceiling used when the model's window is unknown (and the fixed + # threshold when model-relative policy is switched off). + token_threshold: int = 100_000 protected_turns: int = 3 # Recent turns to protect from truncation max_tool_content_length: int = 500 # Max chars before truncating tool output # Bedrock prompt-cache TTL. When more than this many seconds have passed @@ -102,6 +135,13 @@ class CompactionConfig: # truncations can be applied without forcing an otherwise-avoidable # prefix re-write. cache_ttl_seconds: int = 300 + # Model-relative policy (spec §3.1). See CompactionPolicy.resolve. + model_relative_enabled: bool = Defaults.COMPACTION_MODEL_RELATIVE_ENABLED + ceiling_ratio: float = Defaults.COMPACTION_CEILING_RATIO + ceiling_cap_tokens: int = Defaults.COMPACTION_CEILING_CAP_TOKENS + floor_ratio: float = Defaults.COMPACTION_FLOOR_RATIO + hard_ceiling_ratio: float = Defaults.COMPACTION_HARD_CEILING_RATIO + hard_ceiling_multiplier: float = Defaults.COMPACTION_HARD_CEILING_MULTIPLIER @classmethod def from_env(cls) -> "CompactionConfig": @@ -112,4 +152,10 @@ def from_env(cls) -> "CompactionConfig": protected_turns=int(os.environ.get(EnvVars.COMPACTION_PROTECTED_TURNS, str(Defaults.COMPACTION_PROTECTED_TURNS))), max_tool_content_length=int(os.environ.get(EnvVars.COMPACTION_MAX_TOOL_CONTENT_LENGTH, str(Defaults.COMPACTION_MAX_TOOL_CONTENT_LENGTH))), cache_ttl_seconds=int(os.environ.get(EnvVars.COMPACTION_CACHE_TTL_SECONDS, str(Defaults.COMPACTION_CACHE_TTL_SECONDS))), + model_relative_enabled=_env_flag_default_on(EnvVars.COMPACTION_MODEL_RELATIVE_ENABLED), + ceiling_ratio=float(os.environ.get(EnvVars.COMPACTION_CEILING_RATIO, str(Defaults.COMPACTION_CEILING_RATIO))), + ceiling_cap_tokens=int(os.environ.get(EnvVars.COMPACTION_CEILING_CAP_TOKENS, str(Defaults.COMPACTION_CEILING_CAP_TOKENS))), + floor_ratio=float(os.environ.get(EnvVars.COMPACTION_FLOOR_RATIO, str(Defaults.COMPACTION_FLOOR_RATIO))), + hard_ceiling_ratio=float(os.environ.get(EnvVars.COMPACTION_HARD_CEILING_RATIO, str(Defaults.COMPACTION_HARD_CEILING_RATIO))), + hard_ceiling_multiplier=float(os.environ.get(EnvVars.COMPACTION_HARD_CEILING_MULTIPLIER, str(Defaults.COMPACTION_HARD_CEILING_MULTIPLIER))), ) diff --git a/backend/src/agents/main_agent/session/compaction_policy.py b/backend/src/agents/main_agent/session/compaction_policy.py new file mode 100644 index 000000000..90567ec47 --- /dev/null +++ b/backend/src/agents/main_agent/session/compaction_policy.py @@ -0,0 +1,256 @@ +""" +Model-relative compaction policy. + +Spec: docs/specs/compaction-model-relative-thresholds.md. + +Three numbers per model, derived from the catalog's ``maxInputTokens``: + +- **ceiling** — the trigger. ``min(window * CEILING_RATIO, CEILING_CAP_TOKENS)``. + Capped on purpose: a 1M window does not mean a 500k chat prefix is a good + idea — every warm turn re-reads it and every cache bust re-writes it. +- **floor** — the target size after a cut. ``ceiling * FLOOR_RATIO``. Under + Bedrock caching a cut costs one re-write of what *survives*, so a small floor + (deeper, rarer cuts) beats a shallow one (frequent re-writes). +- **hard ceiling** — force a cut even when hysteresis says wait. + ``min(window * HARD_CEILING_RATIO, ceiling * HARD_CEILING_MULTIPLIER)``. + +``COMPACTION_TOKEN_THRESHOLD`` (the historical fixed 100k) is the ceiling when +the window is unknown. With ``COMPACTION_MODEL_RELATIVE_ENABLED=false`` the +policy degrades to exactly the legacy behavior: fixed threshold, the +``cutoffs[-protected_turns]`` turn-count cut, no hysteresis. + +The module also owns the token-aware cut selection (``choose_checkpoint``) and +the per-message estimator it uses. Estimates are calibrated against the turn's +real history token count, so the *ratio* between messages is what the +estimator has to get right, not the absolute number. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Sequence, Tuple, TYPE_CHECKING + +if TYPE_CHECKING: + from .compaction_models import CompactionConfig + +logger = logging.getLogger(__name__) + +# chars/4 is the same heuristic Strands falls back to when native CountTokens +# is unavailable; good enough because the estimates are rescaled to the +# measured history size before use. +CHARS_PER_TOKEN = 4 +# Bedrock/Anthropic image tokens ~ (w*h)/750; dimensions are unknown here, so +# a flat figure that is roughly a 1000x1100 image. +IMAGE_TOKEN_ESTIMATE = 1_500 +# Per-message framing overhead (role, block boundaries). +MESSAGE_OVERHEAD_TOKENS = 4 + + +@dataclass(frozen=True) +class CompactionPolicy: + """Resolved thresholds for one turn on one model.""" + + ceiling: int + floor: Optional[int] + hard_ceiling: Optional[int] + context_window: Optional[int] + # "model_relative" (window known), "fixed" (window unknown, using the + # configured threshold), or "legacy" (kill switch off — turn-count cut). + source: str + + @property + def hysteresis_enabled(self) -> bool: + return self.floor is not None + + def to_dict(self) -> Dict[str, Any]: + return { + "ceiling": self.ceiling, + "floor": self.floor, + "hardCeiling": self.hard_ceiling, + "contextWindow": self.context_window, + "source": self.source, + } + + @classmethod + def resolve( + cls, + config: "CompactionConfig", + context_window: Optional[int], + ) -> "CompactionPolicy": + """Derive the policy for a model window under ``config``. + + ``context_window`` is the catalog's ``maxInputTokens`` for the model + that served the turn, or ``None`` when the lookup missed. + """ + window: Optional[int] = None + if context_window is not None: + try: + window = int(context_window) + except (TypeError, ValueError): + window = None + if window is not None and window <= 0: + window = None + + if not config.model_relative_enabled: + return cls( + ceiling=max(1, int(config.token_threshold)), + floor=None, + hard_ceiling=None, + context_window=window, + source="legacy", + ) + + if window is not None: + ceiling = int(min(window * config.ceiling_ratio, config.ceiling_cap_tokens)) + hard = int(min(window * config.hard_ceiling_ratio, ceiling * config.hard_ceiling_multiplier)) + source = "model_relative" + else: + ceiling = int(config.token_threshold) + hard = int(ceiling * config.hard_ceiling_multiplier) + source = "fixed" + + ceiling = max(1, ceiling) + # The floor must leave hysteresis room below the ceiling. + floor = max(0, min(ceiling - 1, int(ceiling * config.floor_ratio))) + hard = max(ceiling, hard) + return cls( + ceiling=ceiling, + floor=floor, + hard_ceiling=hard, + context_window=window, + source=source, + ) + + +# --------------------------------------------------------------------------- +# Per-message token estimate +# --------------------------------------------------------------------------- + +def _text_tokens(text: Any) -> int: + if not isinstance(text, str): + return 0 + return len(text) // CHARS_PER_TOKEN + + +def _json_tokens(value: Any) -> int: + try: + return len(json.dumps(value, ensure_ascii=False, default=str)) // CHARS_PER_TOKEN + except Exception: # noqa: BLE001 - estimator must never raise + return 0 + + +def _bytes_tokens(source: Any) -> int: + if not isinstance(source, dict): + return 0 + raw = source.get("bytes") + if isinstance(raw, (bytes, bytearray)): + return len(raw) // CHARS_PER_TOKEN + return 0 + + +def _block_tokens(block: Any) -> int: + if not isinstance(block, dict): + return _text_tokens(block) if isinstance(block, str) else 0 + if "text" in block: + return _text_tokens(block.get("text")) + if "image" in block: + return IMAGE_TOKEN_ESTIMATE + if "document" in block: + doc = block.get("document") or {} + return _bytes_tokens(doc.get("source")) if isinstance(doc, dict) else 0 + if "toolUse" in block: + tool_use = block.get("toolUse") or {} + return _json_tokens(tool_use.get("input")) + 8 + if "toolResult" in block: + result = block.get("toolResult") or {} + content = result.get("content") if isinstance(result, dict) else None + if isinstance(content, list): + return sum(_block_tokens(inner) for inner in content) + 8 + return _json_tokens(content) + 8 + if "json" in block: + return _json_tokens(block.get("json")) + if "reasoningContent" in block: + rc = block.get("reasoningContent") or {} + text = rc.get("reasoningText", {}) if isinstance(rc, dict) else {} + return _text_tokens(text.get("text") if isinstance(text, dict) else None) + if "cachePoint" in block: + return 0 + return _json_tokens(block) + + +def estimate_message_tokens(message: Dict[str, Any]) -> int: + """Heuristic token estimate for one Converse message (never raises).""" + if not isinstance(message, dict): + return 0 + content = message.get("content") + if isinstance(content, str): + return _text_tokens(content) + MESSAGE_OVERHEAD_TOKENS + if not isinstance(content, list): + return MESSAGE_OVERHEAD_TOKENS + return sum(_block_tokens(block) for block in content) + MESSAGE_OVERHEAD_TOKENS + + +# --------------------------------------------------------------------------- +# Cut selection +# --------------------------------------------------------------------------- + +def choose_checkpoint( + messages: Sequence[Dict[str, Any]], + cutoffs: Sequence[int], + protected_turns: int, + floor_tokens: int, + history_tokens: Optional[int], +) -> Tuple[int, Optional[int]]: + """Pick the cut (index into ``messages``) that lands retained history at or + below ``floor_tokens`` while keeping as much as fits. + + Returns ``(relative_cut, retained_estimate)``. ``relative_cut == 0`` means + nothing to cut. Candidates are ``cutoffs`` (tool-pair-safe turn starts) no + newer than ``cutoffs[-protected_turns]``; the last ``protected_turns`` + turns are always kept. Among candidates that satisfy the floor the + **oldest** wins. If none does — the protected tail alone exceeds the floor + — the minimum-protection cut is returned and the caller logs it (evicting + inside the tail is the offload escalation, a later PR, not a deeper cut). + + ``history_tokens`` is the measured size of the conversation portion of + the prompt this turn; per-message estimates are rescaled to sum to it so + only their ratios matter. ``None`` leaves the raw estimates as-is. + """ + protected_turns = max(0, int(protected_turns)) + if not cutoffs or len(cutoffs) <= protected_turns: + return 0, None + + newest_allowed = cutoffs[-protected_turns] if protected_turns > 0 else cutoffs[-1] + n = len(messages) + if n == 0: + return int(newest_allowed), None + + estimates = [estimate_message_tokens(m) for m in messages] + raw_total = sum(estimates) + scale = 1.0 + if history_tokens is not None and history_tokens > 0 and raw_total > 0: + scale = history_tokens / raw_total + + # suffix[i] = estimated tokens retained if we cut at i (keep messages[i:]) + suffix = [0] * (n + 1) + for i in range(n - 1, -1, -1): + suffix[i] = suffix[i + 1] + estimates[i] + + def retained(cut: int) -> int: + idx = min(max(int(cut), 0), n) + return int(round(suffix[idx] * scale)) + + candidates = [c for c in cutoffs if c <= newest_allowed] + fitting = [c for c in candidates if retained(c) <= floor_tokens] + if fitting: + cut = min(fitting) + return int(cut), retained(cut) + + logger.info( + "compaction_floor_unreachable: protected tail alone is ~%d tokens " + "(floor=%d); taking the minimum-protection cut at %d", + retained(newest_allowed), floor_tokens, newest_allowed, + ) + return int(newest_allowed), retained(newest_allowed) 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 b889a0630..9a9244121 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 @@ -10,7 +10,11 @@ - Stage 1: Tool content truncation — applied only below the persisted truncation anchor, which moves at checkpoint advances or when the Bedrock prompt cache has already expired between turns -- Stage 2: Checkpoint + Summary — triggered when token threshold exceeded +- Stage 2: Checkpoint + Summary — triggered when the turn's context exceeds + the model-relative ceiling (compaction_policy.py; the cut lands retained + history at the policy floor, and a cut disarms the trigger until the + context drops back under the ceiling — spec: + docs/specs/compaction-model-relative-thresholds.md) Byte-stability contract: between compaction-state changes, restoring the same stored history must produce byte-identical ``agent.messages``. Bedrock prompt @@ -37,6 +41,7 @@ from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig from .compaction_models import CompactionState, CompactionConfig, CompactionResult +from .compaction_policy import CompactionPolicy, choose_checkpoint if TYPE_CHECKING: from strands.agent.agent import Agent @@ -103,6 +108,12 @@ def __init__( self._valid_cutoff_indices: List[int] = [] self._all_messages_for_summary: List[Dict] = [] 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 + # checkpoint is always ``_live_offset + ``, + # so cuts computed over the live list and the slice applied at restore + # share one coordinate system (spec §3.4). + self._live_offset: int = 0 # Session control self.cancelled = False @@ -261,6 +272,7 @@ def initialize(self, agent: "Agent", **kwargs: Any) -> None: self.compaction_state = CompactionState() self._valid_cutoff_indices = [] self._all_messages_for_summary = [] + self._live_offset = 0 # Repair tool-use/tool-result pairing and role alternation on the FINAL # restored list — after compaction slicing/truncation — so it is always @@ -334,6 +346,9 @@ def _apply_compaction(self, agent: "Agent") -> None: else: messages_to_process = all_messages + # The live list now starts at this absolute index. + self._live_offset = offset + # Truncate only messages strictly below the anchor (absolute index), # translated into post-slice coordinates via the checkpoint offset. truncation_count = 0 @@ -407,6 +422,25 @@ def _cache_window_expired(updated_at: Optional[str], ttl_seconds: int) -> bool: last = last.replace(tzinfo=timezone.utc) return (datetime.now(timezone.utc) - last).total_seconds() > ttl_seconds + def _record_ledger_event(self, kind: str, **fields: Any) -> None: + """Hand a compaction decision to the per-call compaction ledger, if present. + + The cost-diagnostics ledger (``record_compaction_event`` / + ``drain_compaction_events`` + ``ContextLedgerHook``) lands whatever is + recorded here on the NEXT model call's ``C#`` cost row as + ``compactionEvents``, next to ``windowRemovedMessages`` and the prefix + token split — the evidence the summary cap and the scheduling rule are + judged on. Resolved by attribute so this is a no-op on a build without + the ledger; fields are ints. Never raises. + """ + recorder = getattr(self, "record_compaction_event", None) + if not callable(recorder): + return + try: + recorder(kind, **{k: int(v) for k, v in fields.items() if isinstance(v, (int, float)) and not isinstance(v, bool)}) + except Exception as e: # noqa: BLE001 + logger.debug("compaction ledger event skipped: %s", e) + # ========================================================================= # Compaction State Persistence # ========================================================================= @@ -731,21 +765,35 @@ async def update_after_turn( self, input_tokens: int, current_messages: Optional[List[Dict]] = None, + context_window: Optional[int] = None, + history_tokens: Optional[int] = None, ) -> Optional[CompactionResult]: """ Update compaction state after a turn completes. - Called by StreamCoordinator with input token count from model response. - Triggers checkpoint creation when token threshold exceeded. + Called by StreamCoordinator with the turn's cache-inclusive input token + count. Resolves the model-relative policy (ceiling / floor / hard + ceiling — spec §3.1), applies the hysteresis rule (§3.3) and, when a + cut is due, chooses a floor-seeking checkpoint (§3.2) in absolute + coordinates (§3.4). Persists the new checkpoint + summary; the slice + itself is applied at the next restore by ``_apply_compaction``. Returns a ``CompactionResult`` when the checkpoint advances on this turn so the caller can emit a ``compaction`` SSE event; otherwise returns ``None``. - ``current_messages`` is the agent's live message list. When provided, - the cutoff cache is re-derived from it so compaction works even when - AgentCoreMemory loads messages via hooks (skipping the initialize-time - prime path). + Args: + input_tokens: cache-inclusive input tokens of the turn's last call. + current_messages: the agent's live message list. When provided, + the cutoff cache is re-derived from it so compaction works even + when AgentCoreMemory loads messages via hooks (skipping the + initialize-time prime path). + context_window: the model's ``maxInputTokens`` from the catalog, + or ``None`` when unknown (falls back to the fixed threshold). + history_tokens: measured size of the conversation portion of the + prompt (the ``messages`` partition of the context breakdown); + calibrates the per-message estimates. ``None`` → use + ``input_tokens``, which biases the cut slightly deeper. """ if not self.compaction_config or not self.compaction_config.enabled: return None @@ -753,16 +801,49 @@ async def update_after_turn( # Re-read persisted state before touching it — on EVERY turn, not just # the first. See ``_adopt_persisted_compaction_state`` (#751). self._adopt_persisted_compaction_state() + state = self.compaction_state + state.last_input_tokens = input_tokens - self.compaction_state.last_input_tokens = input_tokens + policy = CompactionPolicy.resolve(self.compaction_config, context_window) - if input_tokens <= self.compaction_config.token_threshold: - self._save_compaction_state(self.compaction_state) + if input_tokens <= policy.ceiling: + if policy.hysteresis_enabled and not state.armed: + logger.info( + "compaction_rearmed: input=%d <= ceiling=%d (source=%s)", + input_tokens, policy.ceiling, policy.source, + ) + state.armed = True + self._save_compaction_state(state) + return None + + at_hard_ceiling = policy.hard_ceiling is not None and input_tokens >= policy.hard_ceiling + # "Forced" is the spiral signal: a cut that ran while DISARMED because + # the hard ceiling was reached. An armed cut above the hard ceiling is + # just a large ordinary cut. + forced = policy.hysteresis_enabled and not state.armed and at_hard_ceiling + if policy.hysteresis_enabled and not state.armed and not at_hard_ceiling: + # The previous cut has not been observed to take effect yet (the + # slice lands at the next restore) — cutting again now is exactly + # the spiral. Wait for the context to drop under the ceiling, or + # for the hard ceiling. + logger.info( + "compaction_disarmed_noop: input=%d > ceiling=%d, hard=%d (source=%s)", + input_tokens, policy.ceiling, policy.hard_ceiling, policy.source, + ) + self._save_compaction_state(state) return None + if forced: + logger.warning( + "compaction_forced: input=%d >= hard_ceiling=%d while disarmed — " + "the previous cut did not bring the context under the ceiling " + "(summary too large, or slice not yet applied on this agent)", + input_tokens, policy.hard_ceiling, + ) logger.info( f"Threshold exceeded: {input_tokens:,} > " - f"{self.compaction_config.token_threshold:,}" + f"{policy.ceiling:,} (floor={policy.floor}, hard={policy.hard_ceiling}, " + f"window={policy.context_window}, source={policy.source})" ) # Refresh cutoff cache from the agent's current messages — at this @@ -780,10 +861,11 @@ async def update_after_turn( if not self._valid_cutoff_indices: logger.info("No valid cutoff points cached, skipping checkpoint update") - self._save_compaction_state(self.compaction_state) + self._save_compaction_state(state) return None - total_turns = len(self._valid_cutoff_indices) + cutoffs = self._valid_cutoff_indices + total_turns = len(cutoffs) protected_turns = self.compaction_config.protected_turns if total_turns <= protected_turns: @@ -791,23 +873,66 @@ async def update_after_turn( f"Only {total_turns} turns available (need > {protected_turns}), " f"keeping all messages" ) - self._save_compaction_state(self.compaction_state) + self._save_compaction_state(state) + return None + + # Choose the cut in LIVE-LIST coordinates, then translate to absolute. + messages = self._all_messages_for_summary + retained_estimate: Optional[int] = None + if policy.floor is None: + # Legacy (kill switch): keep the last N turns, whatever their size. + relative_cut = cutoffs[-protected_turns] + else: + relative_cut, retained_estimate = choose_checkpoint( + messages, + cutoffs, + protected_turns, + policy.floor, + history_tokens if history_tokens is not None else input_tokens, + ) + + if relative_cut <= 0: + # Everything already fits under the floor — the excess is system + # prompt / tools, which a history cut cannot fix. + logger.info( + "compaction_nothing_to_cut: retained≈%s <= floor=%s with input=%d", + retained_estimate, policy.floor, input_tokens, + ) + self._save_compaction_state(state) return None - new_checkpoint = self._valid_cutoff_indices[-protected_turns] - current_checkpoint = self.compaction_state.checkpoint + new_checkpoint = self._live_offset + relative_cut + current_checkpoint = state.checkpoint if new_checkpoint <= current_checkpoint: - self._save_compaction_state(self.compaction_state) + self._save_compaction_state(state) return None - logger.info(f"Checkpoint update: {current_checkpoint} -> {new_checkpoint}") + logger.info( + "compaction_cut: checkpoint %d -> %d (live_offset=%d, relative_cut=%d, " + "retained≈%s, floor=%s, ceiling=%d, hard=%s, window=%s, forced=%s)", + current_checkpoint, new_checkpoint, self._live_offset, relative_cut, + retained_estimate, policy.floor, policy.ceiling, policy.hard_ceiling, + policy.context_window, forced, + ) + # Per-call compaction ledger: the two decisions the spec asks to see + # on the anatomy — a cut that ran while disarmed, and a cut that could + # not reach the floor because the protected tail alone exceeds it. + if forced: + self._record_ledger_event("forced", checkpoint=new_checkpoint, inputTokens=input_tokens) + if policy.floor is not None and retained_estimate is not None and retained_estimate > policy.floor: + self._record_ledger_event( + "floor_unreachable", + checkpoint=new_checkpoint, inputTokens=input_tokens, retainedTokens=retained_estimate, + ) # Count turns rolled into the summary on THIS event (delta, not - # cumulative) — each inline divider stands on its own. + # cumulative) — each inline divider stands on its own. In absolute + # coordinates: turn starts at or after the previous checkpoint (they + # were retained by the last slice) and before the new one. summarized_turns = sum( - 1 for idx in self._valid_cutoff_indices - if current_checkpoint < idx <= new_checkpoint + 1 for idx in cutoffs + if current_checkpoint <= self._live_offset + idx < new_checkpoint ) # Retrieve or generate summary for compacted messages @@ -815,28 +940,34 @@ async def update_after_turn( if summaries: summary = "\n\n".join(summaries) else: - messages_to_summarize = self._all_messages_for_summary[:new_checkpoint] + messages_to_summarize = messages[:relative_cut] summary = self._generate_fallback_summary(messages_to_summarize) - self.compaction_state.checkpoint = new_checkpoint + state.checkpoint = new_checkpoint # The anchor rides the checkpoint: everything the slice retains stays # byte-identical until the next compaction-state change, so the single # mutation (slice + summary) is paid with exactly one cache re-write. - self.compaction_state.truncation_anchor = max( - self.compaction_state.truncation_anchor, new_checkpoint - ) - self.compaction_state.summary = summary + state.truncation_anchor = max(state.truncation_anchor, new_checkpoint) + state.summary = summary # Running total persisted alongside the rest of the compaction state # so a refresh can rehydrate the end-of-conversation summary indicator. - self.compaction_state.total_summarized_turns += summarized_turns + state.total_summarized_turns += summarized_turns + if policy.hysteresis_enabled: + state.armed = False + state.policy = { + **policy.to_dict(), + "forced": forced, + "inputTokens": input_tokens, + "retainedTokensEstimate": retained_estimate, + } # This save is the compaction event itself — count it. - self._save_compaction_state(self.compaction_state, record_event=True) + self._save_compaction_state(state, record_event=True) logger.info( f"Compaction checkpoint set: {new_checkpoint}, " f"summary_length={len(summary) if summary else 0}, " f"summarized_turns={summarized_turns}, " - f"total_summarized_turns={self.compaction_state.total_summarized_turns}" + f"total_summarized_turns={state.total_summarized_turns}" ) return CompactionResult( @@ -844,6 +975,12 @@ async def update_after_turn( new_checkpoint=new_checkpoint, summarized_turns=summarized_turns, input_tokens=input_tokens, + context_window=policy.context_window, + ceiling=policy.ceiling, + floor=policy.floor, + hard_ceiling=policy.hard_ceiling, + forced=forced, + retained_tokens_estimate=retained_estimate, ) # ========================================================================= diff --git a/backend/src/agents/main_agent/streaming/stream_coordinator.py b/backend/src/agents/main_agent/streaming/stream_coordinator.py index 477b9707d..1513531d5 100644 --- a/backend/src/agents/main_agent/streaming/stream_coordinator.py +++ b/backend/src/agents/main_agent/streaming/stream_coordinator.py @@ -771,9 +771,18 @@ async def stream_response( if total_input_tokens > 0: try: current_messages = getattr(agent, "messages", None) + # Model-relative policy inputs: the catalog + # window (same lookup the badge uses — the + # catalog is cached) and the measured size of + # the conversation portion of the prompt. + # See docs/specs/compaction-model-relative-thresholds.md. + turn_context_window = await self._resolve_context_window(main_agent_wrapper) + history_tokens = self._history_tokens_from_breakdown(agent) compaction_result = await session_manager.update_after_turn( total_input_tokens, current_messages=current_messages, + context_window=turn_context_window, + history_tokens=history_tokens, ) logger.info(f" Compaction state updated: {total_input_tokens:,} input tokens") if compaction_result is not None: @@ -783,6 +792,14 @@ async def stream_response( "newCheckpoint": compaction_result.new_checkpoint, "summarizedTurns": compaction_result.summarized_turns, "inputTokens": compaction_result.input_tokens, + # Additive policy fields (the SPA + # validator ignores unknown keys). + "contextWindow": compaction_result.context_window, + "ceiling": compaction_result.ceiling, + "floor": compaction_result.floor, + "hardCeiling": compaction_result.hard_ceiling, + "forced": compaction_result.forced, + "retainedTokensEstimate": compaction_result.retained_tokens_estimate, } yield f"event: compaction\ndata: {json.dumps(compaction_payload)}\n\n" except Exception as e: @@ -2565,6 +2582,46 @@ def _format_sse_event(self, event: Dict[str, Any]) -> str: logger.error(f"Failed to serialize event: {e}") return f"event: error\ndata: {json.dumps({'error': f'Serialization error: {str(e)}'})}\n\n" + @staticmethod + async def _resolve_context_window(main_agent_wrapper: Any) -> Optional[int]: + """The serving model's ``maxInputTokens`` from the catalog, or ``None``. + + Feeds the model-relative compaction policy. Best-effort: a miss means + the policy falls back to the fixed threshold, never an error. + """ + model_config = getattr(main_agent_wrapper, "model_config", None) + model_id = getattr(model_config, "model_id", None) + if not model_id: + return None + try: + from apis.shared.costs.pricing_config import get_model_by_model_id + + record = await get_model_by_model_id(model_id) + value = getattr(record, "max_input_tokens", None) if record is not None else None + return int(value) if value else None + except Exception as e: # noqa: BLE001 - never let a lookup break the turn + logger.debug(f"Skipping contextWindow lookup for compaction: {e}") + return None + + @staticmethod + def _history_tokens_from_breakdown(agent: Any) -> Optional[int]: + """The ``messages`` partition of this turn's context breakdown, or ``None``. + + Calibrates the compaction policy's per-message estimates against the + measured size of the conversation portion of the prompt. + """ + try: + from agents.main_agent.session.hooks.context_attribution import get_context_breakdown + + breakdown = get_context_breakdown(agent) + for partition in (breakdown or {}).get("partitions", []) or []: + if isinstance(partition, dict) and partition.get("key") == "messages": + tokens = partition.get("tokens") + return int(tokens) if tokens is not None else None + except Exception as e: # noqa: BLE001 + logger.debug(f"Skipping history-token calibration: {e}") + return None + def _log_cache_metrics(self, usage: Dict[str, Any], session_id: str) -> None: """ Log cache performance metrics for monitoring and optimization. diff --git a/backend/tests/agents/main_agent/core/test_conversation_window.py b/backend/tests/agents/main_agent/core/test_conversation_window.py new file mode 100644 index 000000000..246402a7d --- /dev/null +++ b/backend/tests/agents/main_agent/core/test_conversation_window.py @@ -0,0 +1,32 @@ +"""The chat agent must not run under Strands' default 40-message window. + +docs/specs/compaction-model-relative-thresholds.md §3.0: compaction owns +history size; the SDK window would slide the front of the list every turn +past 40 messages (a prefix re-write per turn) and move the coordinates the +compaction checkpoint is expressed in. +""" + +from strands.agent.conversation_manager import SlidingWindowConversationManager + +from agents.main_agent.config.constants import Defaults +from agents.main_agent.core.agent_factory import AgentFactory + + +def test_default_window_is_large_not_forty(monkeypatch): + monkeypatch.delenv("AGENTCORE_CONVERSATION_WINDOW_MESSAGES", raising=False) + cm = AgentFactory.build_conversation_manager() + assert isinstance(cm, SlidingWindowConversationManager) + assert cm.window_size == Defaults.CONVERSATION_WINDOW_MESSAGES + assert cm.window_size >= 1000 + # Overflow recovery keeps tool-result truncation on. + assert cm.should_truncate_results is True + + +def test_env_override_restores_sdk_default(monkeypatch): + monkeypatch.setenv("AGENTCORE_CONVERSATION_WINDOW_MESSAGES", "40") + assert AgentFactory.build_conversation_manager().window_size == 40 + + +def test_garbage_env_falls_back(monkeypatch): + monkeypatch.setenv("AGENTCORE_CONVERSATION_WINDOW_MESSAGES", "lots") + assert AgentFactory.build_conversation_manager().window_size == Defaults.CONVERSATION_WINDOW_MESSAGES diff --git a/backend/tests/agents/main_agent/session/test_compaction_models.py b/backend/tests/agents/main_agent/session/test_compaction_models.py index 898233625..e7cc1c3eb 100644 --- a/backend/tests/agents/main_agent/session/test_compaction_models.py +++ b/backend/tests/agents/main_agent/session/test_compaction_models.py @@ -50,6 +50,8 @@ def test_to_dict_has_camel_case_keys(self): "updatedAt", "totalSummarizedTurns", "truncationAnchor", + "armed", + "policy", } def test_to_dict_values_match(self): @@ -174,3 +176,24 @@ def test_from_env_defaults_when_no_vars(self, monkeypatch): assert config.token_threshold == 100_000 assert config.protected_turns == 3 assert config.max_tool_content_length == 500 + + +# --------------------------------------------------------------------------- +# Hysteresis fields (docs/specs/compaction-model-relative-thresholds.md §3.3) +# --------------------------------------------------------------------------- + +class TestCompactionStateArmed: + def test_default_is_armed_with_no_policy(self): + state = CompactionState() + assert state.armed is True + assert state.policy is None + + def test_legacy_record_without_armed_defaults_to_armed(self): + state = CompactionState.from_dict({"checkpoint": 3}) + assert state.armed is True + + def test_roundtrips_armed_and_policy(self): + state = CompactionState(armed=False, policy={"ceiling": 100, "floor": 25}) + again = CompactionState.from_dict(state.to_dict()) + assert again.armed is False + assert again.policy == {"ceiling": 100, "floor": 25} diff --git a/backend/tests/agents/main_agent/session/test_compaction_policy.py b/backend/tests/agents/main_agent/session/test_compaction_policy.py new file mode 100644 index 000000000..36f6a0190 --- /dev/null +++ b/backend/tests/agents/main_agent/session/test_compaction_policy.py @@ -0,0 +1,316 @@ +"""Model-relative compaction policy — docs/specs/compaction-model-relative-thresholds.md. + +Pins the §3.1 table, the kill switch, the floor-seeking cut (§3.2) and the +hysteresis rule (§3.3) as exercised through ``update_after_turn``. +""" + +from unittest.mock import MagicMock + +import pytest + +from agents.main_agent.session.compaction_models import CompactionConfig, CompactionState +from agents.main_agent.session.compaction_policy import ( + IMAGE_TOKEN_ESTIMATE, + CompactionPolicy, + choose_checkpoint, + estimate_message_tokens, +) + +from .conftest import make_assistant_message, make_conversation, make_user_message + + +# --------------------------------------------------------------------------- +# §3.1 — the table +# --------------------------------------------------------------------------- + +class TestPolicyResolution: + @pytest.mark.parametrize( + "window, ceiling, floor, hard", + [ + (128_000, 64_000, 16_000, 89_600), + (200_000, 100_000, 25_000, 140_000), + (256_000, 100_000, 25_000, 150_000), + (272_000, 100_000, 25_000, 150_000), + (1_000_000, 100_000, 25_000, 150_000), + ], + ) + def test_table_rows(self, window, ceiling, floor, hard): + policy = CompactionPolicy.resolve(CompactionConfig(), window) + assert (policy.ceiling, policy.floor, policy.hard_ceiling) == (ceiling, floor, hard) + assert policy.source == "model_relative" + assert policy.context_window == window + + def test_unknown_window_uses_fixed_threshold(self): + policy = CompactionPolicy.resolve(CompactionConfig(), None) + assert policy.source == "fixed" + assert policy.ceiling == 100_000 + assert policy.floor == 25_000 + assert policy.hard_ceiling == 150_000 + assert policy.hysteresis_enabled + + @pytest.mark.parametrize("bad", [0, -5, "abc"]) + def test_invalid_window_treated_as_unknown(self, bad): + assert CompactionPolicy.resolve(CompactionConfig(), bad).source == "fixed" + + def test_kill_switch_is_legacy(self): + policy = CompactionPolicy.resolve( + CompactionConfig(model_relative_enabled=False, token_threshold=1_000), 1_000_000 + ) + assert policy.source == "legacy" + assert policy.ceiling == 1_000 + assert policy.floor is None + assert policy.hard_ceiling is None + assert not policy.hysteresis_enabled + + def test_hard_never_below_ceiling_and_floor_always_below(self): + cfg = CompactionConfig(hard_ceiling_ratio=0.1, floor_ratio=2.0) + policy = CompactionPolicy.resolve(cfg, 200_000) + assert policy.hard_ceiling >= policy.ceiling + assert policy.floor < policy.ceiling + + def test_from_env_reads_policy_fields(self, monkeypatch): + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_MODEL_RELATIVE_ENABLED", "false") + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_CEILING_RATIO", "0.4") + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_CEILING_CAP_TOKENS", "150000") + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_FLOOR_RATIO", "0.2") + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_HARD_CEILING_RATIO", "0.6") + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_HARD_CEILING_MULTIPLIER", "1.2") + cfg = CompactionConfig.from_env() + assert cfg.model_relative_enabled is False + assert cfg.ceiling_ratio == 0.4 + assert cfg.ceiling_cap_tokens == 150_000 + assert cfg.floor_ratio == 0.2 + assert cfg.hard_ceiling_ratio == 0.6 + assert cfg.hard_ceiling_multiplier == 1.2 + + def test_kill_switch_default_on_and_only_literal_false_disables(self, monkeypatch): + monkeypatch.delenv("AGENTCORE_MEMORY_COMPACTION_MODEL_RELATIVE_ENABLED", raising=False) + assert CompactionConfig.from_env().model_relative_enabled is True + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_MODEL_RELATIVE_ENABLED", "") + assert CompactionConfig.from_env().model_relative_enabled is True + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_MODEL_RELATIVE_ENABLED", "FALSE") + assert CompactionConfig.from_env().model_relative_enabled is False + + +# --------------------------------------------------------------------------- +# Estimator +# --------------------------------------------------------------------------- + +class TestEstimator: + def test_text_scales_with_length(self): + short = estimate_message_tokens(make_user_message("a" * 40)) + long = estimate_message_tokens(make_user_message("a" * 4000)) + assert long > short * 10 + + def test_image_is_flat(self): + msg = {"role": "user", "content": [{"image": {"format": "png", "source": {"bytes": b"x" * 10}}}]} + assert estimate_message_tokens(msg) >= IMAGE_TOKEN_ESTIMATE + + def test_tool_result_counts_inner_text(self): + msg = { + "role": "user", + "content": [{"toolResult": {"toolUseId": "t1", "content": [{"text": "r" * 4000}]}}], + } + assert estimate_message_tokens(msg) >= 1000 + + @pytest.mark.parametrize("weird", [None, "plain", {"role": "user"}, {"role": "user", "content": "str"}, {"content": [None, 3]}]) + def test_never_raises(self, weird): + assert estimate_message_tokens(weird) >= 0 + + +# --------------------------------------------------------------------------- +# §3.2 — floor-seeking cut +# --------------------------------------------------------------------------- + +def _conversation_with_sizes(sizes): + """user/assistant pairs where each user message carries `size` chars.""" + messages = [] + for i, size in enumerate(sizes): + messages.append(make_user_message("u" * size)) + messages.append(make_assistant_message(f"a{i}")) + return messages + + +class TestChooseCheckpoint: + def test_keeps_minimum_protected_turns(self): + msgs = make_conversation(3) + assert choose_checkpoint(msgs, [0, 2, 4], protected_turns=3, floor_tokens=10, history_tokens=None) == (0, None) + + def test_picks_oldest_cut_that_fits_under_floor(self): + # 6 turns; the last three are tiny, turns 2-3 are big. Floor admits the + # tail plus turn 3 but not turn 2 — oldest fitting cut is turn 3 (index 4). + msgs = _conversation_with_sizes([4000, 4000, 4000, 40, 40, 40]) + cutoffs = [0, 2, 4, 6, 8, 10] + # raw: big user ~1000 tok + overhead, small ~10+overhead; calibrate to 3500 total + cut, retained = choose_checkpoint(msgs, cutoffs, 3, floor_tokens=1400, history_tokens=3500) + assert cut == 4 + assert retained is not None and retained <= 1400 + + def test_falls_back_to_minimum_protection_when_tail_too_big(self): + msgs = _conversation_with_sizes([40, 40, 40, 8000, 8000, 8000]) + cutoffs = [0, 2, 4, 6, 8, 10] + cut, retained = choose_checkpoint(msgs, cutoffs, 3, floor_tokens=100, history_tokens=6000) + assert cut == 6 # cutoffs[-3] + assert retained > 100 + + def test_no_cut_when_everything_fits(self): + msgs = _conversation_with_sizes([40] * 6) + cut, _ = choose_checkpoint(msgs, [0, 2, 4, 6, 8, 10], 3, floor_tokens=10_000, history_tokens=200) + assert cut == 0 + + def test_calibration_scales_estimates_to_history_tokens(self): + msgs = _conversation_with_sizes([400] * 6) # equal turns + cutoffs = [0, 2, 4, 6, 8, 10] + # History measured at 6000 tokens → each turn ~1000. Floor 2500 admits + # only the two newest turns... but protection keeps three → min-protection. + cut, retained = choose_checkpoint(msgs, cutoffs, 3, floor_tokens=2500, history_tokens=6000) + assert cut == 6 + # Floor 3500 admits exactly three turns → cut at 6 again (oldest fitting) + cut2, _ = choose_checkpoint(msgs, cutoffs, 3, floor_tokens=3500, history_tokens=6000) + assert cut2 == 6 + # Floor 4500 admits four turns → cut at 4 + cut3, _ = choose_checkpoint(msgs, cutoffs, 3, floor_tokens=4500, history_tokens=6000) + assert cut3 == 4 + + def test_empty_messages_uses_minimum_protection(self): + assert choose_checkpoint([], [0, 2, 4, 6], 3, 10, 100)[0] == 2 + + +# --------------------------------------------------------------------------- +# §3.3 / §3.4 — hysteresis and coordinates through update_after_turn +# --------------------------------------------------------------------------- + +def _armed_manager(make_session_manager, compaction_config, checkpoint=0, armed=True): + mgr = make_session_manager(compaction_config=compaction_config) + mgr.compaction_state = CompactionState(checkpoint=checkpoint, armed=armed) + mgr._save_compaction_state = MagicMock() + mgr._retrieve_session_summaries = MagicMock(return_value=[]) + mgr._valid_cutoff_indices = [0, 2, 4, 6, 8] + mgr._all_messages_for_summary = make_conversation(5) + return mgr + + +class TestHysteresis: + """compaction_config fixture: threshold 1000 → fixed policy ceiling 1000, + floor 250, hard 1500 (window unknown).""" + + @pytest.mark.asyncio + async def test_cut_disarms(self, make_session_manager, compaction_config): + mgr = _armed_manager(make_session_manager, compaction_config) + result = await mgr.update_after_turn(1200) + assert result is not None + assert mgr.compaction_state.armed is False + assert result.forced is False + assert result.ceiling == 1000 and result.floor == 250 and result.hard_ceiling == 1500 + assert mgr.compaction_state.policy["source"] == "fixed" + + @pytest.mark.asyncio + async def test_over_ceiling_while_disarmed_is_a_noop(self, make_session_manager, compaction_config): + mgr = _armed_manager(make_session_manager, compaction_config, checkpoint=4, armed=False) + mgr._valid_cutoff_indices = [0, 2, 4, 6, 8, 10] + mgr._all_messages_for_summary = make_conversation(6) + result = await mgr.update_after_turn(1200) + assert result is None + assert mgr.compaction_state.checkpoint == 4 + assert mgr.compaction_state.armed is False + mgr._save_compaction_state.assert_called_once() + + @pytest.mark.asyncio + async def test_hard_ceiling_forces_a_cut_while_disarmed(self, make_session_manager, compaction_config): + mgr = _armed_manager(make_session_manager, compaction_config, checkpoint=4, armed=False) + mgr._valid_cutoff_indices = [0, 2, 4, 6, 8, 10] + mgr._all_messages_for_summary = make_conversation(6) + result = await mgr.update_after_turn(1500) + assert result is not None and result.forced is True + assert mgr.compaction_state.checkpoint == 6 + assert mgr.compaction_state.policy["forced"] is True + + @pytest.mark.asyncio + async def test_armed_cut_above_hard_ceiling_is_not_forced(self, make_session_manager, compaction_config): + """Forced means "ran while disarmed" — the spiral signal — not "was large".""" + mgr = _armed_manager(make_session_manager, compaction_config) + result = await mgr.update_after_turn(2500) # above hard=1500, but armed + assert result is not None and result.forced is False + assert mgr.compaction_state.policy["forced"] is False + + @pytest.mark.asyncio + async def test_under_ceiling_rearms(self, make_session_manager, compaction_config): + mgr = _armed_manager(make_session_manager, compaction_config, checkpoint=4, armed=False) + assert await mgr.update_after_turn(900) is None + assert mgr.compaction_state.armed is True + + @pytest.mark.asyncio + async def test_spiral_shape_cuts_exactly_once(self, make_session_manager, compaction_config): + """Input pinned above the ceiling for 10 turns: one cut, then no-ops.""" + mgr = _armed_manager(make_session_manager, compaction_config) + advances = 0 + for turn in range(10): + n_turns = 5 + turn + mgr._valid_cutoff_indices = list(range(0, 2 * n_turns, 2)) + mgr._all_messages_for_summary = make_conversation(n_turns) + if await mgr.update_after_turn(1200) is not None: + advances += 1 + assert advances == 1 + + @pytest.mark.asyncio + async def test_legacy_mode_never_disarms_and_uses_turn_count(self, make_session_manager): + cfg = CompactionConfig(enabled=True, token_threshold=1000, protected_turns=3, model_relative_enabled=False) + mgr = _armed_manager(make_session_manager, cfg) + first = await mgr.update_after_turn(1200) + assert first is not None and first.floor is None + assert mgr.compaction_state.checkpoint == 4 # cutoffs[-3] + assert mgr.compaction_state.armed is True + mgr._valid_cutoff_indices = [0, 2, 4, 6, 8, 10] + mgr._all_messages_for_summary = make_conversation(6) + second = await mgr.update_after_turn(1200) + assert second is not None and mgr.compaction_state.checkpoint == 6 + + +class TestCoordinates: + @pytest.mark.asyncio + async def test_checkpoint_is_live_offset_plus_relative_cut(self, make_session_manager, compaction_config): + mgr = _armed_manager(make_session_manager, compaction_config, checkpoint=10) + mgr._live_offset = 10 # restore sliced at absolute 10 + result = await mgr.update_after_turn(1200) + assert result is not None + assert result.previous_checkpoint == 10 + assert result.new_checkpoint == 14 # 10 + cutoffs[-3]=4 + assert mgr.compaction_state.truncation_anchor == 14 + + @pytest.mark.asyncio + async def test_context_window_flows_into_policy(self, make_session_manager): + cfg = CompactionConfig(enabled=True, protected_turns=3) + mgr = _armed_manager(make_session_manager, cfg) + # 128k window → ceiling 64k; 60k is under it → no cut. + assert await mgr.update_after_turn(60_000, context_window=128_000) is None + assert mgr.compaction_state.checkpoint == 0 + # 1M window → ceiling capped at 100k (not 500k) → 150k cuts. + result = await mgr.update_after_turn(150_000, context_window=1_000_000) + assert result is not None and result.context_window == 1_000_000 and result.ceiling == 100_000 + assert result.hard_ceiling == 150_000 + + +class TestCompactionLedgerEvents: + """Decisions are handed to the per-call compaction ledger when it exists + (cost-diagnostics ``record_compaction_event``), and are a no-op otherwise.""" + + @pytest.mark.asyncio + async def test_forced_and_floor_unreachable_are_recorded(self, make_session_manager, compaction_config): + mgr = _armed_manager(make_session_manager, compaction_config, checkpoint=4, armed=False) + mgr._valid_cutoff_indices = [0, 2, 4, 6, 8, 10] + mgr._all_messages_for_summary = make_conversation(6) + mgr.record_compaction_event = MagicMock() + result = await mgr.update_after_turn(1500) # disarmed + hard ceiling → forced + assert result is not None and result.forced + kinds = [c.args[0] for c in mgr.record_compaction_event.call_args_list] + assert "forced" in kinds and "floor_unreachable" in kinds + forced_call = next(c for c in mgr.record_compaction_event.call_args_list if c.args[0] == "forced") + assert forced_call.kwargs == {"checkpoint": 6, "inputTokens": 1500} + floor_call = next(c for c in mgr.record_compaction_event.call_args_list if c.args[0] == "floor_unreachable") + assert floor_call.kwargs["retainedTokens"] > 250 + + @pytest.mark.asyncio + async def test_no_ledger_is_a_noop(self, make_session_manager, compaction_config): + mgr = _armed_manager(make_session_manager, compaction_config) + 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/streaming/test_compaction_sse_emit_once.py b/backend/tests/agents/main_agent/streaming/test_compaction_sse_emit_once.py index 32223b618..8eb0f28e7 100644 --- a/backend/tests/agents/main_agent/streaming/test_compaction_sse_emit_once.py +++ b/backend/tests/agents/main_agent/streaming/test_compaction_sse_emit_once.py @@ -95,7 +95,11 @@ async def update_after_turn( self, input_tokens: int, current_messages: Optional[List[Dict]] = None, + context_window: Optional[int] = None, + history_tokens: Optional[int] = None, ) -> Optional[CompactionResult]: + # Same seam as TurnBasedSessionManager.update_after_turn — the policy + # inputs (context_window / history_tokens) are accepted and ignored. self.calls.append(input_tokens) return self._result @@ -158,6 +162,14 @@ async def test_compaction_sse_emitted_exactly_once_when_checkpoint_advances(): "newCheckpoint": 4, "summarizedTurns": 2, "inputTokens": 150_000, + # Model-relative policy fields (additive; the stub result leaves + # them at their defaults). + "contextWindow": None, + "ceiling": None, + "floor": None, + "hardCeiling": None, + "forced": False, + "retainedTokensEstimate": None, } diff --git a/docs/one-pagers/cost-effectiveness-roadmap.md b/docs/one-pagers/cost-effectiveness-roadmap.md index c9a3450ce..f7a7127dc 100644 --- a/docs/one-pagers/cost-effectiveness-roadmap.md +++ b/docs/one-pagers/cost-effectiveness-roadmap.md @@ -7,7 +7,7 @@ workstream's authority lives in its own spec, and if this page disagrees with a spec, the spec wins and this page gets fixed. *Companion specs:* `compaction-over-threshold-cache-spiral.md` (#833) · -`agent-cache-extra-tools-bypass.md` (#834) · `compaction-v2-versioned-prefix.md` +`agent-cache-extra-tools-bypass.md` (#834) · `compaction-model-relative-thresholds.md` (2026-09-15) · `compaction-v2-versioned-prefix.md` (#835) · `document-context-offload.md` + validation + evaluation (#836) · `quota-cooldown-windows.md` · `tool-search-token-bloat-strategy.md` · `session-workspace-tools.md` · `share-large-conversations-s3-offload.md` · @@ -99,7 +99,7 @@ how the backlog below is ranked: | # | workstream | what it protects | authority | state | |---|---|---|---|---| | W1 | **Measurement** | every other row of this table | #833 PR-1 (`partial_miss`), cohort scan §4.1, fleet anatomy scan, dashboards #699/#700 | **PR-1 merged (#838) and live in dev**; prod awaits a release, and that is when the G0 clock starts. §4.1 cohort scan **run 2026-08-05**; fleet anatomy **run 2026-08-05** (`scan_fleet_prefix_spend.py`, reproducible) | -| W2 | **Prefix stability** | don't rewrite what didn't change — 55.2% of spend | #833 PR-2/3/4 → #835 v2 (gated) | #833 PR-2/3/4 unbuilt. ⚠️ **PR-4 re-ranked 2026-08-05**: the system prompt mutates mid-session in 12.3% of multi-turn conversations, making it the most *general* item here, not a footnote to D2/D3. ⚠️ **#834 has left this row** — G1 disproved its prefix-cost thesis; it is a latency fix and now lives in W6 | +| W2 | **Prefix stability** | don't rewrite what didn't change — 55.2% of spend | #833 PR-2/3/4 → `compaction-model-relative-thresholds.md` (PR-1 in progress 2026-09-15: per-model ceiling/floor/hard ceiling, floor-seeking cut, hysteresis; PR-3 = paid-when-free scheduling) → #835 v2 (gated) | #833 PR-2/3/4 unbuilt. ⚠️ **PR-4 re-ranked 2026-08-05**: the system prompt mutates mid-session in 12.3% of multi-turn conversations, making it the most *general* item here, not a footnote to D2/D3. ⚠️ **#834 has left this row** — G1 disproved its prefix-cost thesis; it is a latency fix and now lives in W6 | | W2b | **Short-conversation cache economics** (new 2026-08-05) | the 43% of spend in sessions of ≤15 calls, which no item in this arc touches | **un-specced — gap** | 700 sessions wrote cache and read none back ($33.92, 7.3% of all write spend). Single-call sessions are 29% of all sessions and spend 67% of their money on writes they can never use. Needs a cachePoint-policy spec for first/short turns | | W6 | **Turn latency** | time-to-answer, not tokens | #834 (bypass narrowing + family promotion) · #841 (runtime session affinity) | **#841 merged and verified in dev** — steady-state turns ~7.6s → ~3.9s. Split: warm container ~7.6→4.8s (all sessions), reused Agent ~4.8→3.9s (cacheable only). #839's treatment arm now equals the ceiling | | W3 | **Payload boundedness** | nothing unbounded enters the prefix | #836 offload · tool-search strategy · workspace tools (PR-1 built) · S3 share-offload | offload PRs unbuilt; citations baseline probe required first | diff --git a/docs/specs/compaction-model-relative-thresholds.md b/docs/specs/compaction-model-relative-thresholds.md new file mode 100644 index 000000000..31eb4d7f5 --- /dev/null +++ b/docs/specs/compaction-model-relative-thresholds.md @@ -0,0 +1,389 @@ +# Compaction relative to the model window — a trigger ceiling, a target floor, and paying the rewrite only when it is free + +**Status:** PR-1 in progress (this document rides with it). PR-2 through PR-5 +unbuilt. Written 2026-09-15. +**Owner:** Phil Merrell +**Related:** `compaction-over-threshold-cache-spiral.md` (#833 — the incident +and the summary-cap PR this spec depends on) · +`compaction-v2-versioned-prefix.md` (#835 — the frozen-segment redesign; this +spec implements v2's I3/I4 *policy* on the v1 machinery so it survives either +way) · `agent-cache-extra-tools-bypass.md` · `document-context-offload.md` · +`gpt-5-6-prompt-caching.md` (why `maxInputTokens` is a pricing cap) · +`docs/one-pagers/cost-effectiveness-roadmap.md` (W2 row) · the 2026-09-15 prod +cost audit (top-5 September users + 7 largest conversations, via +`/admin/costs`), whose measurements §2 quotes + +--- + +## 1. The question this answers + +Compaction fires at a fixed 100,000 tokens +(`AGENTCORE_MEMORY_COMPACTION_TOKEN_THRESHOLD`), regardless of whether the +model's window is 200k, 272k or 1M. Should the trigger vary with the window? + +Yes — but the trigger is the least important number. Under Bedrock prompt +caching the marginal costs are: + +| operation, at a 200k-token prefix (Sonnet 5, `global.*`, $2.00/MTok input) | cost | +|---|---| +| keep the history for one more turn (cache read, 0.1× input) | ~$0.04 | +| change the history once (cache write, 1.25× input) | ~$0.50 | + +Retention is ~12× cheaper than mutation *per event*. What compaction spends +is **prefix rewrites**, and what it saves is **the size of every read after**. +So the two numbers that matter are (a) how small the conversation is after a +compaction — the *floor* — and (b) whether the rewrite lands on a turn that +was going to rewrite anyway. The July replay in the 2026-07-27 measurement +made this concrete: compacting the ten biggest sessions at 120k down to 25k +cut input-side cost 64%; deeper-and-rarer beat shallower-and-more-often in +every trajectory. + +A model-relative threshold that merely *raises* the ceiling on 1M models +would therefore make things worse: every read is larger, every cache bust is +larger, and long-context quality degrades well before a 1M window fills +(`document-context-offload.md` §"context rot"). The design below scales the +trigger with the window but **caps** it, and puts the engineering weight on +the floor and the scheduling. + +## 2. Current state (verified 2026-09-15 on `develop` @ 222e1d26) + +- `CompactionConfig.token_threshold` = 100,000 (`constants.py` Defaults), + compared in `update_after_turn` against the turn's cache-inclusive input + (`inputTokens + cacheRead + cacheWrite`, the only correct context size under + caching — see the coordinator comment at the call site). +- The catalog already carries the window: `ManagedModel.max_input_tokens` + (`maxInputTokens`) — 200k on the Claude 4.x rows, 1,000,000 on Sonnet 5, + 272,000 on the GPT-5.6 family and GPT-6 Astra, 256,000 on Qwen. The stream + coordinator looks it up every turn for the badge (`final_metadata["contextWindow"]`) + and the storage path. **It is not a capability field on the OpenAI rows: it + is the short-context pricing cap** (`curated-models.ts:373`). Deriving + thresholds from it therefore keeps us inside short-context pricing for free. +- The checkpoint is chosen by *turn count*: `cutoffs[-protected_turns]`, i.e. + "keep the last 3 user turns". Turn count is uncorrelated with tokens — the + byte-stability audit found three cheap turns summarized while a 92k tool + result was kept. +- There is **no hysteresis.** Over threshold, the checkpoint advances on every + turn (each new turn pushes `cutoffs[-3]` forward), the LTM summary join is + re-fetched and re-persisted, and `Threshold exceeded` logs every turn — the + spiral session did this 56 times. +- `update_after_turn` receives the agent's live list, which on a restored + session is *already sliced* at the checkpoint, and compares a slice-relative + index against the persisted absolute checkpoint (spiral spec D3; 199 of + 1,238 rows carry the mismatch). +- **Strands' default 40-message sliding window runs underneath all of this.** + `AgentFactory.create_agent` passes no `conversation_manager`, so the SDK + installs `SlidingWindowConversationManager(window_size=40)` and applies it + after every event-loop cycle (`agent.py:1631`). Past 40 messages the front + of `agent.messages` slides every turn. The 2026-09-15 prod cost audit (20 + sessions, $127, content-free) measured the consequence: fingerprint + `messageCount` pinned at 39–41, every turn start reading only tools+system + and re-writing the whole window (session e7e75953 flips exactly at message + 41), and the checkpoint/anchor coordinate mismatch (`ANCHOR_MISMATCH`) on + 14 of 20 sessions — because the list the checkpoint indexes into is being + mutated by something other than compaction. +- The same audit's cache-write attribution for September (Sonnet 5 = $683 of + $742; cache writes ~54% of it): 36% full re-writes after a >5 min pause + (cost scales with the context size at the pause — 17 of 20 sessions + peaked above 100k), 22% live re-writes inside over-100k sessions (the + spiral, still live), 2.6% the hourly system-prompt tick + (`get_current_date_pacific()` renders `%H:00`, so every Pacific hour + boundary flips `systemPromptHash`), ~2% the 40-message window in sub-100k + sessions. Summaries of 23k–40k tokens were present in 6 of 20 sessions. +- Compaction changes bytes in two places only: the restore-time slice in + `_apply_compaction` (cold starts) and — nowhere on a warm agent. That is the + subject of PR-3, not PR-1. +- Strands 1.55 ships the native form of "trigger as a ratio of the window": + `SummarizingConversationManager(proactive_compression={"compression_threshold": r})` + reads `model.context_window_limit` (a `BedrockConfig` key we do not set). It + has no floor, no cache-aware scheduling and no summary budget, and the + 2026-05-18 decision bars it as a bare swap. Per v2 §4.2 it is the *engine* + we would move onto, not the *policy*. + +## 3. Design + +### 3.0 Prerequisite: one owner of history size + +Compaction cannot express a checkpoint in a list that something else is +trimming. PR-1 sets the conversation manager explicitly: +`SlidingWindowConversationManager(window_size=2000, should_truncate_results=True)` +(`AGENTCORE_CONVERSATION_WINDOW_MESSAGES`; `40` restores the SDK default). +The manager is kept rather than replaced with `NullConversationManager` +because its `reduce_context` is the stack's only +`ContextWindowOverflowException` recovery, and that path is independent of +the window size. Consequence to state plainly: conversations between 40 +messages and the ceiling now go to the model whole — more *read* tokens per +turn (0.1×), far fewer *re-writes* (1.25×), and the model sees the +conversation instead of its last 40 messages. Above the ceiling the +compaction policy bounds it. + +### 3.1 Three numbers per model, derived from `maxInputTokens` + +``` +ceiling = min(window × CEILING_RATIO, CEILING_CAP_TOKENS) # trigger +floor = ceiling × FLOOR_RATIO # target after a cut +hard_ceiling = min(window × HARD_CEILING_RATIO, ceiling × HARD_MULT) # force, even on a warm cache +``` + +Defaults: `CEILING_RATIO 0.5`, `CEILING_CAP_TOKENS 100_000`, `FLOOR_RATIO 0.25`, +`HARD_CEILING_RATIO 0.7`, `HARD_MULT 1.5`. Which gives: + +| `maxInputTokens` | ceiling | floor | hard ceiling | +|---|---|---|---| +| 128,000 (a small-window model, for illustration) | 64,000 | 16,000 | 89,600 | +| 200,000 (Claude 4.x, Haiku 4.5) | 100,000 | 25,000 | 140,000 | +| 256,000 (Qwen) | 100,000 | 25,000 | 150,000 | +| 272,000 (GPT-5.6 / GPT-6 pricing cap) | 100,000 | 25,000 | 150,000 | +| 1,000,000 (Sonnet 5) | 100,000 | 25,000 | 150,000 | +| unknown (catalog miss) | `token_threshold` (100,000) | 25,000 | 150,000 | + +So the window scales the ceiling **down** for small-window models and never +up: for every model in the catalog today the cap binds at 100k. That is the +honest answer to "shouldn't the threshold vary with the window" — under +cache economics, no, not upward. Why: + +- **The cap was drafted at 200k for 1M models and moved to 100k on + evidence.** The 2026-09-15 replay of the 20 audited Sonnet 5 sessions + (input side, this spec's PR-3 scheduling rule applied, priced at the + incident's $2.50/$0.20 per MTok) gave: actual today **$102.64**, no + compaction **$208.81**, 100k/25k **$73.70**, 200k/50k **$105.30**. The 200k + policy was worse on 13 of 20 sessions and never better, and roughly equal + to today — raising the ceiling gives back the whole PR-1 win on that cohort. + Mechanism: 36% of cache-write dollars are cold re-writes after a >5 min + pause, and their size is the context *at the pause*; under 200k/50k most + heavy sessions never reach the ceiling and run 100–190k the whole time, so + each return costs ~$0.375 instead of $0.08–0.12. Caveats: 20 sessions, all + Sonnet 5, all heavy; the "actual" column already benefits from the + 40-message window that PR-1 removes, so "no compaction" is the baseline + PR-1 replaces. Raise the cap only when `compaction_forced` and the cost + anatomy show sessions that need more room; it is one constant. +- **Quality is on the same side.** Every warm turn reads the whole prefix; + the offload spec's evidence on context rot says the model is not better at + 400k of chat history than at 100k plus a good summary. +- **The floor at a quarter of the ceiling** is the "deeper and rarer" result: + a session that compacts to 25k and grows back to 100k pays one rewrite of + ~25k and then ~75k tokens' worth of *reads* before the next cut. A session + that compacts 100k → 80k pays a rewrite of 80k every few turns. +- **The hard ceiling** exists so that PR-3's "wait for a free turn" cannot + wait forever. 70% of the window leaves room for the turn's own output and + for one oversized tool result without an overflow. On a 200k window the 70% + term (140k) binds; on larger windows the 1.5× term (150k) does. +- **These are starting points, not tuned constants.** §5 says how to move + them. Every one is an env override (`AGENTCORE_MEMORY_COMPACTION_*`), and + `AGENTCORE_MEMORY_COMPACTION_MODEL_RELATIVE_ENABLED=false` reverts to the + fixed 100k threshold and the legacy turn-count cut exactly. + +### 3.2 A floor-seeking, token-aware checkpoint + +When the trigger fires, the cut is chosen to land the retained history **at +or below the floor**, keeping as much as fits: + +1. Estimate tokens per message from the message's serialized size (chars/4; + flat 1,500 for an inline image; document bytes are already stripped at + restore). Calibrate the estimates so they sum to the turn's *history* + tokens — the `messages` partition from the context-attribution breakdown + when the turn has one, otherwise the full input count. Over-attributing + system/tools tokens to history biases the cut slightly deeper, which is + the safe direction. +2. Candidate cuts are the same tool-pair-safe boundaries as today (user + messages that are not tool results), and never newer than + `cutoffs[-protected_turns]` — the last N turns are always kept, as today. +3. Choose the **oldest** candidate whose retained estimate is ≤ floor. If + even the minimum-protection cut exceeds the floor (a giant tool result + inside the protected tail), take the minimum-protection cut and log it; + evicting inside the protected tail is PR-4's escalation, not a deeper cut. + +### 3.3 Hysteresis: a cut disarms the trigger + +`CompactionState.armed` (persisted, legacy rows default `True`): + +- Over the ceiling **and armed** → cut, then `armed = False`. +- Over the ceiling **and disarmed** → do nothing unless input ≥ hard ceiling, + in which case cut anyway and log `compaction_forced`. A forced cut is the + signal that the previous cut did not take (the summary is too large, or the + slice has not landed on this agent yet) — it is a metric, not a code path + we expect to run. +- At or below the ceiling → `armed = True`. + +This makes v2's I3 ("threshold exceeded twice in a row is a bug by +definition") structural on the v1 code: the spiral's 56 consecutive cuts +become one cut plus 55 no-ops, and the LTM summary fetch stops being a +per-turn call. + +### 3.4 One coordinate system + +`update_after_turn` computes cuts over the agent's live list. The live list +starts at the absolute index the restore sliced at (or 0). The manager now +tracks that offset (`_live_offset`, set in `_apply_compaction`) and persists +`checkpoint = _live_offset + relative_cut`. The persisted checkpoint stays +absolute — the coordinate `_apply_compaction` slices with — and the D3 +comparison (`new <= current`) is finally like-for-like. The deeper D3 +question (the restore window itself moving — `original=74` frozen) is +untouched here and stays with spiral-spec PR-3. + +### 3.5 Pay the rewrite when it is free (PR-3) + +Everything above decides *what* to cut. PR-3 decides *when the bytes change*: + +- Post-turn computes and persists the pending checkpoint + summary (the + expensive part, off the critical path). Nothing is applied. +- Pre-call on the next turn, the pending cut is applied to the live list + **in place** (`agent.messages[:] = ...`, never rebound — the #741 alias) + when any of: the gap since the last call exceeds the Bedrock TTL + (`cacheGapSeconds` > 300 — the prefix was going to rewrite anyway), the + model or the `@`-mentioned agent changed (prefix already invalid), or the + turn's input is projected at or above the hard ceiling. +- Between the ceiling and the hard ceiling on a warm cache, the cut waits. + `rewrite_scheduled` vs `rewrite_forced` is logged per application so I4's + effectiveness is measurable. +- Aliasing the message list across agent instances must also alias the + offset; `_adopt_session_conversation` syncs `_live_offset` when it adopts. + +### 3.6 The rest of the sequence + +- **Bounded summary** (= spiral spec PR-2, unchanged): an 8k-token budget, + re-summarized once with the cheap model at cut time. Without it the floor + is unreachable — a 40k-token summary is larger than the 25k floor. PR-1's + `compaction_forced` metric will show exactly how often this bites until it + lands. +- **Content-class eviction + offload** (PR-4): when the protected tail alone + exceeds the floor, move the oversized tool result or document to the + session workspace behind the retrieval tool and reference it, rather than + cutting deeper or giving up. +- **`context_window_limit` on the model** (PR-3, small): plumb + `maxInputTokens` into `ModelConfig` and set `context_window_limit` in + `to_bedrock_config` (a valid `BedrockConfig` key in 1.55), so Strands' + own `estimate_utilization` and our policy agree on the window, and the + eventual v2 engine swap inherits it. +- **Per-section cache TTL** (PR-5, experiment): tools + system on `1h`, + messages on `5m`, via `CacheConfig(system_prompt_ttl="1h", tools_ttl="1h")`. + The 2026-07-27 model said a *blanket* 1h TTL was a wash (2× write premium + ate the saving) and a *selective* one was the variant worth testing. + Sequence behind the 1.55 cache-point invariant test; measure on + `cacheStatus` before and after. + +## 4. Cost model + +Per-turn input cost for a session sitting at `P` prefix tokens on a model with +base input rate `r`: + +``` +warm turn ≈ 0.10 r P (cache read) +cache-busted turn ≈ 1.25 r P (cache write) +compaction turn ≈ 1.25 r F + summarizer call (F = post-cut size) +``` + +For the July cohort (10 sessions, peaks 112k–597k, ~19 cache tokens written +per output token) the replay gave input-side cost **$90.35 → $32.40** when +cutting at 120k to 25k. The same replay with the ceiling at 200k and the floor +at 50k is the number to produce for the Sonnet 5 row before ratifying §3.1 — +`scan_fleet_prefix_spend.py` and the spiral spec's §4.2 harness already +replay real trajectories under a policy. + +## 5. Quality gate and tuning + +- **Veto before default change in prod:** the spiral spec §4.3 long-session + eval (constraint retention / revision continuity / reference lookup) runs + on PR-1 with the fixed-threshold arm as control. A deeper cut is a bigger + context change than the summary cap, so the veto applies with full force. +- **Tuning knobs move on evidence, not taste:** raise `FLOOR_RATIO` if the + eval shows retention loss; lower `CEILING_CAP_TOKENS` if the Sonnet 5 + cohort's write:read ratio stays worse than 1:5 after PR-3; never raise the + cap above 272k while GPT-family rows share the constants (pricing tier). +- **Summarizer prompt:** preserve standing user instructions and constraints + verbatim (kaizen 2026-05-29 item); this is PR-2's prompt, not PR-1's. + +## 6. PR breakdown + +### PR-1 — policy, floor-seeking cut, hysteresis, coordinates (this PR) + +- `AgentFactory.build_conversation_manager()`: the explicit 2000-message + window (§3.0) — the prerequisite for every coordinate claim below. +- `compaction_policy.py`: `CompactionPolicy.resolve(config, context_window)`, + `estimate_message_tokens`, `choose_checkpoint`. +- `CompactionConfig`: `model_relative_enabled` + the five ratio/cap fields, + all env-backed; `CompactionState.armed` + a `policy` snapshot of the cut; + `CompactionResult` gains `context_window`, `ceiling`, `floor`, + `hard_ceiling`, `forced`, `retained_tokens_estimate`. +- `TurnBasedSessionManager`: `_live_offset`; `update_after_turn(..., + context_window=, history_tokens=)` implements §3.2–§3.4. **No change to + when bytes change** — the slice still applies at restore only. +- Stream coordinator passes the catalog window (already looked up for the + badge) and the breakdown's `messages` tokens; the `compaction` SSE payload + carries the policy fields (additive — the SPA validator ignores extras; + the TS interface gains them as optional). +- Kill switch `AGENTCORE_MEMORY_COMPACTION_MODEL_RELATIVE_ENABLED=false` → + fixed threshold, turn-count cut, no arming. Default on (house style). + +**Acceptance:** unit tests for the table in §3.1 (including the unknown-window +and kill-switch rows); a 5-turn conversation with 1,000-token threshold cuts +to the oldest candidate under the floor; a turn over the ceiling on a disarmed +state is a no-op and a turn at the hard ceiling is a forced cut; the existing +byte-stability suite is unchanged; a replayed spiral-shaped sequence (input +constant above ceiling for 10 turns) produces exactly one checkpoint advance. + +### PR-2 — bounded summary (spiral spec PR-2, as written there) + +**Sequenced immediately after PR-1, ahead of PR-3, on evidence.** In the +2026-09-15 audit all 16 over-100k sessions carried `AGENT_CACHE_BYPASS` +(spreadsheet/word/ppt tools enabled), so they rebuild the agent every turn +and the restore slice already runs for them; zero were warm-agent +(`create_artifact`-only) sessions. What bit in every one was the slice +running and *still* not getting under the threshold — 23k–40k-token summaries +and the protected tail. Session 65b6d4ab: 17 live full re-writes from message +15 onward at 40–100k context, before the 40-message window ever engaged — +the compaction slice re-running on a fresh agent each turn with a changing +summary. That is this PR's problem, not PR-3's. + +### PR-3 — paid-when-free scheduling + `context_window_limit` (§3.5) + +Value on the audited cohort is the `create_artifact` (warm-agent) sessions and +turn latency, not the over-100k dollars — those are PR-1 + PR-2. Two of the +four window-pinned sessions examined (e7e75953, 0d8ba8f8) never exceeded 100k +and were pure 40-message-window effects, which PR-1's §3.0 change alone +recovers. + +**Acceptance:** on a warm agent held above the ceiling, the live list shrinks +on the first turn after a >300s gap and not before (unless hard ceiling); +`partial_miss` on the cut turn only; `test_second_cache_key_for_a_session_shares_the_conversation` +still passes with the in-place apply. + +### PR-4 — content-class eviction and offload escalation (§3.6) + +### PR-5 — selective 1h TTL experiment (§3.6) + +## 7. Observability + +- Log lines: `compaction_cut` (window, ceiling, floor, hard, relative cut, + absolute checkpoint, retained estimate), `compaction_disarmed_noop`, + `compaction_forced`, `compaction_rearmed`. +- The persisted `compaction.policy` map on the session row records what the + last cut believed the window and thresholds were, so + `GET /admin/costs/sessions/{id}/profile` can show it. The + `OVER_COMPACTION_THRESHOLD` diagnosis should read that map instead of the + fixed default once PR-1 is live (small follow-up, not in PR-1). +- **Measurement confounder:** the hourly system-prompt tick (`%H:00` in + `get_current_date_pacific()`) re-writes every session's prefix once an hour + regardless of compaction. Fix it (render the date only, or the hour in a + turn-scoped message) before attributing write:read movement to this spec's + PRs; filed separately, not in PR-1. +- `compactionCount` (already emitted) divided by session-days is the + cadence metric v2 §8 asks for; an alarm on >2/day is the spiral detector. + +## 8. Non-goals + +- Changing the cachePoint layout (tools/system/auto) — v2 §9. +- Replacing the mutation engine with Strands' conversation manager — that is + v2 and stays gated on its §7 criteria; every policy field here maps onto + v2's thin policy layer unchanged. +- Cross-session memory quality; the quota system; model routing. + +## 9. Open questions + +- Should an explicit `AGENTCORE_MEMORY_COMPACTION_TOKEN_THRESHOLD` in an + environment override the model-relative ceiling? PR-1 says no: it is the + fallback for an unknown window only. Flip this if an operator needs a + per-environment ceiling. +- Whether Bedrock prices Claude's 1M window in tiers the way it prices + GPT-6 Astra. The model cards say no long-context premium for the Claude + rows today; if that changes, `CEILING_CAP_TOKENS` is the one constant to + move. diff --git a/frontend/ai.client/src/app/shared/utils/stream-parser/stream-parser-types.ts b/frontend/ai.client/src/app/shared/utils/stream-parser/stream-parser-types.ts index 9b0a07b8a..85d21ba5b 100644 --- a/frontend/ai.client/src/app/shared/utils/stream-parser/stream-parser-types.ts +++ b/frontend/ai.client/src/app/shared/utils/stream-parser/stream-parser-types.ts @@ -218,6 +218,19 @@ export interface CompactionEvent { newCheckpoint: number; summarizedTurns: number; inputTokens: number; + /** + * Model-relative policy the cut was made under (additive, optional — + * docs/specs/compaction-model-relative-thresholds.md). `contextWindow` is + * the catalog's `maxInputTokens`; `ceiling` is the trigger, `floor` the + * target size after the cut, `hardCeiling` the level that forces a cut + * while the trigger is disarmed; `forced` says this cut was one of those. + */ + contextWindow?: number | null; + ceiling?: number | null; + floor?: number | null; + hardCeiling?: number | null; + forced?: boolean; + retainedTokensEstimate?: number | null; } /**