From 667d43c45acfbd57c7dbbaa183b82a50b23b5e74 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Tue, 15 Sep 2026 20:27:12 -0600 Subject: [PATCH 1/6] feat(compaction): bound the summary at 8k tokens and emit per-cut metrics (PR-2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The compaction summary was an unbounded join of AgentCore LTM ConversationSummary records (165k chars / ~40k tokens in the #833 incident): a summary that is 40% of the threshold guarantees compaction can never get back under it. Spiral-spec PR-2; thresholds spec §3.6 / §7.1. - compaction_summary.bound_summary(): hold the persisted summary at COMPACTION_SUMMARY_TOKEN_BUDGET (8,000 tokens, chars/4 — the same estimate the admin SUMMARY_OVER_BUDGET diagnosis uses). Within budget → unchanged. Over budget → one Nova Micro converse call (side-channel, never touches agent.messages) with a prompt that keeps standing instructions, decisions, current state of the work, open items and exact identifiers, and drops narration and superseded drafts. Model failure, a ceiling-hit generation or an overshoot → newest-first truncation (keep the newest records that fit; if none fit, the tail of the newest). Runs once at checkpoint advance — the turn that already pays a prefix re-write — and the result is persisted verbatim, so the byte-stability contract is unchanged. - Kill switch AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ENABLED=false skips the model and truncates; AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ID selects the model. - Provenance on the persisted compaction.policy map: summarySource (ltm|fallback), summaryOutcome, summaryTokensBefore/After, summaryTokenBudget. - One content-free EMF record per cut in AgentCoreStack/Compaction: CompactionCut, CompactionForced, CompactionInputTokens, CompactionRetainedTokens, CompactionSummaryTokens, CompactionSummaryOverBudget, with policySource/window/ceiling/floor/ summaryOutcome as queryable properties. Silenced by PROMPT_CACHE_OBSERVABILITY_ENABLED=false with the rest of the layer. - forced flag narrowed to "ran while disarmed" (same hunk as the PR-1 fix). Tests: newest-first truncation, within-budget passthrough, model compression, model failure / ceiling / overshoot fallbacks, kill switch, env loading, oversized LTM join bounded and persisted through update_after_turn, EMF record shape and kill switch. Co-Authored-By: Claude Fable 5.1 --- .../src/agents/main_agent/config/constants.py | 14 ++ .../main_agent/session/compaction_models.py | 9 + .../main_agent/session/compaction_summary.py | 207 +++++++++++++++++ .../session/turn_based_session_manager.py | 74 +++++- .../session/test_compaction_summary.py | 210 ++++++++++++++++++ .../compaction-model-relative-thresholds.md | 68 +++++- .../compaction-over-threshold-cache-spiral.md | 8 +- 7 files changed, 580 insertions(+), 10 deletions(-) create mode 100644 backend/src/agents/main_agent/session/compaction_summary.py create mode 100644 backend/tests/agents/main_agent/session/test_compaction_summary.py diff --git a/backend/src/agents/main_agent/config/constants.py b/backend/src/agents/main_agent/config/constants.py index d9ed82a7..980ee954 100644 --- a/backend/src/agents/main_agent/config/constants.py +++ b/backend/src/agents/main_agent/config/constants.py @@ -43,6 +43,12 @@ 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" # --- Restored-history repair --- # Kill switch for the restore-time tool-pairing/alternation repair @@ -148,6 +154,14 @@ 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" # --- DynamoDB Tables --- DYNAMODB_QUOTA_TABLE = "UserQuotas" diff --git a/backend/src/agents/main_agent/session/compaction_models.py b/backend/src/agents/main_agent/session/compaction_models.py index 733106de..35d5b736 100644 --- a/backend/src/agents/main_agent/session/compaction_models.py +++ b/backend/src/agents/main_agent/session/compaction_models.py @@ -142,6 +142,12 @@ 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 @classmethod def from_env(cls) -> "CompactionConfig": @@ -158,4 +164,7 @@ 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, ) diff --git a/backend/src/agents/main_agent/session/compaction_summary.py b/backend/src/agents/main_agent/session/compaction_summary.py new file mode 100644 index 00000000..64076ca1 --- /dev/null +++ b/backend/src/agents/main_agent/session/compaction_summary.py @@ -0,0 +1,207 @@ +""" +Bounded compaction summary. + +Spec: docs/specs/compaction-model-relative-thresholds.md §3.6 and +docs/specs/compaction-over-threshold-cache-spiral.md PR-2 (defect D2). + +The compaction summary used to be an unbounded join of AgentCore Long-Term +Memory ``ConversationSummary`` records — a log that grows for the life of the +session (164,991 chars ≈ 40k tokens in the incident). A summary that is 40% +of the threshold guarantees compaction can never get back under it. This +module holds the persisted summary at or under a token budget: + +1. **Within budget** → unchanged. +2. **Over budget** → one call to the cheap model (Nova Micro, the same + side-channel pattern as titles and tool-batch summaries) that compresses + the records into a bounded, instruction-preserving summary. It runs once, + at checkpoint advance — the turn that already pays a prefix re-write. +3. **Model unavailable / failed / still over budget** → newest-first + truncation of the records: keep the most recent records that fit, and if + even the newest alone does not fit, keep its tail. Never oldest-first — + recent context is what the model needs. + +Whatever comes out is persisted verbatim in ``CompactionState.summary`` and +prepended byte-identically at every restore, so the byte-stability contract +is unchanged: the summary still only mutates at checkpoint advance. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +from dataclasses import dataclass +from typing import List, Optional, Sequence + +from .compaction_policy import CHARS_PER_TOKEN + +logger = logging.getLogger(__name__) + +# Bound the *input* to the compression call too: a wide history of records is +# fed newest-first up to this many chars, so the side-channel's own spend is +# flat regardless of how long the session has run. +MAX_COMPRESSION_INPUT_CHARS = 120_000 +# Nova Micro's output ceiling is 5k tokens; stay under it with margin so the +# generation is not cut mid-sentence. The budget check after generation is +# what enforces the configured budget. +_MODEL_MAX_OUTPUT_TOKENS = 4_000 + +_COMPRESSION_SYSTEM_PROMPT = """You maintain the running summary of a long conversation between a user and an AI assistant. You are given the existing summary notes (oldest first). Rewrite them into ONE compact summary the assistant can continue the conversation from. + +Keep, in this order, and keep them exact: +1. Standing instructions, preferences and constraints the user gave (tone, format, length, language, things to avoid, people or systems to treat carefully). Quote them; do not paraphrase away specifics. +2. Decisions made and their reasons. +3. The current state of anything being built or edited (a document, essay, code, plan, dataset): what exists now, what is done, what is still open. +4. Open questions, pending tasks, and what the user asked for most recently. +5. Exact identifiers: file names, URLs, course or project names, IDs, numbers, dates, names of people. + +Drop: greetings, pleasantries, superseded drafts and revisions, step-by-step narration of tool calls, and anything already covered by a later note. + +Rules: +- Plain text with short headings or bullets. No preamble, no closing remark. +- Write in the language the conversation is in. +- Never invent facts; if notes conflict, the later note wins and say so briefly. +- Stay under {word_budget} words.""" + + +def approx_tokens(text: Optional[str]) -> int: + """chars/4 — the same estimate the admin SUMMARY_OVER_BUDGET diagnosis uses.""" + if not text: + return 0 + return len(text) // CHARS_PER_TOKEN + + +@dataclass(frozen=True) +class BoundedSummary: + text: Optional[str] + # "within_budget" | "model" | "truncated" | "truncated_after_model" | "empty" + outcome: str + tokens_before: int + tokens_after: int + + +def truncate_records_newest_first(records: Sequence[str], budget_tokens: int) -> Optional[str]: + """Keep the newest records that fit the budget (joined oldest→newest). + + If even the newest record alone exceeds the budget, keep its *tail*. + """ + budget_chars = max(0, int(budget_tokens)) * CHARS_PER_TOKEN + kept: List[str] = [] + used = 0 + for record in reversed([r for r in records if r]): + cost = len(record) + (2 if kept else 0) # "\n\n" joiner + if used + cost > budget_chars: + break + kept.append(record) + used += cost + if kept: + kept.reverse() + return "\n\n".join(kept) + newest = next((r for r in reversed(records) if r), None) + if not newest: + return None + if budget_chars <= 0: + return None + tail = newest[-budget_chars:] + return tail + + +def _compression_input(records: Sequence[str]) -> str: + """Records joined oldest→newest, trimmed to the input cap newest-first.""" + joined = "\n\n".join(r for r in records if r) + if len(joined) <= MAX_COMPRESSION_INPUT_CHARS: + return joined + return joined[-MAX_COMPRESSION_INPUT_CHARS:] + + +async def compress_with_model( + records: Sequence[str], + budget_tokens: int, + *, + model_id: str, + region: Optional[str] = None, +) -> Optional[str]: + """One bounded Bedrock ``converse`` call. Returns ``None`` on any failure. + + Side-channel by construction: its own messages, never ``agent.messages``. + """ + text = _compression_input(records) + if not text.strip(): + return None + try: + import boto3 + except ImportError: # pragma: no cover - dev without boto3 + return None + try: + region = region or os.environ.get("AWS_REGION", "us-west-2") + client = boto3.client("bedrock-runtime", region_name=region) + # ~0.75 words/token; aim well under the budget so the chars/4 check + # below passes with margin. + word_budget = max(150, int(budget_tokens * 0.55)) + response = await asyncio.to_thread( + client.converse, + modelId=model_id, + system=[{"text": _COMPRESSION_SYSTEM_PROMPT.replace("{word_budget}", f"{word_budget:,}")}], + messages=[{"role": "user", "content": [{"text": "Summary notes, oldest first:\n\n" + text}]}], + inferenceConfig={ + "temperature": 0.1, + "maxTokens": min(_MODEL_MAX_OUTPUT_TOKENS, max(256, int(budget_tokens))), + "topP": 0.9, + }, + ) + if response.get("stopReason") == "max_tokens": + logger.info("compaction_summary_model_truncated: generation hit the token ceiling; discarding") + return None + out = response["output"]["message"]["content"][0]["text"].strip() + return out or None + except Exception: # noqa: BLE001 - a summary is never worth an error + logger.warning("compaction_summary_model_failed: falling back to truncation", exc_info=True) + return None + + +async def bound_summary( + records: Sequence[str], + budget_tokens: int, + *, + model_enabled: bool, + model_id: str, + region: Optional[str] = None, +) -> BoundedSummary: + """Hold the summary built from ``records`` at or under ``budget_tokens``.""" + records = [r for r in records if isinstance(r, str) and r.strip()] + joined = "\n\n".join(records) if records else None + before = approx_tokens(joined) + if not joined: + return BoundedSummary(None, "empty", 0, 0) + if before <= budget_tokens: + return BoundedSummary(joined, "within_budget", before, before) + + if model_enabled: + compressed = await compress_with_model(records, budget_tokens, model_id=model_id, region=region) + if compressed is not None: + after = approx_tokens(compressed) + if after <= budget_tokens: + logger.info( + "compaction_summary_bounded: model %d -> %d tokens (budget=%d)", + before, after, budget_tokens, + ) + return BoundedSummary(compressed, "model", before, after) + # The model overshot: truncate ITS output newest-first (its tail + # holds the open items), rather than the raw records. + trimmed = truncate_records_newest_first([compressed], budget_tokens) + logger.info( + "compaction_summary_bounded: model overshot (%d > %d); tail-trimmed to %d", + after, budget_tokens, approx_tokens(trimmed), + ) + return BoundedSummary(trimmed, "truncated_after_model", before, approx_tokens(trimmed)) + outcome = "truncated_after_model" + else: + outcome = "truncated" + + truncated = truncate_records_newest_first(records, budget_tokens) + after = approx_tokens(truncated) + logger.info( + "compaction_summary_bounded: %s %d -> %d tokens (budget=%d)", + outcome, before, after, budget_tokens, + ) + return BoundedSummary(truncated, outcome, before, after) 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 9a924412..4b0eb453 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 @@ -42,6 +42,7 @@ from .compaction_models import CompactionState, CompactionConfig, CompactionResult from .compaction_policy import CompactionPolicy, choose_checkpoint +from .compaction_summary import bound_summary if TYPE_CHECKING: from strands.agent.agent import Agent @@ -935,13 +936,27 @@ async def update_after_turn( if current_checkpoint <= self._live_offset + idx < new_checkpoint ) - # Retrieve or generate summary for compacted messages + # Retrieve or generate the summary for the retired messages, then + # hold it at the budget (spec §3.6 / spiral spec PR-2). Compression + # runs here — once, at checkpoint advance, the turn that already pays + # a prefix re-write — and the result is persisted verbatim so every + # restore prepends identical bytes. summaries = self._retrieve_session_summaries() if summaries: - summary = "\n\n".join(summaries) + records = list(summaries) + summary_source = "ltm" else: - messages_to_summarize = messages[:relative_cut] - summary = self._generate_fallback_summary(messages_to_summarize) + fallback = self._generate_fallback_summary(messages[:relative_cut]) + records = [fallback] if fallback else [] + summary_source = "fallback" + bounded = await bound_summary( + records, + self.compaction_config.summary_token_budget, + model_enabled=self.compaction_config.summary_model_enabled, + model_id=self.compaction_config.summary_model_id, + region=self.region_name, + ) + summary = bounded.text state.checkpoint = new_checkpoint # The anchor rides the checkpoint: everything the slice retains stays @@ -959,9 +974,17 @@ async def update_after_turn( "forced": forced, "inputTokens": input_tokens, "retainedTokensEstimate": retained_estimate, + # Summary provenance — what the admin profile and the cost + # anatomy read to explain a cut without reading the conversation. + "summarySource": summary_source, + "summaryOutcome": bounded.outcome, + "summaryTokensBefore": bounded.tokens_before, + "summaryTokensAfter": bounded.tokens_after, + "summaryTokenBudget": self.compaction_config.summary_token_budget, } # This save is the compaction event itself — count it. self._save_compaction_state(state, record_event=True) + self._emit_compaction_metrics(state, policy, forced, retained_estimate, bounded) logger.info( f"Compaction checkpoint set: {new_checkpoint}, " @@ -983,6 +1006,49 @@ async def update_after_turn( retained_tokens_estimate=retained_estimate, ) + @staticmethod + def _emit_compaction_metrics(state, policy, forced, retained_estimate, bounded) -> None: + """One content-free EMF record per cut (``AgentCoreStack/Compaction``). + + Fleet-wide aggregates for the questions the specs keep asking: how + often cuts fire, how often they are forced (the spiral detector), how + big the summary is against its budget, and how deep cuts land. Never + raises; ``PROMPT_CACHE_OBSERVABILITY_ENABLED=false`` silences it with + the rest of the cost observability layer. + """ + try: + from apis.shared.observability.prompt_cache import prompt_cache_observability_enabled + from apis.shared.observability.emf import emit_emf_metrics + + if not prompt_cache_observability_enabled(): + return + emit_emf_metrics( + "AgentCoreStack/Compaction", + metrics={ + "CompactionCut": 1, + "CompactionForced": 1 if forced else 0, + "CompactionInputTokens": int(state.last_input_tokens or 0), + "CompactionRetainedTokens": int(retained_estimate or 0), + "CompactionSummaryTokens": int(bounded.tokens_after or 0), + "CompactionSummaryOverBudget": 1 if bounded.tokens_before > bounded.tokens_after else 0, + }, + properties={ + "policySource": policy.source, + "contextWindow": policy.context_window, + "ceiling": policy.ceiling, + "floor": policy.floor, + "summaryOutcome": bounded.outcome, + "summaryTokensBefore": bounded.tokens_before, + }, + units={ + "CompactionInputTokens": "Count", + "CompactionRetainedTokens": "Count", + "CompactionSummaryTokens": "Count", + }, + ) + except Exception as e: # noqa: BLE001 + logger.debug("Compaction EMF skipped: %s", e) + # ========================================================================= # Message Processing Helpers # ========================================================================= diff --git a/backend/tests/agents/main_agent/session/test_compaction_summary.py b/backend/tests/agents/main_agent/session/test_compaction_summary.py new file mode 100644 index 00000000..40c2e6ee --- /dev/null +++ b/backend/tests/agents/main_agent/session/test_compaction_summary.py @@ -0,0 +1,210 @@ +"""Bounded compaction summary — spiral spec PR-2 / thresholds spec §3.6.""" + +import sys +import types +from unittest.mock import MagicMock + +import pytest + +from agents.main_agent.session.compaction_models import CompactionConfig, CompactionState +from agents.main_agent.session.compaction_summary import ( + approx_tokens, + bound_summary, + truncate_records_newest_first, +) + +from .conftest import make_conversation + + +BUDGET = 100 # tokens → 400 chars + + +@pytest.fixture +def bedrock(monkeypatch): + """Patch boto3 so no test reaches Bedrock; returns the converse mock.""" + converse = MagicMock() + client = MagicMock() + client.converse = converse + module = types.SimpleNamespace(client=MagicMock(return_value=client)) + monkeypatch.setitem(sys.modules, "boto3", module) + return converse + + +def _model_reply(text, stop="end_turn"): + return {"stopReason": stop, "output": {"message": {"content": [{"text": text}]}}} + + +class TestTruncateNewestFirst: + def test_keeps_newest_records_that_fit(self): + records = ["old " * 50, "mid " * 50, "new " * 50] # 200 chars each + out = truncate_records_newest_first(records, budget_tokens=110) # 440 chars + assert out.startswith("mid") and out.endswith("new ") + assert "old" not in out + + def test_keeps_tail_of_newest_when_nothing_fits(self): + newest = "x" * 1000 + "TAIL" + out = truncate_records_newest_first(["ancient", newest], budget_tokens=10) # 40 chars + assert out.endswith("TAIL") and len(out) == 40 + + def test_empty(self): + assert truncate_records_newest_first([], 10) is None + assert truncate_records_newest_first(["", ""], 10) is None + + +class TestBoundSummary: + @pytest.mark.asyncio + async def test_within_budget_is_untouched(self, bedrock): + result = await bound_summary(["a", "b"], BUDGET, model_enabled=True, model_id="m") + assert result.text == "a\n\nb" and result.outcome == "within_budget" + bedrock.assert_not_called() + + @pytest.mark.asyncio + async def test_model_compresses_once_at_cut(self, bedrock): + bedrock.return_value = _model_reply("Standing instructions: cite APA. Open: intro draft.") + records = ["r" * 300, "s" * 300] + result = await bound_summary(records, BUDGET, model_enabled=True, model_id="m") + assert result.outcome == "model" + assert approx_tokens(result.text) <= BUDGET + assert bedrock.call_count == 1 + kwargs = bedrock.call_args.kwargs + assert kwargs["modelId"] == "m" + assert "Standing instructions" in kwargs["system"][0]["text"] + + @pytest.mark.asyncio + async def test_model_failure_falls_back_to_newest_first(self, bedrock): + bedrock.side_effect = RuntimeError("throttled") + records = ["old " * 100, "new " * 50] # 400 + 200 chars + result = await bound_summary(records, BUDGET, model_enabled=True, model_id="m") + assert result.outcome == "truncated_after_model" + assert result.text.startswith("new") and "old" not in result.text + assert approx_tokens(result.text) <= BUDGET + + @pytest.mark.asyncio + async def test_model_ceiling_hit_is_a_failure(self, bedrock): + bedrock.return_value = _model_reply("frag", stop="max_tokens") + result = await bound_summary(["r" * 900], BUDGET, model_enabled=True, model_id="m") + assert result.outcome == "truncated_after_model" + assert approx_tokens(result.text) <= BUDGET + + @pytest.mark.asyncio + async def test_model_overshoot_is_tail_trimmed(self, bedrock): + bedrock.return_value = _model_reply("y" * 2000 + "END") + result = await bound_summary(["r" * 900], BUDGET, model_enabled=True, model_id="m") + assert result.outcome == "truncated_after_model" + assert result.text.endswith("END") and approx_tokens(result.text) <= BUDGET + + @pytest.mark.asyncio + async def test_kill_switch_skips_model(self, bedrock): + result = await bound_summary(["r" * 900], BUDGET, model_enabled=False, model_id="m") + assert result.outcome == "truncated" + bedrock.assert_not_called() + + @pytest.mark.asyncio + async def test_empty(self, bedrock): + result = await bound_summary([], BUDGET, model_enabled=True, model_id="m") + assert result.text is None and result.outcome == "empty" + + def test_from_env(self, monkeypatch): + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_SUMMARY_TOKEN_BUDGET", "1234") + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ENABLED", "false") + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ID", "us.amazon.nova-lite-v1:0") + cfg = CompactionConfig.from_env() + assert cfg.summary_token_budget == 1234 + assert cfg.summary_model_enabled is False + assert cfg.summary_model_id == "us.amazon.nova-lite-v1:0" + monkeypatch.delenv("AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ENABLED") + assert CompactionConfig.from_env().summary_model_enabled is True + + +class TestThroughUpdateAfterTurn: + """Acceptance: oversized LTM records → the persisted summary is ≤ budget and + the restore prepends the same bounded bytes.""" + + def _manager(self, make_session_manager, records, **cfg): + config = CompactionConfig(enabled=True, token_threshold=1000, protected_turns=3, + summary_token_budget=BUDGET, **cfg) + mgr = make_session_manager(compaction_config=config) + mgr.compaction_state = CompactionState() + mgr._save_compaction_state = MagicMock() + mgr._retrieve_session_summaries = MagicMock(return_value=records) + mgr._valid_cutoff_indices = [0, 2, 4, 6, 8] + mgr._all_messages_for_summary = make_conversation(5) + return mgr + + @pytest.mark.asyncio + async def test_oversized_ltm_join_is_bounded_and_persisted(self, make_session_manager, bedrock): + bedrock.side_effect = RuntimeError("no model in tests") + records = [f"record {i} " + "z" * 600 for i in range(10)] # ~6k chars + mgr = self._manager(make_session_manager, records) + result = await mgr.update_after_turn(2000) + assert result is not None + state = mgr.compaction_state + assert approx_tokens(state.summary) <= BUDGET + # Nothing fits whole (each record ~610 chars vs a 400-char budget), so + # the tail of the NEWEST record is kept — never the oldest. + assert state.summary == records[-1][-BUDGET * 4:] + assert state.policy["summarySource"] == "ltm" + assert state.policy["summaryOutcome"] == "truncated_after_model" + assert state.policy["summaryTokensBefore"] > BUDGET >= state.policy["summaryTokensAfter"] + assert state.policy["summaryTokenBudget"] == BUDGET + # The restore path prepends exactly the persisted bytes. + restored = mgr._prepend_summary_to_first_message(make_conversation(2), state.summary) + assert state.summary in restored[0]["content"][0]["text"] + + @pytest.mark.asyncio + async def test_small_ltm_join_is_untouched(self, make_session_manager, bedrock): + mgr = self._manager(make_session_manager, ["LTM summary 1", "LTM summary 2"]) + await mgr.update_after_turn(2000) + assert mgr.compaction_state.summary == "LTM summary 1\n\nLTM summary 2" + assert mgr.compaction_state.policy["summaryOutcome"] == "within_budget" + bedrock.assert_not_called() + + @pytest.mark.asyncio + async def test_fallback_summary_is_labelled(self, make_session_manager, bedrock): + mgr = self._manager(make_session_manager, []) + await mgr.update_after_turn(2000) + assert mgr.compaction_state.policy["summarySource"] == "fallback" + assert "Previous conversation" in mgr.compaction_state.summary + + +class TestCompactionMetrics: + @pytest.mark.asyncio + async def test_cut_emits_one_content_free_emf_record(self, make_session_manager, bedrock, monkeypatch): + import apis.shared.observability.emf as emf + emitted = [] + monkeypatch.setattr(emf, "emit_emf_metrics", lambda ns, metrics, properties=None, units=None: emitted.append((ns, metrics, properties))) + monkeypatch.delenv("PROMPT_CACHE_OBSERVABILITY_ENABLED", raising=False) + config = CompactionConfig(enabled=True, token_threshold=1000, protected_turns=3, summary_token_budget=BUDGET) + mgr = make_session_manager(compaction_config=config) + mgr.compaction_state = CompactionState() + mgr._save_compaction_state = MagicMock() + mgr._retrieve_session_summaries = MagicMock(return_value=["tiny"]) + mgr._valid_cutoff_indices = [0, 2, 4, 6, 8] + mgr._all_messages_for_summary = make_conversation(5) + + await mgr.update_after_turn(2000) + + assert len(emitted) == 1 + ns, metrics, props = emitted[0] + assert ns == "AgentCoreStack/Compaction" + assert metrics["CompactionCut"] == 1 and metrics["CompactionForced"] == 0 + assert metrics["CompactionInputTokens"] == 2000 + assert props["summaryOutcome"] == "within_budget" and props["policySource"] == "fixed" + # Content-free: no summary text, no message text in the record. + assert "tiny" not in str(props) + + @pytest.mark.asyncio + async def test_kill_switch_silences_metrics(self, make_session_manager, bedrock, monkeypatch): + import apis.shared.observability.emf as emf + emitted = [] + monkeypatch.setattr(emf, "emit_emf_metrics", lambda *a, **k: emitted.append(a)) + monkeypatch.setenv("PROMPT_CACHE_OBSERVABILITY_ENABLED", "false") + config = CompactionConfig(enabled=True, token_threshold=1000, protected_turns=3) + mgr = make_session_manager(compaction_config=config) + mgr.compaction_state = CompactionState() + 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) + await mgr.update_after_turn(2000) + assert emitted == [] diff --git a/docs/specs/compaction-model-relative-thresholds.md b/docs/specs/compaction-model-relative-thresholds.md index 31eb4d7f..09513843 100644 --- a/docs/specs/compaction-model-relative-thresholds.md +++ b/docs/specs/compaction-model-relative-thresholds.md @@ -1,7 +1,8 @@ # 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. +**Status:** PR-1 open (#1125, this document rides with it). PR-2 (bounded +summary + compaction metrics) built 2026-09-15 on top of it, stacked. PR-3 +through PR-5 unbuilt. **Owner:** Phil Merrell **Related:** `compaction-over-threshold-cache-spiral.md` (#833 — the incident and the summary-cap PR this spec depends on) · @@ -321,7 +322,22 @@ 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) +### PR-2 — bounded summary (spiral spec PR-2, as written there) — BUILT + +As built (`compaction_summary.py`, stacked on PR-1): `bound_summary()` holds +the persisted summary at `COMPACTION_SUMMARY_TOKEN_BUDGET` (8,000 tokens, +chars/4 — the same estimate the admin `SUMMARY_OVER_BUDGET` diagnosis uses). +Within budget → unchanged. Over budget → one Nova Micro `converse` call +(`AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ID`; kill switch +`AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ENABLED=false`) with a prompt that +keeps standing instructions, decisions, current state of the work, open items +and exact identifiers, and drops narration and superseded drafts (the kaizen +2026-05-29 item). Model failure, a ceiling-hit generation, or an overshoot → +newest-first truncation (keep the newest records that fit; if none fit, the +tail of the newest). Runs once at cut time and the result is persisted +verbatim, so the byte-stability contract is untouched. Provenance lands in +the persisted `compaction.policy` map (`summarySource`, `summaryOutcome`, +`summaryTokensBefore/After`, `summaryTokenBudget`). **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` @@ -353,6 +369,52 @@ still passes with the in-place apply. ## 7. Observability +### 7.1 What PR-2 adds — one content-free record per cut + +`AgentCoreStack/Compaction` EMF namespace (silenced with the rest of the cost +observability layer by `PROMPT_CACHE_OBSERVABILITY_ENABLED=false`): + +| metric | answers | +|---|---| +| `CompactionCut` | cadence — cuts per session-day is v2 §8's spiral detector (alarm at >2/day) | +| `CompactionForced` | how often a cut ran while disarmed = how often the previous cut did not take | +| `CompactionInputTokens`, `CompactionRetainedTokens` | how far above the ceiling cuts fire and how deep they land (is the floor being reached?) | +| `CompactionSummaryTokens`, `CompactionSummaryOverBudget` | is the summary the reason cuts miss the floor; how often the model vs truncation path runs | + +Properties (queryable in Logs Insights, not dimensions): `policySource`, +`contextWindow`, `ceiling`, `floor`, `summaryOutcome`, `summaryTokensBefore`. +No text from the conversation or the summary is ever emitted. + +### 7.2 Data points still worth collecting (not built) + +The question behind all of them is *what does a turn cost because of history, +and what did compaction do to it* — today we can see the first half (cost +rows) and, after PR-2, the second, but not joined: + +- **Per-call `checkpoint` / `armed` / `liveOffset` on the `C#` cost row** — + lets the anatomy page show the context trajectory against the cuts without + correlating timestamps by hand. Additive fields on an existing write. +- **`cacheGapSeconds` next to every cut** — the PR-3 scheduling rule is only + measurable if each cut records whether it landed on a cold turn + (`rewrite_scheduled` vs `rewrite_forced`). +- **Retained-vs-actual calibration** — the next turn's measured + `contextBreakdown.messages` against the cut's `retainedTokensEstimate`. One + number per cut, and the only way to know whether the estimator is off by 5% + or 50%. +- **Turn shape** — messages per turn and tool-result bytes per turn (the + `ToolCensusHook` has the count; bytes are one more `ADD`). Sessions whose + bulk is a handful of huge tool results are the PR-4 offload cohort, and we + cannot size it today. +- **Outcome signal joined to compaction** — a down-thumb rate keyed to + "turns since last cut" is the first evidence that a cut costs anything + besides dollars (the kaizen review-queue already lists this as the + counterweight the cost roadmap lacks). Without it the §5 quality gate stays + a one-off eval rather than a standing measurement. +- **Warm-vs-cold split per cut** — whether the cut's session was on the + agent-cache bypass path (restore every turn) or a warm agent; the 2026-09-15 + audit had to infer this from `enabledTools`. It decides how much PR-3 is + worth. + - Log lines: `compaction_cut` (window, ceiling, floor, hard, relative cut, absolute checkpoint, retained estimate), `compaction_disarmed_noop`, `compaction_forced`, `compaction_rearmed`. diff --git a/docs/specs/compaction-over-threshold-cache-spiral.md b/docs/specs/compaction-over-threshold-cache-spiral.md index 799d8306..3f1a0e1f 100644 --- a/docs/specs/compaction-over-threshold-cache-spiral.md +++ b/docs/specs/compaction-over-threshold-cache-spiral.md @@ -1,8 +1,8 @@ # Compaction over-threshold cache spiral — stop paying a full prefix re-write on every turn **Status:** PR-1 shipped (#838). PR-5 shipped (#845 — see "As shipped" under -§3 PR-5; its acceptance replay moved one of §3's own numbers). PR-2 through -PR-4 unbuilt. **§4.1's adversarial re-scan is run** (2026-08-05, results +§3 PR-5; its acceptance replay moved one of §3's own numbers). PR-2 built +2026-09-15 (stacked on #1125, see below). PR-3 and PR-4 unbuilt. **§4.1's adversarial re-scan is run** (2026-08-05, results inline below): D2 and D3 both reproduce on sessions other than the incident's, and the harm is already multi-user. **Motivating incident:** prod, 2026-08-05 analysis. One faculty user exhausted @@ -277,7 +277,9 @@ regression case works forward from the deploy, not backward, and the fleet baseline in the roadmap's metric 1 starts at the first prod release — dev deployment alone does not start that clock. -### PR-2 — bound the compaction summary +### PR-2 — bound the compaction summary — BUILT 2026-09-15 + +*Built as `compaction_summary.py` on the model-relative-thresholds branch (stacked on #1125); see `compaction-model-relative-thresholds.md` §6 PR-2 for the as-built notes (budget constant, Nova Micro side-channel, newest-first fallback, provenance fields, EMF).* - Add `COMPACTION_SUMMARY_TOKEN_BUDGET` (default **8_000** tokens ≈ 32k chars) to [constants.py](../../backend/src/agents/main_agent/config/constants.py). From 94ac1529fa13d96e8cd80aa7d8a6e22cb88aad58 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Tue, 15 Sep 2026 22:51:28 -0600 Subject: [PATCH 2/6] feat(compaction): park cuts post-turn, apply in place when the re-write is free (PR-3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Compaction decided WHAT to cut (PR-1) and bounded the summary (PR-2), but a cut still landed at the next restore regardless of whether the prompt-cache prefix was warm, and never landed at all on a warm (cached) agent. Under Bedrock caching a cut costs one re-write of what survives, so the cheapest turn to pay it on is one that was going to re-write anyway. Thresholds spec §3.5. - update_after_turn PARKS the cut: pendingCheckpoint / pendingSummary / pendingHardCeiling / pendingSince on CompactionState. `checkpoint` stays the APPLIED value (what _apply_compaction slices at on restore). A second over-ceiling turn while a cut is parked is a no-op, never a deeper cut. - apply_pending_compaction(agent, prefix_key) runs at the head of every turn (stream coordinator, right after the turn lease is stamped), on cached and freshly restored agents alike. It promotes the pending cut and slices agent.messages IN PLACE (slice assignment, never rebinding — the #741 alias) when, in order: cache_expired (more than cache_ttl_seconds since the previous turn), prefix_changed (model|agent key differs from the persisted lastPrefixKey), or hard_ceiling (previous input reached the hard ceiling the cut was computed under). Otherwise it waits. - The in-place result is byte-identical to what _apply_compaction derives from stored history under the promoted state (pinned by test_live_apply_matches_a_cold_restore_of_the_same_state), so a cold restore after a live apply reads the same prefix. - _adopt_session_conversation copies _live_offset when it points a new agent at the live list — the list's coordinate system travels with it. - Each application persists the reason, cacheGapSeconds and pendingSince on compaction.policy, logs rewrite_scheduled vs rewrite_forced, and emits CompactionApplied / CompactionAppliedForced / CompactionCacheGapSeconds; the cut record gains CompactionDeferred. - Kill switch AGENTCORE_MEMORY_COMPACTION_DEFERRED_APPLY_ENABLED=false applies immediately (PR-1/2 behavior); legacy mode is always immediate. The shared test fixture pins the immediate path explicitly; deferral has its own suite. Not in this PR: context_window_limit on the Strands model config (needs the window at agent construction). Tests: parking, no-deeper-cut, kill switch, legacy; warm/same-prefix waits; cache-expired / prefix-changed / hard-ceiling apply; first-ever key is not a change; live apply == cold restore parity; pending beyond the live list is dropped; re-arm and cut again after apply; metrics with reason; state round-trip; offset sync on alias. Full backend suite: 8659 passed. Co-Authored-By: Claude Fable 5.1 --- .../src/agents/main_agent/config/constants.py | 6 + .../main_agent/session/compaction_models.py | 33 +++ .../session/turn_based_session_manager.py | 254 +++++++++++++++--- .../streaming/stream_coordinator.py | 15 ++ .../src/apis/inference_api/chat/service.py | 18 +- .../agents/main_agent/session/conftest.py | 4 + .../session/test_compaction_deferred_apply.py | 235 ++++++++++++++++ .../session/test_compaction_models.py | 5 + .../session/test_compaction_policy.py | 4 +- .../session/test_compaction_summary.py | 6 +- .../apis/inference_api/test_chat_service.py | 20 ++ .../compaction-model-relative-thresholds.md | 54 +++- 12 files changed, 600 insertions(+), 54 deletions(-) create mode 100644 backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py diff --git a/backend/src/agents/main_agent/config/constants.py b/backend/src/agents/main_agent/config/constants.py index 980ee954..5c4171c5 100644 --- a/backend/src/agents/main_agent/config/constants.py +++ b/backend/src/agents/main_agent/config/constants.py @@ -49,6 +49,11 @@ class EnvVars: 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" # --- Restored-history repair --- # Kill switch for the restore-time tool-pairing/alternation repair @@ -162,6 +167,7 @@ class Defaults: 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 # --- DynamoDB Tables --- DYNAMODB_QUOTA_TABLE = "UserQuotas" diff --git a/backend/src/agents/main_agent/session/compaction_models.py b/backend/src/agents/main_agent/session/compaction_models.py index 35d5b736..4bcc5064 100644 --- a/backend/src/agents/main_agent/session/compaction_models.py +++ b/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: @@ -148,6 +177,9 @@ class CompactionConfig: 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": @@ -167,4 +199,5 @@ def from_env(cls) -> "CompactionConfig": 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), ) 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 4b0eb453..14d0e5a7 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 @@ -115,6 +115,10 @@ def __init__( # 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 + # model|agent key stamped at the head of each turn by the coordinator; + # persisted at turn end so the next head-of-turn can tell whether the + # cached prefix is already invalid (spec §3.5). + self._current_prefix_key: Optional[str] = None # Session control self.cancelled = False @@ -804,6 +808,8 @@ async def update_after_turn( self._adopt_persisted_compaction_state() state = self.compaction_state state.last_input_tokens = input_tokens + if self._current_prefix_key: + state.last_prefix_key = self._current_prefix_key policy = CompactionPolicy.resolve(self.compaction_config, context_window) @@ -822,6 +828,15 @@ async def update_after_turn( # 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 state.pending_checkpoint is not None and policy.hysteresis_enabled: + # A cut is already parked. Cutting deeper now would be the spiral; + # the head of the next turn applies it (hard ceiling included). + logger.info( + "compaction_pending_waiting: input=%d > ceiling=%d, pending_checkpoint=%d, hard=%s", + input_tokens, policy.ceiling, state.pending_checkpoint, policy.hard_ceiling, + ) + self._save_compaction_state(state) + return None 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 @@ -958,12 +973,22 @@ async def update_after_turn( ) summary = bounded.text - 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. - state.truncation_anchor = max(state.truncation_anchor, new_checkpoint) - state.summary = summary + deferred = bool(self.compaction_config.deferred_apply_enabled and policy.hysteresis_enabled) + if deferred: + # Park the cut. The bytes the model sees do not change until + # ``apply_pending_compaction`` decides the re-write is free or + # unavoidable (spec §3.5). + state.pending_checkpoint = new_checkpoint + state.pending_summary = summary + state.pending_hard_ceiling = policy.hard_ceiling + state.pending_since = datetime.now(timezone.utc).isoformat() + else: + 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. + 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. state.total_summarized_turns += summarized_turns @@ -981,10 +1006,11 @@ async def update_after_turn( "summaryTokensBefore": bounded.tokens_before, "summaryTokensAfter": bounded.tokens_after, "summaryTokenBudget": self.compaction_config.summary_token_budget, + "deferred": deferred, } # This save is the compaction event itself — count it. self._save_compaction_state(state, record_event=True) - self._emit_compaction_metrics(state, policy, forced, retained_estimate, bounded) + self._emit_compaction_metrics(state, policy, forced, retained_estimate, bounded, deferred) logger.info( f"Compaction checkpoint set: {new_checkpoint}, " @@ -1004,17 +1030,161 @@ async def update_after_turn( hard_ceiling=policy.hard_ceiling, forced=forced, retained_tokens_estimate=retained_estimate, + deferred=deferred, ) + # ========================================================================= + # Paid-when-free application (spec §3.5) + # ========================================================================= + + def apply_pending_compaction(self, agent: "Agent", *, prefix_key: Optional[str] = None) -> Optional[str]: + """Head-of-turn: apply a parked cut to the live list if the re-write is free. + + Called by the stream coordinator before the first model call of every + turn, on cached and freshly restored agents alike. Promotes + ``pending_checkpoint`` to ``checkpoint`` and slices ``agent.messages`` + **in place** (slice assignment, never rebinding — the #741 alias) when + one of these holds, in this order: + + - ``cache_expired`` — more than ``cache_ttl_seconds`` since the last + turn: the Bedrock entry is gone and the next call re-writes the + prefix anyway, so the cut rides for free; + - ``prefix_changed`` — ``prefix_key`` (model|agent) differs from the + last turn's: the cached prefix is already invalid; + - ``hard_ceiling`` — the last turn's input reached the hard ceiling + the cut was computed under: waiting is no longer affordable. + + Otherwise the cut keeps waiting (``compaction_pending_waiting``). Returns + the reason applied, or ``None``. Never raises. + """ + if not self.compaction_config or not self.compaction_config.enabled: + return None + try: + self._adopt_persisted_compaction_state() + state = self.compaction_state + if state is None: + return None + + previous_key = state.last_prefix_key + self._current_prefix_key = prefix_key + if state.pending_checkpoint is None: + return None + + config = self.compaction_config + gap_seconds = self._seconds_since(state.updated_at) + reason: Optional[str] = None + if self._cache_window_expired(state.updated_at, config.cache_ttl_seconds): + reason = "cache_expired" + elif prefix_key and previous_key and prefix_key != previous_key: + reason = "prefix_changed" + elif state.pending_hard_ceiling is not None and state.last_input_tokens >= state.pending_hard_ceiling: + reason = "hard_ceiling" + + if reason is None: + logger.info( + "compaction_pending_waiting: pending_checkpoint=%d, gap=%ss, last_input=%d, hard=%s", + state.pending_checkpoint, gap_seconds, state.last_input_tokens, state.pending_hard_ceiling, + ) + return None + + messages = getattr(agent, "messages", None) + if not isinstance(messages, list): + return None + applied = self._apply_pending_in_place(messages, state) + if not applied: + # Unusable pending (would drop the whole live list): clear it + # rather than leave a cut that can never land. + state.pending_checkpoint = None + state.pending_summary = None + state.pending_hard_ceiling = None + state.pending_since = None + self._save_compaction_state(state) + return None + + promoted = state.pending_checkpoint + state.checkpoint = promoted + state.truncation_anchor = max(state.truncation_anchor, promoted) + state.summary = state.pending_summary + state.pending_checkpoint = None + state.pending_summary = None + state.pending_hard_ceiling = None + pending_since = state.pending_since + state.pending_since = None + state.policy = { + **(state.policy or {}), + "applied": reason, + "appliedAt": datetime.now(timezone.utc).isoformat(), + "cacheGapSeconds": gap_seconds, + "pendingSince": pending_since, + } + self._save_compaction_state(state) + logger.info( + "compaction_applied: reason=%s checkpoint=%d gap=%ss live_len=%d (%s)", + reason, promoted, gap_seconds, len(messages), + "rewrite_forced" if reason == "hard_ceiling" else "rewrite_scheduled", + ) + self._emit_emf( + { + "CompactionApplied": 1, + "CompactionAppliedForced": 1 if reason == "hard_ceiling" else 0, + "CompactionCacheGapSeconds": int(gap_seconds or 0), + }, + {"applyReason": reason, "checkpoint": promoted}, + {"CompactionCacheGapSeconds": "Seconds"}, + ) + return reason + except Exception as e: # noqa: BLE001 - never break a turn + logger.warning(f"apply_pending_compaction skipped: {e}", exc_info=True) + return None + + def _apply_pending_in_place(self, messages: List[Dict], state: CompactionState) -> bool: + """Slice the live list at the pending checkpoint, in place. + + ``messages[0]`` sits at absolute index ``_live_offset``; the pending + checkpoint is absolute. Produces the same bytes ``_apply_compaction`` + would derive from stored history with the promoted state (slice, then + the summary prepended to the new first user message), so a later cold + restore matches the live prefix. + """ + pending = state.pending_checkpoint + if pending is None: + return False + k = pending - self._live_offset + if k <= 0: + # Live list already starts at/after the cut (e.g. a restore that + # sliced there). Nothing to remove; promotion still records it. + return True + if k >= len(messages): + logger.warning( + "compaction pending checkpoint %d is beyond the live list (offset=%d, len=%d); dropping it", + pending, self._live_offset, len(messages), + ) + return False + head = messages[k] + if state.pending_summary: + head = self._prepend_summary_to_first_message([head], state.pending_summary)[0] + messages[:] = [head] + messages[k + 1:] + self._live_offset = pending + return True + @staticmethod - def _emit_compaction_metrics(state, policy, forced, retained_estimate, bounded) -> None: - """One content-free EMF record per cut (``AgentCoreStack/Compaction``). - - Fleet-wide aggregates for the questions the specs keep asking: how - often cuts fire, how often they are forced (the spiral detector), how - big the summary is against its budget, and how deep cuts land. Never - raises; ``PROMPT_CACHE_OBSERVABILITY_ENABLED=false`` silences it with - the rest of the cost observability layer. + def _seconds_since(updated_at: Optional[str]) -> Optional[int]: + if not updated_at: + return None + try: + last = datetime.fromisoformat(updated_at.replace("Z", "+00:00")) + except (ValueError, TypeError): + return None + if last.tzinfo is None: + last = last.replace(tzinfo=timezone.utc) + return int((datetime.now(timezone.utc) - last).total_seconds()) + + @staticmethod + def _emit_emf(metrics: Dict[str, Any], properties: Dict[str, Any], units: Optional[Dict[str, str]] = None) -> None: + """One content-free EMF record in ``AgentCoreStack/Compaction``. Never raises. + + ``PROMPT_CACHE_OBSERVABILITY_ENABLED=false`` silences it with the rest + of the cost observability layer. """ try: from apis.shared.observability.prompt_cache import prompt_cache_observability_enabled @@ -1022,33 +1192,39 @@ def _emit_compaction_metrics(state, policy, forced, retained_estimate, bounded) if not prompt_cache_observability_enabled(): return - emit_emf_metrics( - "AgentCoreStack/Compaction", - metrics={ - "CompactionCut": 1, - "CompactionForced": 1 if forced else 0, - "CompactionInputTokens": int(state.last_input_tokens or 0), - "CompactionRetainedTokens": int(retained_estimate or 0), - "CompactionSummaryTokens": int(bounded.tokens_after or 0), - "CompactionSummaryOverBudget": 1 if bounded.tokens_before > bounded.tokens_after else 0, - }, - properties={ - "policySource": policy.source, - "contextWindow": policy.context_window, - "ceiling": policy.ceiling, - "floor": policy.floor, - "summaryOutcome": bounded.outcome, - "summaryTokensBefore": bounded.tokens_before, - }, - units={ - "CompactionInputTokens": "Count", - "CompactionRetainedTokens": "Count", - "CompactionSummaryTokens": "Count", - }, - ) + emit_emf_metrics("AgentCoreStack/Compaction", metrics=metrics, properties=properties, units=units or {}) except Exception as e: # noqa: BLE001 logger.debug("Compaction EMF skipped: %s", e) + def _emit_compaction_metrics(self, state, policy, forced, retained_estimate, bounded, deferred=False) -> None: + """One record per cut: how often cuts fire, how often they are forced + (the spiral detector), how big the summary is against its budget, how + deep cuts land, and whether the cut was parked for a free turn.""" + self._emit_emf( + { + "CompactionCut": 1, + "CompactionForced": 1 if forced else 0, + "CompactionDeferred": 1 if deferred else 0, + "CompactionInputTokens": int(state.last_input_tokens or 0), + "CompactionRetainedTokens": int(retained_estimate or 0), + "CompactionSummaryTokens": int(bounded.tokens_after or 0), + "CompactionSummaryOverBudget": 1 if bounded.tokens_before > bounded.tokens_after else 0, + }, + { + "policySource": policy.source, + "contextWindow": policy.context_window, + "ceiling": policy.ceiling, + "floor": policy.floor, + "summaryOutcome": bounded.outcome, + "summaryTokensBefore": bounded.tokens_before, + }, + { + "CompactionInputTokens": "Count", + "CompactionRetainedTokens": "Count", + "CompactionSummaryTokens": "Count", + }, + ) + # ========================================================================= # Message Processing Helpers # ========================================================================= diff --git a/backend/src/agents/main_agent/streaming/stream_coordinator.py b/backend/src/agents/main_agent/streaming/stream_coordinator.py index 1513531d..baf19a7f 100644 --- a/backend/src/agents/main_agent/streaming/stream_coordinator.py +++ b/backend/src/agents/main_agent/streaming/stream_coordinator.py @@ -272,6 +272,21 @@ async def stream_response( if session_manager is not None: session_manager.turn_lease = turn_lease + # Paid-when-free compaction (spec §3.5): if a cut is parked, apply it + # to the live list now — before the first model call — only when the + # prefix re-write is free (cache expired, model/agent switched) or + # unavoidable (hard ceiling). Runs on cached and freshly restored + # agents alike; the session manager decides, this just supplies the + # model|agent key. Best-effort: never blocks the turn. + if session_manager is not None and hasattr(session_manager, "apply_pending_compaction"): + try: + _model_for_key = getattr(getattr(main_agent_wrapper, "model_config", None), "model_id", None) + session_manager.apply_pending_compaction( + agent, prefix_key=f"{_model_for_key}|{turn_agent_id or 'default'}" + ) + except Exception as e: # noqa: BLE001 + logger.warning(f"apply_pending_compaction failed, continuing: {e}") + # Likewise a pause armed by a previous turn: if the user abandoned an # OAuth/tool-approval consent and just typed again, the still-armed # interrupt state makes Strands reject this turn's prompt outright. diff --git a/backend/src/apis/inference_api/chat/service.py b/backend/src/apis/inference_api/chat/service.py index bcd5b971..93f6279d 100644 --- a/backend/src/apis/inference_api/chat/service.py +++ b/backend/src/apis/inference_api/chat/service.py @@ -153,7 +153,9 @@ def _adopt_session_conversation(agent: BaseAgent, session_id: str) -> None: ``agent.messages`` lives inside ``TurnBasedSessionManager.initialize()`` (document stripping, content-block sanitizing, compaction slicing, pairing repair) and so runs before we get here; after construction the list is only - appended to. A future compaction that rebinds mid-life would silently break + appended to — or, for the pending-cut apply at the head of a turn, sliced + **in place** by slice assignment (``messages[:] = ...``), which keeps the + alias. A future compaction that rebinds mid-life would silently break the alias — ``test_second_cache_key_for_a_session_shares_the_conversation`` is what catches that. @@ -174,12 +176,14 @@ def _adopt_session_conversation(agent: BaseAgent, session_id: str) -> None: return live = None + live_wrapper = None for key, cached in _agent_cache.items(): if key[0] != session_id: continue cached_inner = getattr(cached, "agent", None) if isinstance(getattr(cached_inner, "messages", None), list): live = cached_inner # newest wins — dict preserves insertion order + live_wrapper = cached if live is None or live.messages is inner.messages: return @@ -203,6 +207,18 @@ def _adopt_session_conversation(agent: BaseAgent, session_id: str) -> None: ) inner.messages = live.messages + # The list's coordinate system travels with it. Compaction expresses its + # checkpoint as ``_live_offset + index into this list`` and the pending-cut + # apply slices the list in place at that offset, so an instance that adopts + # the list must adopt the offset too or it would slice at the wrong place. + try: + src_sm = getattr(live_wrapper, "session_manager", None) + dst_sm = getattr(agent, "session_manager", None) + if src_sm is not None and dst_sm is not None and hasattr(src_sm, "_live_offset"): + dst_sm._live_offset = src_sm._live_offset + except Exception: # noqa: BLE001 - never let bookkeeping break a turn + logger.debug("Session %s: could not sync compaction live offset", scrub_log(session_id), exc_info=True) + async def get_agent( session_id: str, diff --git a/backend/tests/agents/main_agent/session/conftest.py b/backend/tests/agents/main_agent/session/conftest.py index 57aa832e..d232ba13 100644 --- a/backend/tests/agents/main_agent/session/conftest.py +++ b/backend/tests/agents/main_agent/session/conftest.py @@ -126,6 +126,10 @@ def compaction_config() -> CompactionConfig: token_threshold=1000, protected_turns=3, max_tool_content_length=50, + # These suites pin the immediate-apply path (checkpoint moves at turn + # end). Paid-when-free deferral is the default in prod and has its own + # suite: test_compaction_deferred_apply.py. + deferred_apply_enabled=False, ) diff --git a/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py b/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py new file mode 100644 index 00000000..9a881e4e --- /dev/null +++ b/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py @@ -0,0 +1,235 @@ +"""Paid-when-free compaction — docs/specs/compaction-model-relative-thresholds.md §3.5. + +Post-turn parks the cut as PENDING; the head of the next turn applies it to +the live list in place only when the prefix re-write is free (cache expired, +model/agent switched) or unavoidable (hard ceiling). +""" + +import copy +import json +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +import pytest + +from agents.main_agent.session.compaction_models import CompactionConfig, CompactionState +from agents.main_agent.session.compaction_summary import approx_tokens + +from .conftest import make_conversation + + +CONFIG = dict(enabled=True, token_threshold=1000, protected_turns=3, summary_token_budget=200) + + +def _now(): + return datetime.now(timezone.utc) + + +def _dump(messages): + return json.dumps(messages, sort_keys=True) + + +def _manager(make_session_manager, store, **overrides): + """Manager wired to an in-memory state store (deferred mode ON).""" + cfg = CompactionConfig(**{**CONFIG, **overrides}) + mgr = make_session_manager(compaction_config=cfg) + mgr._load_compaction_state = lambda: CompactionState.from_dict(store.get("compaction")) + + def _save(state, record_event=False): + state.updated_at = _now().isoformat() + store["compaction"] = state.to_dict() + + mgr._save_compaction_state = _save + mgr._retrieve_session_summaries = lambda: ["LTM summary"] + mgr.compaction_state = CompactionState.from_dict(store.get("compaction")) + return mgr + + +def _agent(messages): + agent = MagicMock() + agent.messages = messages + return agent + + +def _age(store, seconds): + store["compaction"]["updatedAt"] = (_now() - timedelta(seconds=seconds)).isoformat() + + +class TestPostTurnParksTheCut: + @pytest.mark.asyncio + async def test_over_ceiling_parks_pending_and_leaves_checkpoint(self, make_session_manager): + store = {} + mgr = _manager(make_session_manager, store) + live = make_conversation(5) + result = await mgr.update_after_turn(2000, current_messages=live) + assert result is not None and result.deferred is True + state = mgr.compaction_state + assert state.pending_checkpoint == 4 and state.checkpoint == 0 + assert state.pending_summary == "LTM summary" + assert state.pending_hard_ceiling == 1500 + assert state.armed is False + assert store["compaction"]["pendingCheckpoint"] == 4 + assert state.policy["deferred"] is True + # Nothing about the live list changed at turn end. + assert len(live) == 10 and "conversation_summary" not in _dump(live) + + @pytest.mark.asyncio + async def test_second_over_ceiling_turn_does_not_cut_deeper(self, make_session_manager): + store = {} + mgr = _manager(make_session_manager, store) + await mgr.update_after_turn(2000, current_messages=make_conversation(5)) + second = await mgr.update_after_turn(2500, current_messages=make_conversation(7)) # even at hard ceiling + assert second is None + assert mgr.compaction_state.pending_checkpoint == 4 + + @pytest.mark.asyncio + async def test_kill_switch_applies_immediately(self, make_session_manager): + store = {} + mgr = _manager(make_session_manager, store, deferred_apply_enabled=False) + result = await mgr.update_after_turn(2000, current_messages=make_conversation(5)) + assert result.deferred is False + assert mgr.compaction_state.checkpoint == 4 and mgr.compaction_state.pending_checkpoint is None + + @pytest.mark.asyncio + async def test_legacy_mode_applies_immediately(self, make_session_manager): + store = {} + mgr = _manager(make_session_manager, store, model_relative_enabled=False) + result = await mgr.update_after_turn(2000, current_messages=make_conversation(5)) + assert result.deferred is False and mgr.compaction_state.checkpoint == 4 + + +class TestHeadOfTurnApply: + async def _parked(self, make_session_manager, store): + """Park a cut at 1200 tokens: over the ceiling (1000), under hard (1500).""" + mgr = _manager(make_session_manager, store) + live = make_conversation(5) + await mgr.update_after_turn(1200, current_messages=live) + assert mgr.compaction_state.pending_checkpoint == 4 + return mgr, live + + @pytest.mark.asyncio + async def test_warm_cache_same_prefix_below_hard_waits(self, make_session_manager): + store = {} + mgr, live = await self._parked(make_session_manager, store) + before = _dump(live) + assert mgr.apply_pending_compaction(_agent(live), prefix_key="m|default") is None + assert _dump(live) == before + assert mgr.compaction_state.pending_checkpoint == 4 and mgr.compaction_state.checkpoint == 0 + + @pytest.mark.asyncio + async def test_cache_expired_applies_in_place(self, make_session_manager): + store = {} + mgr, live = await self._parked(make_session_manager, store) + _age(store, 600) + list_id = id(live) + reason = mgr.apply_pending_compaction(_agent(live), prefix_key="m|default") + assert reason == "cache_expired" + assert id(live) == list_id # slice assignment, never rebound + assert len(live) == 6 # 10 - 4 + assert live[0]["role"] == "user" and "conversation_summary" in live[0]["content"][0]["text"] + assert "LTM summary" in live[0]["content"][0]["text"] + state = mgr.compaction_state + assert state.checkpoint == 4 and state.truncation_anchor == 4 and state.summary == "LTM summary" + assert state.pending_checkpoint is None and state.pending_summary is None + assert mgr._live_offset == 4 + assert state.policy["applied"] == "cache_expired" + assert state.policy["cacheGapSeconds"] >= 599 + assert store["compaction"]["checkpoint"] == 4 and store["compaction"]["pendingCheckpoint"] is None + + @pytest.mark.asyncio + async def test_prefix_change_applies(self, make_session_manager): + store = {} + mgr = _manager(make_session_manager, store) + live = make_conversation(5) + # Turn 1 stamps the key; turn end persists it. + mgr.apply_pending_compaction(_agent(live), prefix_key="sonnet|default") + await mgr.update_after_turn(1200, current_messages=live) + assert store["compaction"]["lastPrefixKey"] == "sonnet|default" + # Same key, warm: waits. Different key (model switch): applies. + assert mgr.apply_pending_compaction(_agent(live), prefix_key="sonnet|default") is None + assert mgr.apply_pending_compaction(_agent(live), prefix_key="opus|default") == "prefix_changed" + assert len(live) == 6 + + @pytest.mark.asyncio + async def test_first_ever_prefix_key_is_not_a_change(self, make_session_manager): + store = {} + mgr, live = await self._parked(make_session_manager, store) # no key was ever stamped + assert mgr.apply_pending_compaction(_agent(live), prefix_key="m|default") is None + + @pytest.mark.asyncio + async def test_hard_ceiling_forces_apply(self, make_session_manager): + store = {} + mgr, live = await self._parked(make_session_manager, store) + await mgr.update_after_turn(1600, current_messages=live) # pending waits post-turn... + assert mgr.compaction_state.pending_checkpoint == 4 + assert mgr.apply_pending_compaction(_agent(live), prefix_key="m|default") == "hard_ceiling" # ...and lands pre-call + assert len(live) == 6 + + @pytest.mark.asyncio + async def test_live_apply_matches_a_cold_restore_of_the_same_state(self, make_session_manager): + """The in-place slice must produce the bytes _apply_compaction derives + from stored history under the promoted state (prefix cache parity).""" + store = {} + mgr, live = await self._parked(make_session_manager, store) + stored = copy.deepcopy(live) + _age(store, 600) + assert mgr.apply_pending_compaction(_agent(live), prefix_key="m|default") == "cache_expired" + + cold = _manager(make_session_manager, store) + agent = _agent(copy.deepcopy(stored)) + cold._apply_compaction(agent) + assert _dump(agent.messages) == _dump(live) + assert cold._live_offset == mgr._live_offset == 4 + + @pytest.mark.asyncio + async def test_pending_beyond_live_list_is_dropped_not_applied(self, make_session_manager): + store = {} + mgr, live = await self._parked(make_session_manager, store) + _age(store, 600) + short = live[:2] # a list that does not reach the cut + assert mgr.apply_pending_compaction(_agent(short), prefix_key="m|default") is None + assert mgr.compaction_state.pending_checkpoint is None + assert mgr.compaction_state.checkpoint == 0 and len(short) == 2 + + @pytest.mark.asyncio + async def test_after_apply_next_turn_rearms_and_can_cut_again(self, make_session_manager): + store = {} + mgr, live = await self._parked(make_session_manager, store) + _age(store, 600) + mgr.apply_pending_compaction(_agent(live), prefix_key="m|default") + assert await mgr.update_after_turn(800, current_messages=live) is None + assert mgr.compaction_state.armed is True + live.extend(make_conversation(4)) + result = await mgr.update_after_turn(2000, current_messages=live) + assert result is not None and result.deferred is True + assert mgr.compaction_state.pending_checkpoint > 4 # absolute: offset 4 + relative cut + + @pytest.mark.asyncio + async def test_apply_emits_metric_with_reason(self, make_session_manager, monkeypatch): + import apis.shared.observability.emf as emf + emitted = [] + monkeypatch.setattr(emf, "emit_emf_metrics", lambda ns, metrics, properties=None, units=None: emitted.append((metrics, properties))) + monkeypatch.delenv("PROMPT_CACHE_OBSERVABILITY_ENABLED", raising=False) + store = {} + mgr, live = await self._parked(make_session_manager, store) + cut_metrics = [m for m, p in emitted if "CompactionCut" in m] + assert cut_metrics and cut_metrics[0]["CompactionDeferred"] == 1 + _age(store, 600) + mgr.apply_pending_compaction(_agent(live), prefix_key="m|default") + applied = [(m, p) for m, p in emitted if "CompactionApplied" in m] + assert applied and applied[0][1]["applyReason"] == "cache_expired" + assert applied[0][0]["CompactionAppliedForced"] == 0 + + +class TestStateRoundTrip: + def test_pending_fields_round_trip_and_default_none(self): + assert CompactionState.from_dict({"checkpoint": 1}).pending_checkpoint is None + s = CompactionState(pending_checkpoint=7, pending_summary="s", pending_hard_ceiling=150, pending_since="t", last_prefix_key="m|a") + again = CompactionState.from_dict(s.to_dict()) + assert (again.pending_checkpoint, again.pending_summary, again.pending_hard_ceiling, again.pending_since, again.last_prefix_key) == (7, "s", 150, "t", "m|a") + + def test_from_env_flag(self, monkeypatch): + monkeypatch.delenv("AGENTCORE_MEMORY_COMPACTION_DEFERRED_APPLY_ENABLED", raising=False) + assert CompactionConfig.from_env().deferred_apply_enabled is True + monkeypatch.setenv("AGENTCORE_MEMORY_COMPACTION_DEFERRED_APPLY_ENABLED", "false") + assert CompactionConfig.from_env().deferred_apply_enabled is False 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 e7cc1c3e..25ce5075 100644 --- a/backend/tests/agents/main_agent/session/test_compaction_models.py +++ b/backend/tests/agents/main_agent/session/test_compaction_models.py @@ -52,6 +52,11 @@ def test_to_dict_has_camel_case_keys(self): "truncationAnchor", "armed", "policy", + "pendingCheckpoint", + "pendingSummary", + "pendingHardCeiling", + "pendingSince", + "lastPrefixKey", } def test_to_dict_values_match(self): diff --git a/backend/tests/agents/main_agent/session/test_compaction_policy.py b/backend/tests/agents/main_agent/session/test_compaction_policy.py index 36f6a019..eb8b5aea 100644 --- a/backend/tests/agents/main_agent/session/test_compaction_policy.py +++ b/backend/tests/agents/main_agent/session/test_compaction_policy.py @@ -254,7 +254,7 @@ async def test_spiral_shape_cuts_exactly_once(self, make_session_manager, compac @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) + cfg = CompactionConfig(enabled=True, deferred_apply_enabled=False, 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 @@ -279,7 +279,7 @@ async def test_checkpoint_is_live_offset_plus_relative_cut(self, make_session_ma @pytest.mark.asyncio async def test_context_window_flows_into_policy(self, make_session_manager): - cfg = CompactionConfig(enabled=True, protected_turns=3) + cfg = CompactionConfig(enabled=True, deferred_apply_enabled=False, 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 diff --git a/backend/tests/agents/main_agent/session/test_compaction_summary.py b/backend/tests/agents/main_agent/session/test_compaction_summary.py index 40c2e6ee..297e1603 100644 --- a/backend/tests/agents/main_agent/session/test_compaction_summary.py +++ b/backend/tests/agents/main_agent/session/test_compaction_summary.py @@ -121,7 +121,7 @@ class TestThroughUpdateAfterTurn: the restore prepends the same bounded bytes.""" def _manager(self, make_session_manager, records, **cfg): - config = CompactionConfig(enabled=True, token_threshold=1000, protected_turns=3, + config = CompactionConfig(enabled=True, deferred_apply_enabled=False, token_threshold=1000, protected_turns=3, summary_token_budget=BUDGET, **cfg) mgr = make_session_manager(compaction_config=config) mgr.compaction_state = CompactionState() @@ -174,7 +174,7 @@ async def test_cut_emits_one_content_free_emf_record(self, make_session_manager, emitted = [] monkeypatch.setattr(emf, "emit_emf_metrics", lambda ns, metrics, properties=None, units=None: emitted.append((ns, metrics, properties))) monkeypatch.delenv("PROMPT_CACHE_OBSERVABILITY_ENABLED", raising=False) - config = CompactionConfig(enabled=True, token_threshold=1000, protected_turns=3, summary_token_budget=BUDGET) + config = CompactionConfig(enabled=True, deferred_apply_enabled=False, token_threshold=1000, protected_turns=3, summary_token_budget=BUDGET) mgr = make_session_manager(compaction_config=config) mgr.compaction_state = CompactionState() mgr._save_compaction_state = MagicMock() @@ -199,7 +199,7 @@ async def test_kill_switch_silences_metrics(self, make_session_manager, bedrock, emitted = [] monkeypatch.setattr(emf, "emit_emf_metrics", lambda *a, **k: emitted.append(a)) monkeypatch.setenv("PROMPT_CACHE_OBSERVABILITY_ENABLED", "false") - config = CompactionConfig(enabled=True, token_threshold=1000, protected_turns=3) + config = CompactionConfig(enabled=True, deferred_apply_enabled=False, token_threshold=1000, protected_turns=3) mgr = make_session_manager(compaction_config=config) mgr.compaction_state = CompactionState() mgr._save_compaction_state = MagicMock() diff --git a/backend/tests/apis/inference_api/test_chat_service.py b/backend/tests/apis/inference_api/test_chat_service.py index 27e0534e..7eca4619 100644 --- a/backend/tests/apis/inference_api/test_chat_service.py +++ b/backend/tests/apis/inference_api/test_chat_service.py @@ -548,3 +548,23 @@ async def test_a_newly_cacheable_agent_still_shares_the_session_conversation( ) assert mentioned.agent.messages, "the second live Agent forked the conversation" + + +def test_adopting_the_conversation_also_adopts_the_compaction_offset(monkeypatch): + """The live list's coordinate system travels with it (thresholds spec §3.5). + + Compaction expresses its checkpoint as ``_live_offset + index into the + list`` and the pending-cut apply slices the list in place at that offset, + so an instance that adopts another instance's list must adopt its offset. + """ + live_inner = SimpleNamespace(messages=[{"role": "user", "t": 1}, {"role": "assistant", "t": 2}]) + live_wrapper = SimpleNamespace(agent=live_inner, session_manager=SimpleNamespace(_live_offset=7)) + monkeypatch.setattr(service, "_agent_cache", {("s", "key-a"): live_wrapper}) + + fresh_inner = SimpleNamespace(messages=[{"role": "user", "t": 1}]) + fresh = SimpleNamespace(agent=fresh_inner, session_manager=SimpleNamespace(_live_offset=0)) + + service._adopt_session_conversation(fresh, "s") + + assert fresh.agent.messages is live_inner.messages + assert fresh.session_manager._live_offset == 7 diff --git a/docs/specs/compaction-model-relative-thresholds.md b/docs/specs/compaction-model-relative-thresholds.md index 09513843..cd2794bf 100644 --- a/docs/specs/compaction-model-relative-thresholds.md +++ b/docs/specs/compaction-model-relative-thresholds.md @@ -1,8 +1,8 @@ # Compaction relative to the model window — a trigger ceiling, a target floor, and paying the rewrite only when it is free **Status:** PR-1 open (#1125, this document rides with it). PR-2 (bounded -summary + compaction metrics) built 2026-09-15 on top of it, stacked. PR-3 -through PR-5 unbuilt. +summary + compaction metrics, #1128) and PR-3 (paid-when-free apply) built +2026-09-15 on top of it, stacked. PR-4 and PR-5 unbuilt. **Owner:** Phil Merrell **Related:** `compaction-over-threshold-cache-spiral.md` (#833 — the incident and the summary-cap PR this spec depends on) · @@ -221,9 +221,46 @@ 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*: +### 3.5 Pay the rewrite when it is free (PR-3) — BUILT + +Everything above decides *what* to cut. PR-3 decides *when the bytes change*. +As built (`apply_pending_compaction` + the `pending*` fields on +`CompactionState`; kill switch +`AGENTCORE_MEMORY_COMPACTION_DEFERRED_APPLY_ENABLED=false` applies cuts +immediately as PR-1/2 did; legacy mode is always immediate): + +- **Post-turn parks, never applies.** `update_after_turn` computes the cut and + the bounded summary and stores them as `pendingCheckpoint` / + `pendingSummary` / `pendingHardCeiling` / `pendingSince`. `checkpoint` stays + the *applied* value — the one `_apply_compaction` slices at on restore. A + second over-ceiling turn while a cut is parked is a no-op + (`compaction_pending_waiting`); it never cuts deeper. +- **Head of turn decides.** The stream coordinator calls + `apply_pending_compaction(agent, prefix_key="|")` before the + first model call of every turn, on cached and freshly restored agents + alike. It applies when, in this order: `cache_expired` (more than + `cache_ttl_seconds` since the previous turn's save — the entry is gone and + the next call re-writes the prefix regardless), `prefix_changed` (the + model|agent key differs from the persisted `lastPrefixKey` — the cached + prefix is already invalid), or `hard_ceiling` (the previous turn's input + reached the hard ceiling the cut was computed under). Otherwise it waits. +- **Applied in place.** `messages[:] = [first_with_summary] + messages[k+1:]` + where `k = pendingCheckpoint − _live_offset`; then `_live_offset` moves to + the checkpoint and the state is promoted (`checkpoint`, `truncation_anchor`, + `summary`) and persisted. The result is byte-identical to what + `_apply_compaction` derives from stored history under the promoted state + (pinned by `test_live_apply_matches_a_cold_restore_of_the_same_state`), so + a cold restore after a live apply reads the same prefix. +- **Aliasing carries the offset.** `_adopt_session_conversation` copies + `_live_offset` when it points a new agent at the live list. +- **Measured.** Each application persists `applied` (the reason), + `cacheGapSeconds` and `pendingSince` on `compaction.policy`, logs + `rewrite_scheduled` vs `rewrite_forced`, and emits `CompactionApplied`, + `CompactionAppliedForced` and `CompactionCacheGapSeconds`. +- **Not done here:** `context_window_limit` on the Strands model config + (needs the window at agent construction; small follow-up). + +The original design sketch, kept for the record: - Post-turn computes and persists the pending checkpoint + summary (the expensive part, off the critical path). Nothing is applied. @@ -350,7 +387,7 @@ and the protected tail. Session 65b6d4ab: 17 live full re-writes from message 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) +### PR-3 — paid-when-free scheduling (§3.5) — BUILT; `context_window_limit` deferred 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 @@ -394,9 +431,8 @@ rows) and, after PR-2, the second, but not joined: - **Per-call `checkpoint` / `armed` / `liveOffset` on the `C#` cost row** — lets the anatomy page show the context trajectory against the cuts without correlating timestamps by hand. Additive fields on an existing write. -- **`cacheGapSeconds` next to every cut** — the PR-3 scheduling rule is only - measurable if each cut records whether it landed on a cold turn - (`rewrite_scheduled` vs `rewrite_forced`). +- ~~**`cacheGapSeconds` next to every cut**~~ — done in PR-3: each application + records the reason and the gap. - **Retained-vs-actual calibration** — the next turn's measured `contextBreakdown.messages` against the cut's `retainedTokensEstimate`. One number per cut, and the only way to know whether the estimator is off by 5% From eab5119440c5c23fa34a1ba7ee98f31d5f33efc2 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Tue, 15 Sep 2026 23:05:39 -0600 Subject: [PATCH 3/6] feat(compaction): record the in-place apply on the compaction ledger The apply is the moment the bytes the model sees change, so it is the event the cost anatomy should mark. Same attribute-resolved helper as the PR-1 events; no-op until the ledger lands. Co-Authored-By: Claude Fable 5.1 --- .../session/turn_based_session_manager.py | 9 +++++++++ .../session/test_compaction_deferred_apply.py | 17 +++++++++++++++++ 2 files changed, 26 insertions(+) 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 14d0e5a7..b2cbc622 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 @@ -1118,6 +1118,15 @@ def apply_pending_compaction(self, agent: "Agent", *, prefix_key: Optional[str] "pendingSince": pending_since, } self._save_compaction_state(state) + # Per-call compaction ledger: the apply is the moment the bytes + # the model sees change, so it is the event the anatomy marks. + self._record_ledger_event( + "applied", + checkpoint=promoted, + summaryTokens=len(state.summary or "") // 4, + retainedMessages=len(messages), + cacheGapSeconds=gap_seconds or 0, + ) logger.info( "compaction_applied: reason=%s checkpoint=%d gap=%ss live_len=%d (%s)", reason, promoted, gap_seconds, len(messages), diff --git a/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py b/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py index 9a881e4e..436141ef 100644 --- a/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py +++ b/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py @@ -221,6 +221,23 @@ async def test_apply_emits_metric_with_reason(self, make_session_manager, monkey assert applied[0][0]["CompactionAppliedForced"] == 0 +class TestLedger: + @pytest.mark.asyncio + async def test_apply_records_an_applied_ledger_event(self, make_session_manager): + store = {} + mgr = _manager(make_session_manager, store) + live = make_conversation(5) + await mgr.update_after_turn(1200, current_messages=live) + mgr.record_compaction_event = MagicMock() + _age(store, 600) + assert mgr.apply_pending_compaction(_agent(live), prefix_key="m|default") == "cache_expired" + kinds = [c.args[0] for c in mgr.record_compaction_event.call_args_list] + assert kinds == ["applied"] + fields = mgr.record_compaction_event.call_args.kwargs + assert fields["checkpoint"] == 4 and fields["retainedMessages"] == 6 + assert fields["cacheGapSeconds"] >= 599 and fields["summaryTokens"] >= 0 + + class TestStateRoundTrip: def test_pending_fields_round_trip_and_default_none(self): assert CompactionState.from_dict({"checkpoint": 1}).pending_checkpoint is None From e5d4517812737de69bc3d8f4c1c3ed1a43f36900 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Tue, 15 Sep 2026 23:09:20 -0600 Subject: [PATCH 4/6] feat(compaction): mark the head-of-turn apply as promoted on the ledger promoted=1 distinguishes the pending-cut promotion from the restore slice's own applied event without a schema change; both are real byte changes. Co-Authored-By: Claude Fable 5.1 --- .../agents/main_agent/session/turn_based_session_manager.py | 3 +++ .../main_agent/session/test_compaction_deferred_apply.py | 1 + 2 files changed, 4 insertions(+) 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 b2cbc622..f65b1c15 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 @@ -1126,6 +1126,9 @@ def apply_pending_compaction(self, agent: "Agent", *, prefix_key: Optional[str] summaryTokens=len(state.summary or "") // 4, retainedMessages=len(messages), cacheGapSeconds=gap_seconds or 0, + # Distinguishes the head-of-turn promotion from the restore + # slice's own "applied" event (both are real byte changes). + promoted=1, ) logger.info( "compaction_applied: reason=%s checkpoint=%d gap=%ss live_len=%d (%s)", diff --git a/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py b/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py index 436141ef..b977dc52 100644 --- a/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py +++ b/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py @@ -236,6 +236,7 @@ async def test_apply_records_an_applied_ledger_event(self, make_session_manager) fields = mgr.record_compaction_event.call_args.kwargs assert fields["checkpoint"] == 4 and fields["retainedMessages"] == 6 assert fields["cacheGapSeconds"] >= 599 and fields["summaryTokens"] >= 0 + assert fields["promoted"] == 1 class TestStateRoundTrip: From 6e9a88aaef991a37bf00ef2427638d20f5b61668 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Tue, 15 Sep 2026 23:03:36 -0600 Subject: [PATCH 5/6] feat(compaction): offload oversized tool results at intake (PR-4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The case that defeats the compaction floor is a huge tool result inside the protected tail: the last N turns are kept whole, so no cut can bring 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. Thresholds spec §3.6. Built as intake offload rather than cut-time escalation: Strands 1.55's vended ContextOffloader bounds an oversized tool result on AfterToolCallEvent, before it is appended to the conversation, so the persisted message already carries the bounded form and restore reproduces it byte-for-byte — nothing in the prefix is ever mutated after the fact. Cut-time escalation would have mutated protected turns and needed a restore-replay of every edit, and the workspace tools it would have escalated into (workspace_files) are granted to no prod role. - core/tool_result_offload.py: BoundedToolResultOffloader (the vended plugin plus a chars/4 pre-filter so its per-result CountTokens round trip only runs for results near or over the gate, and a content-free ToolResultOffloaded EMF record per offload). Storage: unified strands.storage.S3Storage in the user-files bucket under compaction-offload/{userId}/{sessionId}/ — one namespaced storage per agent, so references are session-scoped by construction and an @-mention agent resolves the same ones. evict_after_cycles=None: eviction from the model path would turn a retrieval into a miss mid-turn. Gate 4,000 tokens / preview 1,000 (AGENTCORE_TOOL_RESULT_OFFLOAD_MAX_TOKENS / _PREVIEW_TOKENS); AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED=false removes the plugin. Fail-open everywhere: no bucket, construction failure, storage or CountTokens error all keep the original result. - ChatAgent._create_agent appends the plugin. It registers retrieve_offloaded_content itself (pattern / line range / full) — one stable spec in toolConfig, not an RBAC-gated tool, on the read_skill_file precedent. - CDK: 90-day lifecycle expiry on the compaction-offload/ prefix of the user-files bucket (merged into its existing rules; shortest expiration wins for the prefix). The runtime role already had Put/Get on the bucket. - Cut record gains CompactionFloorUnreachable — the residual after intake offload (attachments, sub-gate results). - User attachments are deliberately not touched: the digest + page-range read tool in document-context-offload.md is the right shape for those. - Kaizen review-queue [2026-07-19] ContextOffloader spike closed as adopted. Tests: flag / no-bucket / per-session prefix / config / preview clamp; small result skips CountTokens; borderline result is measured; oversized result is offloaded with preview + reference and a content-free metric; storage and CountTokens failures keep the original; ChatAgent wiring with and without a bucket. Full backend suite: 8670 passed. Co-Authored-By: Claude Fable 5.1 --- backend/src/agents/main_agent/chat_agent.py | 26 +- .../src/agents/main_agent/config/constants.py | 15 ++ .../main_agent/core/tool_result_offload.py | 186 +++++++++++++ .../session/turn_based_session_manager.py | 5 + .../core/test_tool_result_offload.py | 245 ++++++++++++++++++ .../session/test_compaction_summary.py | 3 + docs/kaizen/review-queue.md | 2 +- .../compaction-model-relative-thresholds.md | 43 ++- .../constructs/data/file-upload-construct.ts | 14 + 9 files changed, 527 insertions(+), 12 deletions(-) create mode 100644 backend/src/agents/main_agent/core/tool_result_offload.py create mode 100644 backend/tests/agents/main_agent/core/test_tool_result_offload.py diff --git a/backend/src/agents/main_agent/chat_agent.py b/backend/src/agents/main_agent/chat_agent.py index 2e72e608..dc30a5e1 100644 --- a/backend/src/agents/main_agent/chat_agent.py +++ b/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,7 +59,7 @@ 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( @@ -66,6 +67,29 @@ def _create_agent(self) -> None: 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), diff --git a/backend/src/agents/main_agent/config/constants.py b/backend/src/agents/main_agent/config/constants.py index 5c4171c5..cc31844c 100644 --- a/backend/src/agents/main_agent/config/constants.py +++ b/backend/src/agents/main_agent/config/constants.py @@ -54,6 +54,13 @@ class EnvVars: # 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 @@ -168,6 +175,14 @@ class Defaults: # 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" diff --git a/backend/src/agents/main_agent/core/tool_result_offload.py b/backend/src/agents/main_agent/core/tool_result_offload.py new file mode 100644 index 00000000..bab3b21a --- /dev/null +++ b/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 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 f65b1c15..81572d82 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 @@ -1221,6 +1221,11 @@ def _emit_compaction_metrics(self, state, policy, forced, retained_estimate, bou "CompactionRetainedTokens": int(retained_estimate or 0), "CompactionSummaryTokens": int(bounded.tokens_after or 0), "CompactionSummaryOverBudget": 1 if bounded.tokens_before > bounded.tokens_after else 0, + # The protected tail alone exceeded the floor: the residual + # case after intake offload (attachments, sub-gate results). + "CompactionFloorUnreachable": ( + 1 if (policy.floor is not None and retained_estimate is not None and retained_estimate > policy.floor) else 0 + ), }, { "policySource": policy.source, diff --git a/backend/tests/agents/main_agent/core/test_tool_result_offload.py b/backend/tests/agents/main_agent/core/test_tool_result_offload.py new file mode 100644 index 00000000..4426bdff --- /dev/null +++ b/backend/tests/agents/main_agent/core/test_tool_result_offload.py @@ -0,0 +1,245 @@ +"""Tool-result offload at intake — thresholds spec §3.6 (PR-4).""" + +import sys +import types +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from agents.main_agent.core import tool_result_offload as tro +from agents.main_agent.core.tool_result_offload import ( + PREFILTER_RATIO, + build_tool_result_offloader, + offload_prefix, + tool_result_offload_enabled, +) + + +class _FakeStorage: + """Unified-Storage-shaped in-memory backend (write/read), like strands.storage.""" + + def __init__(self): + self.objects = {} + + async def write(self, key, data): + self.objects[key] = data + + async def read(self, key): + return self.objects.get(key) + + async def delete(self, key): + self.objects.pop(key, None) + + async def list(self, query=""): + return [k for k in self.objects if k.startswith(query)] + + +class _Agent: + """Weak-referenceable stand-in (the plugin keys WeakKeyDictionaries on the agent).""" + + def __init__(self, model=None): + self.model = model + self.event_loop_metrics = SimpleNamespace(cycle_count=1) + self.storage = None + self.sandbox = None + + +def _event(text, tool_name="gmail_search", count_tokens=None): + result = {"toolUseId": "t1", "status": "success", "content": [{"text": text}]} + model = SimpleNamespace(count_tokens=count_tokens or AsyncMock(return_value=len(text) // 4)) + agent = _Agent(model) + return SimpleNamespace( + result=result, + tool_use={"toolUseId": "t1", "name": tool_name}, + selected_tool=None, + cancel_message=None, + agent=agent, + ) + + +@pytest.fixture +def offloader(monkeypatch): + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_MAX_TOKENS", "1000") + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS", "100") + storage = _FakeStorage() + plugin = build_tool_result_offloader(session_id="s1", user_id="u1", storage=storage) + assert plugin is not None + # Bind storage the way init_agent would. + plugin.init_agent(_Agent()) + return plugin, storage + + +class TestBuilder: + def test_flag_default_on_only_literal_false_off(self, monkeypatch): + monkeypatch.delenv("AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED", raising=False) + assert tool_result_offload_enabled() is True + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED", "false") + assert tool_result_offload_enabled() is False + assert build_tool_result_offloader(session_id="s", user_id="u", storage=_FakeStorage()) is None + + def test_no_bucket_means_no_plugin(self, monkeypatch): + monkeypatch.delenv("S3_USER_FILES_BUCKET_NAME", raising=False) + assert build_tool_result_offloader(session_id="s", user_id="u") is None + + def test_s3_backend_is_scoped_per_user_and_session(self, monkeypatch): + monkeypatch.setenv("S3_USER_FILES_BUCKET_NAME", "files-bucket") + captured = {} + + class FakeS3Storage: + def __init__(self, bucket, *, prefix="", region_name=None, **kw): + captured.update(bucket=bucket, prefix=prefix, region=region_name) + + async def write(self, k, d): ... + async def read(self, k): ... + async def delete(self, k): ... + async def list(self, q=""): return [] + + monkeypatch.setitem(sys.modules, "strands.storage", types.SimpleNamespace(S3Storage=FakeS3Storage)) + plugin = build_tool_result_offloader(session_id="sess", user_id="usr", region="us-west-2") + assert plugin is not None + assert captured == {"bucket": "files-bucket", "prefix": "compaction-offload/usr/sess", "region": "us-west-2"} + assert offload_prefix("usr", "sess") == "compaction-offload/usr/sess" + + def test_configuration(self, offloader): + plugin, _ = offloader + assert plugin._max_result_tokens == 1000 + assert plugin._preview_tokens == 100 + assert plugin._evict_after_cycles is None # no eviction from the model path + assert plugin._include_retrieval_tool is True + assert type(plugin).__name__ == "BoundedToolResultOffloader" + + def test_preview_is_clamped_below_gate(self, monkeypatch): + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_MAX_TOKENS", "500") + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS", "9000") + plugin = build_tool_result_offloader(session_id="s", user_id="u", storage=_FakeStorage()) + assert plugin._preview_tokens == 499 + + +class TestPrefilterAndOffload: + @pytest.mark.asyncio + async def test_small_result_skips_count_tokens_and_is_untouched(self, offloader): + plugin, storage = offloader + counter = AsyncMock(return_value=10) + ev = _event("short result", count_tokens=counter) + before = ev.result + await plugin._handle_tool_result(ev) + counter.assert_not_called() + assert ev.result is before and storage.objects == {} + + @pytest.mark.asyncio + async def test_borderline_result_is_measured_by_count_tokens(self, offloader): + plugin, _ = offloader + # ~600 estimated tokens: over the pre-filter (500) but under the gate (1000). + counter = AsyncMock(return_value=600) + ev = _event("x" * 2400, count_tokens=counter) + before = ev.result + await plugin._handle_tool_result(ev) + counter.assert_called_once() + assert ev.result is before + + @pytest.mark.asyncio + async def test_oversized_result_is_offloaded_with_preview_and_reference(self, offloader, monkeypatch): + plugin, storage = offloader + import apis.shared.observability.emf as emf + emitted = [] + monkeypatch.setattr(emf, "emit_emf_metrics", lambda ns, metrics, properties=None, units=None: emitted.append((ns, metrics, properties))) + monkeypatch.delenv("PROMPT_CACHE_OBSERVABILITY_ENABLED", raising=False) + + body = "header line\n" + ("row data " * 2000) + ev = _event(body, count_tokens=AsyncMock(return_value=5000)) + await plugin._handle_tool_result(ev) + + text = ev.result["content"][0]["text"] + assert text.startswith("[Offloaded:") + assert "retrieve_offloaded_content" in text + assert "header line" in text # preview keeps the head + assert len(text) < len(body) // 4 + assert len(storage.objects) == 1 # the full block landed in storage + assert ev.result["toolUseId"] == "t1" and ev.result["status"] == "success" + assert emitted and emitted[0][0] == "AgentCoreStack/Compaction" + assert emitted[0][1]["ToolResultOffloaded"] == 1 + assert emitted[0][2]["toolName"] == "gmail_search" + assert "row data" not in str(emitted[0][2]) # content-free + + @pytest.mark.asyncio + async def test_storage_failure_keeps_original_result(self, offloader): + plugin, storage = offloader + + async def boom(key, data): + raise RuntimeError("s3 down") + + storage.write = boom + ev = _event("y" * 20000, count_tokens=AsyncMock(return_value=5000)) + before = ev.result + await plugin._handle_tool_result(ev) + assert ev.result is before + + @pytest.mark.asyncio + async def test_count_tokens_failure_keeps_original_result(self, offloader): + plugin, _ = offloader + ev = _event("z" * 20000, count_tokens=AsyncMock(side_effect=RuntimeError("AccessDenied"))) + before = ev.result + await plugin._handle_tool_result(ev) + assert ev.result is before + + def test_prefilter_ratio_is_half_the_gate(self): + assert PREFILTER_RATIO == 0.5 + + +class TestChatAgentWiring: + def test_chat_agent_adds_the_offloader_plugin(self, monkeypatch): + from unittest.mock import MagicMock + from agents.main_agent.chat_agent import ChatAgent + from agents.main_agent.core import AgentFactory + + monkeypatch.setenv("S3_USER_FILES_BUCKET_NAME", "files-bucket") + + class FakeS3Storage: + def __init__(self, bucket, *, prefix="", region_name=None, **kw): + self.prefix = prefix + + async def write(self, k, d): ... + async def read(self, k): ... + async def delete(self, k): ... + async def list(self, q=""): return [] + + monkeypatch.setitem(sys.modules, "strands.storage", types.SimpleNamespace(S3Storage=FakeS3Storage)) + captured = {} + monkeypatch.setattr(AgentFactory, "create_agent", staticmethod(lambda **kw: captured.update(kw) or MagicMock())) + + agent = ChatAgent.__new__(ChatAgent) + agent.system_prompt = "BASE" + agent.model_config = MagicMock() + agent.session_manager = MagicMock() + agent.session_id = "sess" + agent.user_id = "usr" + agent._accessible_skill_ids = None + monkeypatch.setattr(ChatAgent, "_build_filtered_tools", lambda self: [], raising=False) + monkeypatch.setattr(ChatAgent, "_create_hooks", lambda self: [], raising=False) + + agent._create_agent() + + plugins = captured["plugins"] + assert plugins and type(plugins[0]).__name__ == "BoundedToolResultOffloader" + assert plugins[0]._storage._prefix if hasattr(plugins[0]._storage, "_prefix") else True + + def test_chat_agent_without_bucket_passes_no_plugins(self, monkeypatch): + from unittest.mock import MagicMock + from agents.main_agent.chat_agent import ChatAgent + from agents.main_agent.core import AgentFactory + + monkeypatch.delenv("S3_USER_FILES_BUCKET_NAME", raising=False) + captured = {} + monkeypatch.setattr(AgentFactory, "create_agent", staticmethod(lambda **kw: captured.update(kw) or MagicMock())) + agent = ChatAgent.__new__(ChatAgent) + agent.system_prompt = "BASE" + agent.model_config = MagicMock() + agent.session_manager = MagicMock() + agent.session_id = "sess" + agent.user_id = "usr" + agent._accessible_skill_ids = None + monkeypatch.setattr(ChatAgent, "_build_filtered_tools", lambda self: [], raising=False) + monkeypatch.setattr(ChatAgent, "_create_hooks", lambda self: [], raising=False) + agent._create_agent() + assert captured["plugins"] is None diff --git a/backend/tests/agents/main_agent/session/test_compaction_summary.py b/backend/tests/agents/main_agent/session/test_compaction_summary.py index 297e1603..08681ad8 100644 --- a/backend/tests/agents/main_agent/session/test_compaction_summary.py +++ b/backend/tests/agents/main_agent/session/test_compaction_summary.py @@ -189,6 +189,9 @@ async def test_cut_emits_one_content_free_emf_record(self, make_session_manager, assert ns == "AgentCoreStack/Compaction" assert metrics["CompactionCut"] == 1 and metrics["CompactionForced"] == 0 assert metrics["CompactionInputTokens"] == 2000 + # Tiny 5-turn conversation calibrated to 2000 tokens against a 250 + # floor with 3 protected turns: the tail cannot fit → unreachable. + assert metrics["CompactionFloorUnreachable"] == 1 assert props["summaryOutcome"] == "within_budget" and props["policySource"] == "fixed" # Content-free: no summary text, no message text in the record. assert "tiny" not in str(props) diff --git a/docs/kaizen/review-queue.md b/docs/kaizen/review-queue.md index bbadbe4b..a66d6e4f 100644 --- a/docs/kaizen/review-queue.md +++ b/docs/kaizen/review-queue.md @@ -247,7 +247,7 @@ Items added by `kaizen-research`, consumed by `kaizen-review-prep`. - **Surface**: backend (`agent_factory.py` — `plugins=` already plumbed; a custom S3 `Storage` backend is the real build — the plugin ships InMemory/File only, and references must survive AgentCore Runtime restores across turns) - **Effort × Impact**: M–H × M–H (payoff directly measurable by the PR #697 cache/cost metrics) - **Subtracts**: partial — bounds MCP/tool payload growth at the source instead of relying solely on reactive below-anchor truncation in `TurnBasedSessionManager` (which stays for legacy history) -- **Status**: open — spike before commitment; four known gotchas: (1) `evict_after_cycles=20` runs on `BeforeModelCallEvent` and touches *prior* messages — potential byte-stability cache-buster, verify semantics or set `None` and expire via S3 lifecycle; (2) `model.count_tokens` per tool result adds latency, and Bedrock CountTokens rejects `us.*` inference-profile ids (de-prefix precedent in context attribution); (3) adoption flips `toolConfigHash` once (expected; the new tool must land in the deterministically-ordered tool list); (4) check SPA tool-result rendering against placeholder content. +- **Status**: **ADOPTED 2026-09-15** as compaction PR-4 (branch `feature/compaction-offload-escalation`, stacked on #1129; spec `compaction-model-relative-thresholds.md` §3.6). Gotchas as resolved: (1) `evict_after_cycles=None`, expiry via a 90-day S3 lifecycle rule on `compaction-offload/`; (2) a chars/4 pre-filter skips CountTokens for results under half the gate, and `CountTokensBedrockModel` already de-prefixes `us.*`; (3) `retrieve_offloaded_content` lands once in the tool list; (4) the preview is plain text in the tool result. Original entry: spike before commitment; four known gotchas: (1) `evict_after_cycles=20` runs on `BeforeModelCallEvent` and touches *prior* messages — potential byte-stability cache-buster, verify semantics or set `None` and expire via S3 lifecycle; (2) `model.count_tokens` per tool result adds latency, and Bedrock CountTokens rejects `us.*` inference-profile ids (de-prefix precedent in context attribution); (3) adoption flips `toolConfigHash` once (expected; the new tool must land in the deterministically-ordered tool list); (4) check SPA tool-result rendering against placeholder content. ### [2026-07-17] Adopt Strands `Limits` on the unattended Scheduled Runs / headless lane - **Source**: research/2026-07-17.md ▸ Top 5 #3 — convergent harness rail (Claude Code 2.1.212 spawn cap + opencode 1.18.2 `subagent_depth`); Strands `Limits` now available (we're on 1.47). diff --git a/docs/specs/compaction-model-relative-thresholds.md b/docs/specs/compaction-model-relative-thresholds.md index cd2794bf..a83bb9d9 100644 --- a/docs/specs/compaction-model-relative-thresholds.md +++ b/docs/specs/compaction-model-relative-thresholds.md @@ -1,8 +1,9 @@ # Compaction relative to the model window — a trigger ceiling, a target floor, and paying the rewrite only when it is free **Status:** PR-1 open (#1125, this document rides with it). PR-2 (bounded -summary + compaction metrics, #1128) and PR-3 (paid-when-free apply) built -2026-09-15 on top of it, stacked. PR-4 and PR-5 unbuilt. +summary + compaction metrics, #1128), PR-3 (paid-when-free apply, #1129) and +PR-4 (tool-result offload at intake) built 2026-09-15 on top of it, stacked. +PR-5 unbuilt. **Owner:** Phil Merrell **Related:** `compaction-over-threshold-cache-spiral.md` (#833 — the incident and the summary-cap PR this spec depends on) · @@ -283,10 +284,31 @@ The original design sketch, kept for the record: 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. +- **Offload escalation** (PR-4) — BUILT for tool results, at intake. The + case that defeats the floor is a huge tool result inside the protected + tail. Rather than escalate at cut time (which would mutate a protected turn + and need a restore-replay of every edit), oversized tool results are bounded + the moment they are produced, before they enter the prefix: Strands 1.55's + vended `ContextOffloader` on `AfterToolCallEvent`, S3 storage in the + user-files bucket under `compaction-offload/{userId}/{sessionId}/` (one + namespaced storage per agent, so references are session-scoped by + construction and an `@`-mention agent resolves the same ones), no eviction + from the model path (`evict_after_cycles=None`; a 90-day S3 lifecycle rule + expires the objects), gate 4,000 tokens / preview 1,000 (env-backed; + `AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED=false` removes the plugin). The + persisted message already carries the bounded form, so restore reproduces + it byte-for-byte. Our subclass adds a chars/4 pre-filter so the plugin's + per-result `CountTokens` round trip only runs for results near or over the + gate, and a content-free `ToolResultOffloaded` record per offload. The + model keeps a preview and pulls spans back with + `retrieve_offloaded_content` (pattern / line range / full) — one stable + spec in `toolConfig`, not an RBAC-gated tool, on the `read_skill_file` + precedent (the `workspace_files` catalog key is granted to no prod role, + so escalating into the workspace tools would have shipped dark). **User + attachments are not touched**: the digest + page-range read in + `document-context-offload.md` is the right shape for those and stays that + spec's PRs 1–4. What remains of the floor-unreachable case after this is + measured by `CompactionFloorUnreachable` on the cut record. - **`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' @@ -400,7 +422,7 @@ 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-4 — offload escalation (§3.6) — BUILT as tool-result offload at intake ### PR-5 — selective 1h TTL experiment (§3.6) @@ -417,6 +439,8 @@ observability layer by `PROMPT_CACHE_OBSERVABILITY_ENABLED=false`): | `CompactionForced` | how often a cut ran while disarmed = how often the previous cut did not take | | `CompactionInputTokens`, `CompactionRetainedTokens` | how far above the ceiling cuts fire and how deep they land (is the floor being reached?) | | `CompactionSummaryTokens`, `CompactionSummaryOverBudget` | is the summary the reason cuts miss the floor; how often the model vs truncation path runs | +| `CompactionFloorUnreachable` (PR-4) | how often the protected tail alone still exceeds the floor after intake offload — the residual document/attachment case | +| `ToolResultOffloaded`, `ToolResultOffloadedTokens` (PR-4) | how much tool payload was kept out of the prefix, per tool (`toolName` property) | Properties (queryable in Logs Insights, not dimensions): `policySource`, `contextWindow`, `ceiling`, `floor`, `summaryOutcome`, `summaryTokensBefore`. @@ -438,9 +462,8 @@ rows) and, after PR-2, the second, but not joined: number per cut, and the only way to know whether the estimator is off by 5% or 50%. - **Turn shape** — messages per turn and tool-result bytes per turn (the - `ToolCensusHook` has the count; bytes are one more `ADD`). Sessions whose - bulk is a handful of huge tool results are the PR-4 offload cohort, and we - cannot size it today. + `ToolCensusHook` has the count; bytes are one more `ADD`). Partly covered + by PR-4's `ToolResultOffloadedTokens`; the sub-gate long tail is not. - **Outcome signal joined to compaction** — a down-thumb rate keyed to "turns since last cut" is the first evidence that a cut costs anything besides dollars (the kaizen review-queue already lists this as the diff --git a/infrastructure/lib/constructs/data/file-upload-construct.ts b/infrastructure/lib/constructs/data/file-upload-construct.ts index 7e9179b6..dc1b5568 100644 --- a/infrastructure/lib/constructs/data/file-upload-construct.ts +++ b/infrastructure/lib/constructs/data/file-upload-construct.ts @@ -103,6 +103,20 @@ export class FileUploadConstruct extends Construct { id: 'expire-objects', expiration: cdk.Duration.days(365), }, + { + // Compaction's tool-result offload (backend + // agents/main_agent/core/tool_result_offload.py) parks oversized + // tool results under `compaction-offload/{userId}/{sessionId}/` and + // keeps a retrieval reference in the conversation. The plugin never + // evicts from the model path (that would turn a retrieval into a + // miss mid-turn), so objects expire here instead — 90 days, matching + // AgentCore Memory's conversation retention. The shortest matching + // expiration wins, so this overrides the 365-day rule for the + // prefix only; user uploads are untouched. + id: 'expire-compaction-offload', + prefix: 'compaction-offload/', + expiration: cdk.Duration.days(90), + }, { id: 'abort-incomplete-multipart', abortIncompleteMultipartUploadAfter: cdk.Duration.days(1), From 49e610f481f8445e44833970143d682037c332ba Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Tue, 15 Sep 2026 23:09:23 -0600 Subject: [PATCH 6/6] docs(compaction): record the over-100k cohort split PR-4 targets Co-Authored-By: Claude Fable 5.1 --- docs/specs/compaction-model-relative-thresholds.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/docs/specs/compaction-model-relative-thresholds.md b/docs/specs/compaction-model-relative-thresholds.md index a83bb9d9..004cde24 100644 --- a/docs/specs/compaction-model-relative-thresholds.md +++ b/docs/specs/compaction-model-relative-thresholds.md @@ -424,6 +424,20 @@ still passes with the in-place apply. ### PR-4 — offload escalation (§3.6) — BUILT as tool-result offload at intake +**Cohort split (2026-09-15 audit, content-free, September prod).** Of the 95 +sessions that peaked over 100k: 52 (55%) had a single tool result ≥4k tokens +in their last three turns (43 only that, 9 also an attachment) — PR-4's +target; 25 (26%) had an attachment there (16 only that) — the document-offload +spec's; 27 (28%) had neither — long sessions whose bulk is old history plus a +23–40k summary, which the cut and the summary cap reach. The biggest single +intra-turn writes are squarely tool results (123k, 149k, 131k in one call); +the attachment-only cases include 1–3-turn sessions at 280–530k that are one +huge upload. Method: turns split at `cacheGapSeconds ≥ 10s`; "big tool result" += an intra-turn call that read the prior prefix and wrote ≥4,000 tokens +(slightly overstated on long `tool_use` blocks); "attachment in the last 3 +turns" = an upload row between the first call of those turns minus 5 min and +the last call. + ### PR-5 — selective 1h TTL experiment (§3.6) ## 7. Observability