diff --git a/backend/src/agents/main_agent/config/constants.py b/backend/src/agents/main_agent/config/constants.py index d9ed82a7..5c4171c5 100644 --- a/backend/src/agents/main_agent/config/constants.py +++ b/backend/src/agents/main_agent/config/constants.py @@ -43,6 +43,17 @@ class EnvVars: # (a prefix re-write per turn, and it moves the coordinates the compaction # checkpoint is expressed in). Setting this to 40 restores the SDK default. CONVERSATION_WINDOW_MESSAGES = "AGENTCORE_CONVERSATION_WINDOW_MESSAGES" + # Bounded compaction summary (spiral spec PR-2 / + # compaction-model-relative-thresholds.md §3.6). Budget in tokens; the + # re-summarize call is a Nova Micro side-channel with its own kill switch. + COMPACTION_SUMMARY_TOKEN_BUDGET = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_TOKEN_BUDGET" + COMPACTION_SUMMARY_MODEL_ENABLED = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ENABLED" + COMPACTION_SUMMARY_MODEL_ID = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ID" + # Paid-when-free scheduling (thresholds spec §3.5): a cut is computed and + # persisted as PENDING post-turn and applied to the live list pre-call only + # when the prefix re-write is free (cache expired, model/agent switched) + # or unavoidable (hard ceiling). "false" applies cuts immediately (PR-1/2). + COMPACTION_DEFERRED_APPLY_ENABLED = "AGENTCORE_MEMORY_COMPACTION_DEFERRED_APPLY_ENABLED" # --- Restored-history repair --- # Kill switch for the restore-time tool-pairing/alternation repair @@ -148,6 +159,15 @@ class Defaults: # prompt. Overflow recovery (reduce_context on ContextWindowOverflow) still # works at any window size. CONVERSATION_WINDOW_MESSAGES = 2000 + # 8k tokens ≈ 32k chars: a third of the 25k floor, so a bounded summary + # can never by itself hold a session above the ceiling (the incident's + # summary was 40k tokens against a 100k threshold). Same figure the admin + # SUMMARY_OVER_BUDGET diagnosis reads. + COMPACTION_SUMMARY_TOKEN_BUDGET = 8_000 + COMPACTION_SUMMARY_MODEL_ENABLED = True + # Same cheap model as the title and tool-batch side-channels. + COMPACTION_SUMMARY_MODEL_ID = "us.amazon.nova-micro-v1:0" + COMPACTION_DEFERRED_APPLY_ENABLED = True # --- 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..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: @@ -142,6 +171,15 @@ class CompactionConfig: floor_ratio: float = Defaults.COMPACTION_FLOOR_RATIO hard_ceiling_ratio: float = Defaults.COMPACTION_HARD_CEILING_RATIO hard_ceiling_multiplier: float = Defaults.COMPACTION_HARD_CEILING_MULTIPLIER + # Bounded summary (spec §3.6 / spiral spec PR-2). The persisted summary is + # held at or under this many tokens (chars/4), compressed once at cut time + # by the cheap model, with newest-first truncation as the fallback. + summary_token_budget: int = Defaults.COMPACTION_SUMMARY_TOKEN_BUDGET + summary_model_enabled: bool = Defaults.COMPACTION_SUMMARY_MODEL_ENABLED + summary_model_id: str = Defaults.COMPACTION_SUMMARY_MODEL_ID + # Paid-when-free scheduling (spec §3.5). Only meaningful with the + # model-relative policy on; legacy mode always applies immediately. + deferred_apply_enabled: bool = Defaults.COMPACTION_DEFERRED_APPLY_ENABLED @classmethod def from_env(cls) -> "CompactionConfig": @@ -158,4 +196,8 @@ def from_env(cls) -> "CompactionConfig": floor_ratio=float(os.environ.get(EnvVars.COMPACTION_FLOOR_RATIO, str(Defaults.COMPACTION_FLOOR_RATIO))), hard_ceiling_ratio=float(os.environ.get(EnvVars.COMPACTION_HARD_CEILING_RATIO, str(Defaults.COMPACTION_HARD_CEILING_RATIO))), hard_ceiling_multiplier=float(os.environ.get(EnvVars.COMPACTION_HARD_CEILING_MULTIPLIER, str(Defaults.COMPACTION_HARD_CEILING_MULTIPLIER))), + summary_token_budget=int(os.environ.get(EnvVars.COMPACTION_SUMMARY_TOKEN_BUDGET, str(Defaults.COMPACTION_SUMMARY_TOKEN_BUDGET))), + summary_model_enabled=_env_flag_default_on(EnvVars.COMPACTION_SUMMARY_MODEL_ENABLED), + summary_model_id=os.environ.get(EnvVars.COMPACTION_SUMMARY_MODEL_ID, "").strip() or Defaults.COMPACTION_SUMMARY_MODEL_ID, + deferred_apply_enabled=_env_flag_default_on(EnvVars.COMPACTION_DEFERRED_APPLY_ENABLED), ) 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..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 @@ -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 @@ -114,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 @@ -803,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) @@ -821,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 @@ -935,20 +951,44 @@ 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) - - 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 + 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 + + 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 @@ -959,9 +999,18 @@ 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, + "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, deferred) logger.info( f"Compaction checkpoint set: {new_checkpoint}, " @@ -981,6 +1030,211 @@ 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) + # 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, + # 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)", + 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 _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 + from apis.shared.observability.emf import emit_emf_metrics + + if not prompt_cache_observability_enabled(): + return + 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", + }, ) # ========================================================================= 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..b977dc52 --- /dev/null +++ b/backend/tests/agents/main_agent/session/test_compaction_deferred_apply.py @@ -0,0 +1,253 @@ +"""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 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 + assert fields["promoted"] == 1 + + +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 new file mode 100644 index 00000000..297e1603 --- /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, 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() + 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, 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() + 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, 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() + 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/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 31eb4d7f..cd2794bf 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, #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) · @@ -220,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. @@ -321,7 +359,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` @@ -334,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 @@ -353,6 +406,51 @@ 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**~~ — 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% + 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).