feat(compaction): offload oversized tool results at intake (PR-4)

backend/src/agents/main_agent/chat_agent.py

@@ -6,6 +6,7 @@
 """
 
 import logging
+import os
 from typing import Any, AsyncGenerator, Dict, List, Optional
 
 from agents.main_agent.base_agent import BaseAgent
@@ -58,14 +59,37 @@ def _create_agent(self) -> None:
             # files (added after tool filtering — it is infrastructure, not an
             # RBAC-gated tool, and is implicitly scoped to the turn's skills).
             plugin, read_skill_file = build_skills_runtime(self._accessible_skill_ids)
-            plugins = [plugin] if plugin else None
+            plugins = [plugin] if plugin else []
             if plugin:
                 tools = list(tools) + [read_skill_file]
                 logger.info(
                     "ChatAgent: skills disclosure enabled (%d accessible skill ids)",
                     len(self._accessible_skill_ids or []),
                 )
 
+            # Tool-result offload at intake (compaction PR-4): oversized tool
+            # results become a bounded preview + retrieval references before
+            # they enter the cacheable prefix. Fail-open: None when off or
+            # unconfigured. The plugin registers retrieve_offloaded_content
+            # itself — one stable spec in toolConfig, not an RBAC-gated tool,
+            # like read_skill_file above.
+            from agents.main_agent.core.tool_result_offload import build_tool_result_offloader
+
+            offload_session = getattr(self, "session_id", None)
+            offload_user = getattr(self, "user_id", None)
+            offloader = (
+                build_tool_result_offloader(
+                    session_id=offload_session,
+                    user_id=offload_user,
+                    region=os.environ.get("AWS_REGION"),
+                )
+                if offload_session and offload_user
+                else None
+            )
+            if offloader is not None:
+                plugins.append(offloader)
+            plugins = plugins or None
+
             self.agent = AgentFactory.create_agent(
                 model_config=self.model_config,
                 system_prompt=self._system_prompt_for(tools),

backend/src/agents/main_agent/config/constants.py

@@ -43,6 +43,24 @@ class EnvVars:
     # (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"
+    # Bounded compaction summary (spiral spec PR-2 /
+    # compaction-model-relative-thresholds.md §3.6). Budget in tokens; the
+    # re-summarize call is a Nova Micro side-channel with its own kill switch.
+    COMPACTION_SUMMARY_TOKEN_BUDGET = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_TOKEN_BUDGET"
+    COMPACTION_SUMMARY_MODEL_ENABLED = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ENABLED"
+    COMPACTION_SUMMARY_MODEL_ID = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ID"
+    # Paid-when-free scheduling (thresholds spec §3.5): a cut is computed and
+    # persisted as PENDING post-turn and applied to the live list pre-call only
+    # when the prefix re-write is free (cache expired, model/agent switched)
+    # or unavoidable (hard ceiling). "false" applies cuts immediately (PR-1/2).
+    COMPACTION_DEFERRED_APPLY_ENABLED = "AGENTCORE_MEMORY_COMPACTION_DEFERRED_APPLY_ENABLED"
+    # Tool-result offload at intake (thresholds spec §3.6 / PR-4): oversized
+    # tool results are stored in S3 (user-files bucket, per-session prefix) and
+    # replaced in context by a bounded preview + retrieval references before
+    # they ever enter the cacheable prefix. Strands' vended ContextOffloader.
+    TOOL_RESULT_OFFLOAD_ENABLED = "AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED"
+    TOOL_RESULT_OFFLOAD_MAX_TOKENS = "AGENTCORE_TOOL_RESULT_OFFLOAD_MAX_TOKENS"
+    TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS = "AGENTCORE_TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS"
 
     # --- Restored-history repair ---
     # Kill switch for the restore-time tool-pairing/alternation repair
@@ -148,6 +166,23 @@ class Defaults:
     # prompt. Overflow recovery (reduce_context on ContextWindowOverflow) still
     # works at any window size.
     CONVERSATION_WINDOW_MESSAGES = 2000
+    # 8k tokens ≈ 32k chars: a third of the 25k floor, so a bounded summary
+    # can never by itself hold a session above the ceiling (the incident's
+    # summary was 40k tokens against a 100k threshold). Same figure the admin
+    # SUMMARY_OVER_BUDGET diagnosis reads.
+    COMPACTION_SUMMARY_TOKEN_BUDGET = 8_000
+    COMPACTION_SUMMARY_MODEL_ENABLED = True
+    # Same cheap model as the title and tool-batch side-channels.
+    COMPACTION_SUMMARY_MODEL_ID = "us.amazon.nova-micro-v1:0"
+    COMPACTION_DEFERRED_APPLY_ENABLED = True
+    # Tool-result offload gate. 4k is well under the 25k compaction floor, so a
+    # protected tail of a few big results can no longer hold a session above
+    # the ceiling on its own; the 1k preview keeps the part of a result models
+    # actually quote (headers, first rows, the first error).
+    TOOL_RESULT_OFFLOAD_ENABLED = True
+    TOOL_RESULT_OFFLOAD_MAX_TOKENS = 4_000
+    TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS = 1_000
+    TOOL_RESULT_OFFLOAD_S3_PREFIX = "compaction-offload"
 
     # --- DynamoDB Tables ---
     DYNAMODB_QUOTA_TABLE = "UserQuotas"

backend/src/agents/main_agent/core/tool_result_offload.py

@@ -0,0 +1,186 @@
+"""
+Tool-result offload at intake — the compaction escalation for oversized tool
+results (docs/specs/compaction-model-relative-thresholds.md §3.6, PR-4).
+
+The compaction cut keeps the last ``protected_turns`` turns whole. When one of
+those turns carries a 90k-token tool result, the protected tail alone exceeds
+the floor and no cut can get the session back under the ceiling
+(``compaction_floor_unreachable``). Cutting inside the tail would drop the
+turn the user is working on; the right move is to keep the *reference* and
+drop the *bytes*.
+
+Strands 1.55's vended ``ContextOffloader`` does exactly that at the cheapest
+possible moment — ``AfterToolCallEvent``, before the result is appended to the
+conversation. The persisted message already carries the bounded form, so a
+restore reproduces it byte-for-byte and nothing in the prefix is ever
+mutated after the fact (the byte-stability contract in CLAUDE.md). The model
+keeps a preview and can pull any span back with ``retrieve_offloaded_content``
+(pattern / line range / full).
+
+What this module owns on top of the plugin:
+
+- **Storage**: S3 in the user-files bucket under
+  ``compaction-offload/{user_id}/{session_id}/`` — one namespaced storage per
+  agent, so references are scoped to the session by construction (a session
+  cannot retrieve another session's content) and a second agent instance for
+  the same session (an ``@``-mention) resolves the same references.
+- **No eviction from the model path.** ``evict_after_cycles=None``: the
+  plugin's cycle-based eviction runs on ``BeforeModelCallEvent`` and would
+  turn a retrieval into a miss mid-conversation. Objects expire by S3
+  lifecycle instead (see the file-upload construct).
+- **A cheap pre-filter.** The plugin sizes every result with
+  ``model.count_tokens`` — a Bedrock CountTokens round trip per tool call.
+  ``BoundedToolResultOffloader`` estimates with chars/4 first and only lets
+  results near or over the gate reach the API.
+- **A content-free record per offload** (``AgentCoreStack/Compaction``:
+  ``ToolResultOffloaded`` + token count), so the data point "how much tool
+  payload were we about to put in the prefix" exists.
+
+Documents attached by the user are NOT handled here: the digest + page-range
+read tool in ``document-context-offload.md`` is the right shape for those, and
+this module deliberately does not strip them.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Optional
+
+from agents.main_agent.config.constants import Defaults, EnvVars
+from agents.main_agent.session.compaction_policy import estimate_message_tokens
+
+logger = logging.getLogger(__name__)
+
+# Below this fraction of the gate the chars/4 estimate is trusted outright
+# and CountTokens is skipped. The heuristic is ~±25% on English text and JSON;
+# half the gate leaves that margin twice over.
+PREFILTER_RATIO = 0.5
+
+
+def tool_result_offload_enabled() -> bool:
+    """Default ON with a kill switch (house style): only the literal "false" disables."""
+    return os.environ.get(EnvVars.TOOL_RESULT_OFFLOAD_ENABLED, "").strip().lower() != "false"
+
+
+def _int_env(name: str, default: int) -> int:
+    raw = os.environ.get(name, "").strip()
+    try:
+        return int(raw) if raw else default
+    except ValueError:
+        return default
+
+
+def _import_plugin():
+    from strands.vended_plugins.context_offloader import ContextOffloader
+
+    return ContextOffloader
+
+
+class _OffloaderMixin:
+    """Behavior layered on the vended plugin; kept separate so it can be tested
+    against a stub base class without importing the real one."""
+
+    async def _handle_tool_result(self, event: Any) -> None:  # type: ignore[override]
+        try:
+            result = getattr(event, "result", None)
+            if not isinstance(result, dict):
+                return
+            content = result.get("content")
+            if not isinstance(content, list):
+                return
+            # Cheap pre-filter: skip the CountTokens round trip for results
+            # that cannot be anywhere near the gate.
+            estimate = estimate_message_tokens({"role": "user", "content": [{"toolResult": result}]})
+            if estimate < self._max_result_tokens * PREFILTER_RATIO:  # type: ignore[attr-defined]
+                return
+            before = event.result
+            await super()._handle_tool_result(event)  # type: ignore[misc]
+            if event.result is not before:
+                self._record_offload(event, estimate)
+        except Exception:  # noqa: BLE001 - offload is never worth a failed tool call
+            logger.warning("tool-result offload skipped, keeping the original result", exc_info=True)
+
+    @staticmethod
+    def _record_offload(event: Any, estimate: int) -> None:
+        tool_name = None
+        try:
+            tool_name = (getattr(event, "tool_use", None) or {}).get("name")
+        except Exception:  # noqa: BLE001
+            pass
+        logger.info("tool_result_offloaded: tool=%s est_tokens=%d", tool_name, estimate)
+        try:
+            from apis.shared.observability.prompt_cache import prompt_cache_observability_enabled
+            from apis.shared.observability.emf import emit_emf_metrics
+
+            if prompt_cache_observability_enabled():
+                emit_emf_metrics(
+                    "AgentCoreStack/Compaction",
+                    metrics={"ToolResultOffloaded": 1, "ToolResultOffloadedTokens": int(estimate)},
+                    properties={"toolName": tool_name},
+                    units={"ToolResultOffloadedTokens": "Count"},
+                )
+        except Exception as e:  # noqa: BLE001
+            logger.debug("offload EMF skipped: %s", e)
+
+
+def _offloader_class():
+    """``BoundedToolResultOffloader`` built lazily so importing this module never
+    imports the plugin (and boto3) on paths that do not use it."""
+    ContextOffloader = _import_plugin()
+
+    class BoundedToolResultOffloader(_OffloaderMixin, ContextOffloader):  # type: ignore[misc, valid-type]
+        pass
+
+    BoundedToolResultOffloader.__name__ = "BoundedToolResultOffloader"
+    return BoundedToolResultOffloader
+
+
+def offload_prefix(user_id: str, session_id: str) -> str:
+    return f"{Defaults.TOOL_RESULT_OFFLOAD_S3_PREFIX}/{user_id}/{session_id}"
+
+
+def build_tool_result_offloader(
+    *,
+    session_id: str,
+    user_id: str,
+    region: Optional[str] = None,
+    storage: Any = None,
+) -> Optional[Any]:
+    """The plugin for one agent, or ``None`` when off or unconfigured (fail-open).
+
+    ``storage`` overrides the S3 backend (tests). Returns ``None`` — never
+    raises — when the flag is off, the user-files bucket is not configured,
+    or the plugin cannot be constructed.
+    """
+    if not tool_result_offload_enabled():
+        return None
+    try:
+        max_tokens = max(1, _int_env(EnvVars.TOOL_RESULT_OFFLOAD_MAX_TOKENS, Defaults.TOOL_RESULT_OFFLOAD_MAX_TOKENS))
+        preview = _int_env(EnvVars.TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS, Defaults.TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS)
+        preview = max(0, min(preview, max_tokens - 1))
+
+        if storage is None:
+            bucket = os.environ.get("S3_USER_FILES_BUCKET_NAME", "").strip()
+            if not bucket:
+                logger.info("tool-result offload disabled: S3_USER_FILES_BUCKET_NAME is not set")
+                return None
+            from strands.storage import S3Storage
+
+            storage = S3Storage(
+                bucket,
+                prefix=offload_prefix(user_id, session_id),
+                region_name=region or os.environ.get("AWS_REGION", "us-west-2"),
+            )
+
+        cls = _offloader_class()
+        return cls(
+            storage=storage,
+            max_result_tokens=max_tokens,
+            preview_tokens=preview,
+            include_retrieval_tool=True,
+            evict_after_cycles=None,
+        )
+    except Exception:  # noqa: BLE001
+        logger.warning("tool-result offload disabled: plugin construction failed", exc_info=True)
+        return None

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

@@ -48,6 +48,18 @@ class CompactionState:
     # 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
+    # Paid-when-free scheduling (spec §3.5). A cut is computed post-turn and
+    # parked here; ``apply_pending_compaction`` promotes it to ``checkpoint``
+    # (and slices the live list in place) pre-call, only when the prefix
+    # re-write is free or unavoidable. ``checkpoint`` above is always the
+    # APPLIED one — what the restore slices at.
+    pending_checkpoint: Optional[int] = None
+    pending_summary: Optional[str] = None
+    pending_hard_ceiling: Optional[int] = None
+    pending_since: Optional[str] = None
+    # model id + agent id of the last turn; a change means the cached prefix
+    # is already invalid, so a pending cut can ride the same re-write.
+    last_prefix_key: Optional[str] = None
 
     def to_dict(self) -> Dict[str, Any]:
         """Convert to dictionary for DynamoDB storage."""
@@ -60,6 +72,11 @@ def to_dict(self) -> Dict[str, Any]:
             "truncationAnchor": self.truncation_anchor,
             "armed": self.armed,
             "policy": self.policy,
+            "pendingCheckpoint": self.pending_checkpoint,
+            "pendingSummary": self.pending_summary,
+            "pendingHardCeiling": self.pending_hard_ceiling,
+            "pendingSince": self.pending_since,
+            "lastPrefixKey": self.last_prefix_key,
         }
 
     @classmethod
