From 667d43c45acfbd57c7dbbaa183b82a50b23b5e74 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Tue, 15 Sep 2026 20:27:12 -0600 Subject: [PATCH 1/4] 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/4] 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/4] 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/4] 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: