feat(compaction): model-relative thresholds, floor-seeking cut, hysteresis (PR-1)

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"

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)

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

@@ -1,13 +1,16 @@
 """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
 
 __all__ = [
     "SessionFactory",
     "CompactionState",
     "CompactionConfig",
+    "CompactionResult",
+    "CompactionPolicy",
     "TurnBasedSessionManager",
     "PreviewSessionManager",
     "is_preview_session",

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,16 +80,19 @@ 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,
         )
 
 
 @dataclass
 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,14 +125,23 @@ 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
     # since the previous turn, the cache entry has already expired, so pending
     # 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))),
         )