@@ -82,6 +99,15 @@ def from_dict(cls, data: Optional[Dict[str, Any]]) -> "CompactionState":
             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,
+            pending_checkpoint=(
+                int(data["pendingCheckpoint"]) if data.get("pendingCheckpoint") is not None else None
+            ),
+            pending_summary=data.get("pendingSummary"),
+            pending_hard_ceiling=(
+                int(data["pendingHardCeiling"]) if data.get("pendingHardCeiling") is not None else None
+            ),
+            pending_since=data.get("pendingSince"),
+            last_prefix_key=data.get("lastPrefixKey"),
         )
 
 
@@ -110,6 +136,9 @@ class CompactionResult:
     # ceiling — the signal that the previous cut did not take.
     forced: bool = False
     retained_tokens_estimate: Optional[int] = None
+    # True when the cut was parked as pending (applied pre-call later under
+    # the paid-when-free rule) rather than promoted to the checkpoint now.
+    deferred: bool = False
 
 
 def _env_flag_default_on(name: str) -> bool:
@@ -142,6 +171,15 @@ class CompactionConfig:
     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
+    # Bounded summary (spec §3.6 / spiral spec PR-2). The persisted summary is
+    # held at or under this many tokens (chars/4), compressed once at cut time
+    # by the cheap model, with newest-first truncation as the fallback.
+    summary_token_budget: int = Defaults.COMPACTION_SUMMARY_TOKEN_BUDGET
+    summary_model_enabled: bool = Defaults.COMPACTION_SUMMARY_MODEL_ENABLED
+    summary_model_id: str = Defaults.COMPACTION_SUMMARY_MODEL_ID
+    # Paid-when-free scheduling (spec §3.5). Only meaningful with the
+    # model-relative policy on; legacy mode always applies immediately.
+    deferred_apply_enabled: bool = Defaults.COMPACTION_DEFERRED_APPLY_ENABLED
 
     @classmethod
     def from_env(cls) -> "CompactionConfig":
@@ -158,4 +196,8 @@ def from_env(cls) -> "CompactionConfig":
             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))),
+            summary_token_budget=int(os.environ.get(EnvVars.COMPACTION_SUMMARY_TOKEN_BUDGET, str(Defaults.COMPACTION_SUMMARY_TOKEN_BUDGET))),
+            summary_model_enabled=_env_flag_default_on(EnvVars.COMPACTION_SUMMARY_MODEL_ENABLED),
+            summary_model_id=os.environ.get(EnvVars.COMPACTION_SUMMARY_MODEL_ID, "").strip() or Defaults.COMPACTION_SUMMARY_MODEL_ID,
+            deferred_apply_enabled=_env_flag_default_on(EnvVars.COMPACTION_DEFERRED_APPLY_ENABLED),
         )