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).