diff --git a/backend/scripts/probe_static_prefix_ttl.py b/backend/scripts/probe_static_prefix_ttl.py new file mode 100644 index 000000000..27b313c0e --- /dev/null +++ b/backend/scripts/probe_static_prefix_ttl.py @@ -0,0 +1,169 @@ +"""Does a 1h TTL on the tools + system cachePoints pay for itself on Claude/Bedrock? + +The gate for AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL=1h +(docs/specs/compaction-model-relative-thresholds.md §3.6, PR-5). A caching +default must never be adopted on inspection alone — #954 shipped on a wrong +premise and measured 57% more expensive live before #956 reverted it — so this +script measures the two arms against the same static prefix: + + arm 5m : tools + system points with no ttl (today's shape) + arm 1h : tools + system points with ttl "1h"; message point unchanged + +Per arm: call 1 (the write), sleep --gap-seconds, call 2 (the read-or-rewrite). +It reports cacheRead / cacheWrite per call, whether Bedrock accepted the 1h +point at all, and prices the pair at the model's own rates — 1.25x base for a +5m write, 2x base for a 1h write, 0.1x for a read — so the break-even is read +off the output rather than argued. + +What "pays" means: with a gap between 5 and 60 minutes, the 1h arm's second +call should READ the static prefix (cacheRead ≈ static tokens) where the 5m +arm re-WRITES it. Over a session the 1h arm costs +0.75x base on every static +write and saves 1.15x base on every cold-within-the-hour return; it pays when +the second event is more frequent than the first. Run this with a gap of +~420s (past 5m) and again with ~60s (inside 5m) to see both regimes. + +Read-only apart from the model invocations. ⚠️ Real spend: four calls at +~8k input tokens each plus the sleep. Nothing is written anywhere. + +Usage: + cd backend + AWS_PROFILE=dev-ai uv run python scripts/probe_static_prefix_ttl.py --gap-seconds 420 + AWS_PROFILE=dev-ai uv run python scripts/probe_static_prefix_ttl.py --gap-seconds 60 --model-id us.anthropic.claude-haiku-4-5-20251001-v1:0 + +Baseline, dev-ai us-west-2, 2026-09-16, Haiku 4.5, gap 420s: + 5m first read 0 write 6251 second read 0 write 6251 + 1h first read 0 write 6251 second read 5924 write 327 <- honored + pair $0.017197 (5m) vs $0.015130 (1h): 1h CHEAPER by 12% at this gap +Same day, gap 60s (both arms warm): + 5m second read 6251 write 0 ; 1h second read 6251 write 0 + pair $0.009289 (5m) vs $0.014446 (1h): 1h MORE EXPENSIVE by $0.005157 + = the 0.75x-base premium on the first write, nothing to recover inside 5m +""" + +from __future__ import annotations + +import argparse +import sys +import time +from typing import Any, Dict, List + +import boto3 + +# Clears every Claude family's cache minimum (4,096 on Haiku) with margin. +_SYSTEM_TEXT = ("You are a careful assistant. " * 40 + "Policy: answer briefly. ") * 20 +_TOOL_SPECS: List[Dict[str, Any]] = [ + { + "toolSpec": { + "name": f"tool_{i}", + "description": "A probe tool that does nothing useful. " * 12, + "inputSchema": {"json": {"type": "object", "properties": {"q": {"type": "string"}}}}, + } + } + for i in range(6) +] + +RATES = { # $/MTok base input, Global CRIS; adjust for the model under test + "us.anthropic.claude-haiku-4-5-20251001-v1:0": 1.10, + "global.anthropic.claude-sonnet-5": 2.00, + "us.anthropic.claude-sonnet-4-6": 3.30, +} + + +def _request(model_id: str, ttl: str | None, marker: str) -> Dict[str, Any]: + point: Dict[str, Any] = {"type": "default"} + if ttl: + point["ttl"] = ttl + return { + "modelId": model_id, + "system": [{"text": _SYSTEM_TEXT + f"\nProbe arm: {marker}."}, {"cachePoint": dict(point)}], + "toolConfig": {"tools": _TOOL_SPECS + [{"cachePoint": dict(point)}]}, + "messages": [{"role": "user", "content": [{"text": "Reply with the single word OK."}, {"cachePoint": {"type": "default"}}]}], + "inferenceConfig": {"maxTokens": 5}, + } + + +def _call(client: Any, req: Dict[str, Any]) -> Dict[str, int]: + resp = client.converse(**req) + u = resp.get("usage", {}) + return { + "input": int(u.get("inputTokens", 0)), + "read": int(u.get("cacheReadInputTokens", 0)), + "write": int(u.get("cacheWriteInputTokens", 0)), + } + + +def _price(usage: Dict[str, int], base: float, write_mult: float) -> float: + return (usage["input"] * base + usage["read"] * base * 0.1 + usage["write"] * base * write_mult) / 1e6 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--model-id", default="us.anthropic.claude-haiku-4-5-20251001-v1:0") + ap.add_argument("--region", default="us-west-2") + ap.add_argument("--gap-seconds", type=int, default=420) + ap.add_argument("--base-rate", type=float, default=None, help="$/MTok base input; defaults from a small table") + args = ap.parse_args() + + base = args.base_rate or RATES.get(args.model_id) + if base is None: + print(f"no base rate for {args.model_id}; pass --base-rate", file=sys.stderr) + return 2 + client = boto3.client("bedrock-runtime", region_name=args.region) + + arms = [("5m", None), ("1h", "1h")] + results: Dict[str, Any] = {} + # Distinct markers per arm so the two arms never share a cache entry. + for name, ttl in arms: + marker = f"{name}-{int(time.time())}" + req = _request(args.model_id, ttl, marker) + try: + first = _call(client, req) + except Exception as e: # noqa: BLE001 + results[name] = {"error": f"{type(e).__name__}: {e}"} + continue + results[name] = {"first": first, "marker": marker} + if all("error" in r for r in results.values()): + print(results) + return 1 + + print(f"sleeping {args.gap_seconds}s so the 5m entry {'expires' if args.gap_seconds > 300 else 'stays warm'} ...") + time.sleep(args.gap_seconds) + + for name, ttl in arms: + if "error" in results[name]: + continue + # Re-send the SAME request (same marker) — identical bytes. + try: + results[name]["second"] = _call(client, _request(args.model_id, ttl, results[name]["marker"])) + except Exception as e: # noqa: BLE001 + results[name]["error"] = f"{type(e).__name__}: {e}" + + print(f"\nmodel={args.model_id} base=${base}/MTok gap={args.gap_seconds}s\n") + print(f"{'arm':<4} {'call':<7} {'input':>7} {'read':>8} {'write':>8} {'$':>10}") + total = {} + for name, ttl in arms: + r = results[name] + if "error" in r: + print(f"{name:<4} ERROR {r['error']}") + continue + mult = 2.0 if ttl == "1h" else 1.25 + cost = 0.0 + for which in ("first", "second"): + u = r[which] + c = _price(u, base, mult) + cost += c + print(f"{name:<4} {which:<7} {u['input']:>7} {u['read']:>8} {u['write']:>8} {c:>10.6f}") + total[name] = cost + print(f"{name:<4} {'pair':<7} {'':>7} {'':>8} {'':>8} {cost:>10.6f}") + if "5m" in total and "1h" in total: + delta = total["1h"] - total["5m"] + verdict = "1h CHEAPER" if delta < 0 else "1h MORE EXPENSIVE" + print(f"\n{verdict} by ${abs(delta):.6f} for this pair at a {args.gap_seconds}s gap") + second = results["1h"].get("second", {}) + if args.gap_seconds > 300 and second.get("read", 0) == 0: + print("⚠️ the 1h arm did NOT read after the gap — Bedrock may not honor ttl=1h for this model; do not enable the flag") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/src/agents/main_agent/chat_agent.py b/backend/src/agents/main_agent/chat_agent.py index 2e72e608f..dc30a5e1c 100644 --- a/backend/src/agents/main_agent/chat_agent.py +++ b/backend/src/agents/main_agent/chat_agent.py @@ -6,6 +6,7 @@ """ import logging +import os from typing import Any, AsyncGenerator, Dict, List, Optional from agents.main_agent.base_agent import BaseAgent @@ -58,7 +59,7 @@ def _create_agent(self) -> None: # files (added after tool filtering — it is infrastructure, not an # RBAC-gated tool, and is implicitly scoped to the turn's skills). plugin, read_skill_file = build_skills_runtime(self._accessible_skill_ids) - plugins = [plugin] if plugin else None + plugins = [plugin] if plugin else [] if plugin: tools = list(tools) + [read_skill_file] logger.info( @@ -66,6 +67,29 @@ def _create_agent(self) -> None: len(self._accessible_skill_ids or []), ) + # Tool-result offload at intake (compaction PR-4): oversized tool + # results become a bounded preview + retrieval references before + # they enter the cacheable prefix. Fail-open: None when off or + # unconfigured. The plugin registers retrieve_offloaded_content + # itself — one stable spec in toolConfig, not an RBAC-gated tool, + # like read_skill_file above. + from agents.main_agent.core.tool_result_offload import build_tool_result_offloader + + offload_session = getattr(self, "session_id", None) + offload_user = getattr(self, "user_id", None) + offloader = ( + build_tool_result_offloader( + session_id=offload_session, + user_id=offload_user, + region=os.environ.get("AWS_REGION"), + ) + if offload_session and offload_user + else None + ) + if offloader is not None: + plugins.append(offloader) + plugins = plugins or None + self.agent = AgentFactory.create_agent( model_config=self.model_config, system_prompt=self._system_prompt_for(tools), diff --git a/backend/src/agents/main_agent/config/constants.py b/backend/src/agents/main_agent/config/constants.py index d9ed82a7e..9ed45ba46 100644 --- a/backend/src/agents/main_agent/config/constants.py +++ b/backend/src/agents/main_agent/config/constants.py @@ -43,6 +43,30 @@ class EnvVars: # (a prefix re-write per turn, and it moves the coordinates the compaction # checkpoint is expressed in). Setting this to 40 restores the SDK default. CONVERSATION_WINDOW_MESSAGES = "AGENTCORE_CONVERSATION_WINDOW_MESSAGES" + # Bounded compaction summary (spiral spec PR-2 / + # compaction-model-relative-thresholds.md §3.6). Budget in tokens; the + # re-summarize call is a Nova Micro side-channel with its own kill switch. + COMPACTION_SUMMARY_TOKEN_BUDGET = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_TOKEN_BUDGET" + COMPACTION_SUMMARY_MODEL_ENABLED = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ENABLED" + COMPACTION_SUMMARY_MODEL_ID = "AGENTCORE_MEMORY_COMPACTION_SUMMARY_MODEL_ID" + # Paid-when-free scheduling (thresholds spec §3.5): a cut is computed and + # persisted as PENDING post-turn and applied to the live list pre-call only + # when the prefix re-write is free (cache expired, model/agent switched) + # or unavoidable (hard ceiling). "false" applies cuts immediately (PR-1/2). + COMPACTION_DEFERRED_APPLY_ENABLED = "AGENTCORE_MEMORY_COMPACTION_DEFERRED_APPLY_ENABLED" + # Tool-result offload at intake (thresholds spec §3.6 / PR-4): oversized + # tool results are stored in S3 (user-files bucket, per-session prefix) and + # replaced in context by a bounded preview + retrieval references before + # they ever enter the cacheable prefix. Strands' vended ContextOffloader. + TOOL_RESULT_OFFLOAD_ENABLED = "AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED" + TOOL_RESULT_OFFLOAD_MAX_TOKENS = "AGENTCORE_TOOL_RESULT_OFFLOAD_MAX_TOKENS" + TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS = "AGENTCORE_TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS" + # Selective long cache TTL on the STATIC prefix (thresholds spec §3.6, + # PR-5). "1h" puts a 1-hour TTL on the tools and system cachePoints only; + # the message-level point stays at Bedrock's 5-minute default. Unset/empty + # = today's shape. An experiment arm: default OFF until the live probe + # (scripts/probe_static_prefix_ttl.py) and the cost rows say it pays. + PROMPT_CACHE_STATIC_PREFIX_TTL = "AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL" # --- Restored-history repair --- # Kill switch for the restore-time tool-pairing/alternation repair @@ -148,6 +172,29 @@ class Defaults: # prompt. Overflow recovery (reduce_context on ContextWindowOverflow) still # works at any window size. CONVERSATION_WINDOW_MESSAGES = 2000 + # 8k tokens ≈ 32k chars: a third of the 25k floor, so a bounded summary + # can never by itself hold a session above the ceiling (the incident's + # summary was 40k tokens against a 100k threshold). Same figure the admin + # SUMMARY_OVER_BUDGET diagnosis reads. + COMPACTION_SUMMARY_TOKEN_BUDGET = 8_000 + COMPACTION_SUMMARY_MODEL_ENABLED = True + # Same cheap model as the title and tool-batch side-channels. + COMPACTION_SUMMARY_MODEL_ID = "us.amazon.nova-micro-v1:0" + COMPACTION_DEFERRED_APPLY_ENABLED = True + # Tool-result offload gate. 4k is well under the 25k compaction floor, so a + # protected tail of a few big results can no longer hold a session above + # the ceiling on its own; the 1k preview keeps the part of a result models + # actually quote (headers, first rows, the first error). + TOOL_RESULT_OFFLOAD_ENABLED = True + TOOL_RESULT_OFFLOAD_MAX_TOKENS = 4_000 + TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS = 1_000 + TOOL_RESULT_OFFLOAD_S3_PREFIX = "compaction-offload" + # Default OFF, deliberately against the flags-default-on house style: a + # caching default adopted on inspection alone has already shipped wrong + # once (#954 measured 57% more expensive live before #956 reverted it). + # The 1h write premium is 2x base vs 1.25x at 5m, so this is a bet on the + # gap distribution that has to be measured, not read off the source. + PROMPT_CACHE_STATIC_PREFIX_TTL = "" # --- DynamoDB Tables --- DYNAMODB_QUOTA_TABLE = "UserQuotas" diff --git a/backend/src/agents/main_agent/core/agent_factory.py b/backend/src/agents/main_agent/core/agent_factory.py index e95511546..180a52697 100644 --- a/backend/src/agents/main_agent/core/agent_factory.py +++ b/backend/src/agents/main_agent/core/agent_factory.py @@ -292,6 +292,13 @@ def create_agent( # on a NON-Anthropic model this block is passed through untouched and # Bedrock rejects the call with AccessDeniedException. # + # PR-5 (thresholds spec §3.6): the point is placed TTL-less on purpose. + # With AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL=1h, ModelConfig sets + # CacheConfig(system_prompt_ttl="1h", tools_ttl="1h") and upstream's + # _apply_system_cache_ttl rewrites THIS point's ttl ("an explicit + # system_prompt_ttl string is honored as written"); the tools point + # gets its own. Flag unset → no ttl key anywhere → today's bytes. + # # RE-VERIFY BEFORE ANY BUMP PAST 1.55.0. This is a statement about # upstream internals and it has already rotted once. Re-check # _should_cache_system's guard, CacheConfig.system_prompt_ttl's diff --git a/backend/src/agents/main_agent/core/model_config.py b/backend/src/agents/main_agent/core/model_config.py index 8ec486b89..f6a702582 100644 --- a/backend/src/agents/main_agent/core/model_config.py +++ b/backend/src/agents/main_agent/core/model_config.py @@ -285,6 +285,25 @@ def from_env(cls) -> "RetryConfig": ) +# Bedrock's long cache TTL. Only "1h" is a change from the default; anything +# else (unset, "5m", garbage) means "today's shape" — no ttl key on any point, +# which is what keeps the static prefix bytes identical across the flip. +LONG_CACHE_TTL = "1h" + + +def static_prefix_cache_ttl() -> Optional[str]: + """The long TTL to put on the tools + system cachePoints, or ``None``. + + Read from ``AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL`` at agent + construction (a cached agent keeps the arm it was built under). See + docs/specs/compaction-model-relative-thresholds.md §3.6 PR-5 for the + economics: 2x write premium on the static segments in exchange for + reading them, rather than re-writing them, on every 5–60 minute pause. + """ + raw = os.environ.get(EnvVars.PROMPT_CACHE_STATIC_PREFIX_TTL, "").strip().lower() + return LONG_CACHE_TTL if raw == LONG_CACHE_TTL else None + + @dataclass class ModelConfig: """Configuration for multi-provider LLM models. @@ -342,6 +361,15 @@ def get_provider(self) -> ModelProvider: # Default to configured provider return self.provider + def long_ttl_static_prefix(self) -> bool: + """True when this model's tools + system cachePoints carry the 1h TTL. + + The cost path uses it to bill the static segment's cache writes at + Bedrock's 1h premium (2x base) instead of the 5m one (1.25x) — the + correction that keeps the experiment arm's own cost rows honest. + """ + return bool(self.caching_enabled and self.bedrock_cache_points_supported() and static_prefix_cache_ttl()) + def bedrock_cache_points_supported(self) -> bool: """Whether a hand-placed Bedrock system cachePoint may be sent. @@ -471,10 +499,22 @@ def to_bedrock_config(self) -> Dict[str, Any]: # that reaches Bedrock without going through that factory. if self.caching_enabled: from strands.models import CacheConfig + + # PR-5 (thresholds spec §3.6): an explicit "1h" on the two STATIC + # points only. system_prompt_ttl as a string is "honored as + # written" by _apply_system_cache_ttl, which rewrites the TTL on + # the hand-placed, TTL-less system point AgentFactory places; + # tools_ttl as a string sets the tools point's own TTL. The + # message-level auto point carries no ttl (cache_config.ttl stays + # unset) and so stays at 5m — tools(1h) → system(1h) → messages(5m) + # is the non-increasing order Bedrock requires. Off (the default) + # emits exactly today's bytes. + supported = self.bedrock_cache_points_supported() + long_ttl = static_prefix_cache_ttl() if supported else None config["cache_config"] = CacheConfig( strategy="auto", - system_prompt_ttl=True, - tools_ttl=self.bedrock_cache_points_supported(), + system_prompt_ttl=long_ttl or True, + tools_ttl=(long_ttl or True) if supported else False, ) if self.retry_config: diff --git a/backend/src/agents/main_agent/core/tool_result_offload.py b/backend/src/agents/main_agent/core/tool_result_offload.py new file mode 100644 index 000000000..bab3b21af --- /dev/null +++ b/backend/src/agents/main_agent/core/tool_result_offload.py @@ -0,0 +1,186 @@ +""" +Tool-result offload at intake — the compaction escalation for oversized tool +results (docs/specs/compaction-model-relative-thresholds.md §3.6, PR-4). + +The compaction cut keeps the last ``protected_turns`` turns whole. When one of +those turns carries a 90k-token tool result, the protected tail alone exceeds +the floor and no cut can get the session back under the ceiling +(``compaction_floor_unreachable``). Cutting inside the tail would drop the +turn the user is working on; the right move is to keep the *reference* and +drop the *bytes*. + +Strands 1.55's vended ``ContextOffloader`` does exactly that at the cheapest +possible moment — ``AfterToolCallEvent``, before the result is appended to the +conversation. The persisted message already carries the bounded form, so a +restore reproduces it byte-for-byte and nothing in the prefix is ever +mutated after the fact (the byte-stability contract in CLAUDE.md). The model +keeps a preview and can pull any span back with ``retrieve_offloaded_content`` +(pattern / line range / full). + +What this module owns on top of the plugin: + +- **Storage**: S3 in the user-files bucket under + ``compaction-offload/{user_id}/{session_id}/`` — one namespaced storage per + agent, so references are scoped to the session by construction (a session + cannot retrieve another session's content) and a second agent instance for + the same session (an ``@``-mention) resolves the same references. +- **No eviction from the model path.** ``evict_after_cycles=None``: the + plugin's cycle-based eviction runs on ``BeforeModelCallEvent`` and would + turn a retrieval into a miss mid-conversation. Objects expire by S3 + lifecycle instead (see the file-upload construct). +- **A cheap pre-filter.** The plugin sizes every result with + ``model.count_tokens`` — a Bedrock CountTokens round trip per tool call. + ``BoundedToolResultOffloader`` estimates with chars/4 first and only lets + results near or over the gate reach the API. +- **A content-free record per offload** (``AgentCoreStack/Compaction``: + ``ToolResultOffloaded`` + token count), so the data point "how much tool + payload were we about to put in the prefix" exists. + +Documents attached by the user are NOT handled here: the digest + page-range +read tool in ``document-context-offload.md`` is the right shape for those, and +this module deliberately does not strip them. +""" + +from __future__ import annotations + +import logging +import os +from typing import Any, Optional + +from agents.main_agent.config.constants import Defaults, EnvVars +from agents.main_agent.session.compaction_policy import estimate_message_tokens + +logger = logging.getLogger(__name__) + +# Below this fraction of the gate the chars/4 estimate is trusted outright +# and CountTokens is skipped. The heuristic is ~±25% on English text and JSON; +# half the gate leaves that margin twice over. +PREFILTER_RATIO = 0.5 + + +def tool_result_offload_enabled() -> bool: + """Default ON with a kill switch (house style): only the literal "false" disables.""" + return os.environ.get(EnvVars.TOOL_RESULT_OFFLOAD_ENABLED, "").strip().lower() != "false" + + +def _int_env(name: str, default: int) -> int: + raw = os.environ.get(name, "").strip() + try: + return int(raw) if raw else default + except ValueError: + return default + + +def _import_plugin(): + from strands.vended_plugins.context_offloader import ContextOffloader + + return ContextOffloader + + +class _OffloaderMixin: + """Behavior layered on the vended plugin; kept separate so it can be tested + against a stub base class without importing the real one.""" + + async def _handle_tool_result(self, event: Any) -> None: # type: ignore[override] + try: + result = getattr(event, "result", None) + if not isinstance(result, dict): + return + content = result.get("content") + if not isinstance(content, list): + return + # Cheap pre-filter: skip the CountTokens round trip for results + # that cannot be anywhere near the gate. + estimate = estimate_message_tokens({"role": "user", "content": [{"toolResult": result}]}) + if estimate < self._max_result_tokens * PREFILTER_RATIO: # type: ignore[attr-defined] + return + before = event.result + await super()._handle_tool_result(event) # type: ignore[misc] + if event.result is not before: + self._record_offload(event, estimate) + except Exception: # noqa: BLE001 - offload is never worth a failed tool call + logger.warning("tool-result offload skipped, keeping the original result", exc_info=True) + + @staticmethod + def _record_offload(event: Any, estimate: int) -> None: + tool_name = None + try: + tool_name = (getattr(event, "tool_use", None) or {}).get("name") + except Exception: # noqa: BLE001 + pass + logger.info("tool_result_offloaded: tool=%s est_tokens=%d", tool_name, estimate) + try: + from apis.shared.observability.prompt_cache import prompt_cache_observability_enabled + from apis.shared.observability.emf import emit_emf_metrics + + if prompt_cache_observability_enabled(): + emit_emf_metrics( + "AgentCoreStack/Compaction", + metrics={"ToolResultOffloaded": 1, "ToolResultOffloadedTokens": int(estimate)}, + properties={"toolName": tool_name}, + units={"ToolResultOffloadedTokens": "Count"}, + ) + except Exception as e: # noqa: BLE001 + logger.debug("offload EMF skipped: %s", e) + + +def _offloader_class(): + """``BoundedToolResultOffloader`` built lazily so importing this module never + imports the plugin (and boto3) on paths that do not use it.""" + ContextOffloader = _import_plugin() + + class BoundedToolResultOffloader(_OffloaderMixin, ContextOffloader): # type: ignore[misc, valid-type] + pass + + BoundedToolResultOffloader.__name__ = "BoundedToolResultOffloader" + return BoundedToolResultOffloader + + +def offload_prefix(user_id: str, session_id: str) -> str: + return f"{Defaults.TOOL_RESULT_OFFLOAD_S3_PREFIX}/{user_id}/{session_id}" + + +def build_tool_result_offloader( + *, + session_id: str, + user_id: str, + region: Optional[str] = None, + storage: Any = None, +) -> Optional[Any]: + """The plugin for one agent, or ``None`` when off or unconfigured (fail-open). + + ``storage`` overrides the S3 backend (tests). Returns ``None`` — never + raises — when the flag is off, the user-files bucket is not configured, + or the plugin cannot be constructed. + """ + if not tool_result_offload_enabled(): + return None + try: + max_tokens = max(1, _int_env(EnvVars.TOOL_RESULT_OFFLOAD_MAX_TOKENS, Defaults.TOOL_RESULT_OFFLOAD_MAX_TOKENS)) + preview = _int_env(EnvVars.TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS, Defaults.TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS) + preview = max(0, min(preview, max_tokens - 1)) + + if storage is None: + bucket = os.environ.get("S3_USER_FILES_BUCKET_NAME", "").strip() + if not bucket: + logger.info("tool-result offload disabled: S3_USER_FILES_BUCKET_NAME is not set") + return None + from strands.storage import S3Storage + + storage = S3Storage( + bucket, + prefix=offload_prefix(user_id, session_id), + region_name=region or os.environ.get("AWS_REGION", "us-west-2"), + ) + + cls = _offloader_class() + return cls( + storage=storage, + max_result_tokens=max_tokens, + preview_tokens=preview, + include_retrieval_tool=True, + evict_after_cycles=None, + ) + except Exception: # noqa: BLE001 + logger.warning("tool-result offload disabled: plugin construction failed", exc_info=True) + return None diff --git a/backend/src/agents/main_agent/session/compaction_models.py b/backend/src/agents/main_agent/session/compaction_models.py index 733106de5..4bcc50646 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 000000000..64076ca17 --- /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 9a9244121..81572d82c 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,216 @@ 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, + # The protected tail alone exceeded the floor: the residual + # case after intake offload (attachments, sub-gate results). + "CompactionFloorUnreachable": ( + 1 if (policy.floor is not None and retained_estimate is not None and retained_estimate > policy.floor) else 0 + ), + }, + { + "policySource": policy.source, + "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 1513531d5..838a376d0 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. @@ -667,6 +682,9 @@ async def stream_response( # persisted (one C# record per assistant message). if main_agent_wrapper and hasattr(main_agent_wrapper, "model_config"): model_id = main_agent_wrapper.model_config.model_id + # PR-5: when the static prefix carries the 1h TTL, + # its cache writes are billed at the 1h premium. + long_ttl_static_tokens = self._long_ttl_static_prefix_tokens(main_agent_wrapper, agent) try: turn_total = 0.0 turn_input_cost = 0.0 @@ -680,6 +698,7 @@ async def stream_response( msg_cost = await self._calculate_streaming_cost( model_id=model_id, usage=msg_usage, + long_ttl_static_prefix_tokens=long_ttl_static_tokens, ) if msg_cost is None: continue @@ -2997,7 +3016,13 @@ async def _store_message_metadata( # Calculate cost if we have both usage and pricing if token_usage and pricing_snapshot: - cost_result = self._calculate_message_cost(usage=accumulated_metadata.get("usage", {}), pricing=pricing_snapshot) + cost_result = self._calculate_message_cost( + usage=accumulated_metadata.get("usage", {}), + pricing=pricing_snapshot, + long_ttl_static_prefix_tokens=self._long_ttl_static_prefix_tokens( + agent, getattr(agent, "agent", None) + ), + ) if cost_result is not None: cost = cost_result @@ -3025,6 +3050,15 @@ async def _store_message_metadata( ) if context_window is not None: metadata_kwargs["contextWindow"] = context_window + # PR-5 experiment arm marker (extra field via extra="allow"), + # so the cost anatomy can split 1h-static-prefix turns from + # the 5m baseline without guessing from the write:read shape. + try: + if agent is not None and getattr(agent, "model_config", None) is not None and \ + getattr(agent.model_config, "long_ttl_static_prefix", lambda: False)(): + metadata_kwargs["staticPrefixTtl"] = "1h" + except Exception: # noqa: BLE001 + pass # Prompt-cache prefix fingerprints for this model call # (extra field via extra="allow"; persisted on the cost row @@ -3109,6 +3143,35 @@ def _extract_model_version(self, model_id: str) -> Optional[str]: return part.split(":")[0] return None + @staticmethod + def _long_ttl_static_prefix_tokens(main_agent_wrapper: Any, strands_agent: Any) -> Optional[int]: + """Size of the tools + system segment when it carries the 1h cache TTL, else ``None``. + + PR-5 (thresholds spec §3.6): Bedrock bills a 1h cache write at 2x base, + not the 1.25x the catalog's cacheWritePricePerMtok carries, and usage + does not split writes by TTL. The context-attribution breakdown knows + the static segment's size, and the read count tells whether it was + written this call (see CostCalculator.calculate_message_cost). + """ + try: + model_config = getattr(main_agent_wrapper, "model_config", None) + predicate = getattr(model_config, "long_ttl_static_prefix", None) + if not callable(predicate) or not predicate(): + return None + from agents.main_agent.session.hooks.context_attribution import get_context_breakdown + + breakdown = get_context_breakdown(strands_agent) if strands_agent is not None else None + if not breakdown: + return None + total = 0 + for partition in breakdown.get("partitions", []) or []: + if isinstance(partition, dict) and partition.get("key") in ("system", "tools"): + total += int(partition.get("tokens") or 0) + return total or None + except Exception as e: # noqa: BLE001 + logger.debug(f"long-TTL static prefix size unavailable: {e}") + return None + async def _get_pricing_snapshot(self, model_id: str) -> Optional[Dict[str, Any]]: """ Get pricing snapshot from managed models database @@ -3137,7 +3200,12 @@ async def _get_pricing_snapshot(self, model_id: str) -> Optional[Dict[str, Any]] logger.error(f"Failed to get pricing snapshot for {model_id}: {e}") return None - def _calculate_message_cost(self, usage: Dict[str, Any], pricing: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + def _calculate_message_cost( + self, + usage: Dict[str, Any], + pricing: Optional[Dict[str, Any]], + long_ttl_static_prefix_tokens: Optional[int] = None, + ) -> Optional[Dict[str, Any]]: """ Calculate message cost from usage and pricing @@ -3160,7 +3228,9 @@ def _calculate_message_cost(self, usage: Dict[str, Any], pricing: Optional[Dict[ else: pricing_dict = pricing - total_cost, breakdown = CostCalculator.calculate_message_cost(usage, pricing_dict) + total_cost, breakdown = CostCalculator.calculate_message_cost( + usage, pricing_dict, long_ttl_static_prefix_tokens=long_ttl_static_prefix_tokens + ) return { "total": total_cost, "inputCost": breakdown.input_cost, @@ -3173,7 +3243,12 @@ def _calculate_message_cost(self, usage: Dict[str, Any], pricing: Optional[Dict[ logger.error(f"Failed to calculate message cost: {e}") return None - async def _calculate_streaming_cost(self, model_id: str, usage: Dict[str, Any]) -> Optional[Dict[str, Any]]: + async def _calculate_streaming_cost( + self, + model_id: str, + usage: Dict[str, Any], + long_ttl_static_prefix_tokens: Optional[int] = None, + ) -> Optional[Dict[str, Any]]: """ Calculate cost for streaming response to send to client in real-time. @@ -3208,7 +3283,7 @@ async def _calculate_streaming_cost(self, model_id: str, usage: Dict[str, Any]) ) # Calculate cost using the calculator - return self._calculate_message_cost(usage, pricing) + return self._calculate_message_cost(usage, pricing, long_ttl_static_prefix_tokens=long_ttl_static_prefix_tokens) except Exception as e: logger.warning(f"Failed to calculate streaming cost: {e}") diff --git a/backend/src/apis/inference_api/chat/service.py b/backend/src/apis/inference_api/chat/service.py index bcd5b9719..93f6279d9 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/src/apis/shared/costs/calculator.py b/backend/src/apis/shared/costs/calculator.py index 788e64e0e..0c26040fc 100644 --- a/backend/src/apis/shared/costs/calculator.py +++ b/backend/src/apis/shared/costs/calculator.py @@ -6,17 +6,25 @@ - Multi-provider cost support (Bedrock, OpenAI, Gemini) """ -from typing import Dict, Tuple +from typing import Dict, Optional, Tuple from .models import CostBreakdown class CostCalculator: """Calculate costs from token usage and pricing""" + # Bedrock's cache-write premium for a 1-hour TTL, as a multiple of the + # model's base input rate (5m is 1.25x — that is what the catalog's + # cacheWritePricePerMtok already carries). See the prompt-cache contract + # in CLAUDE.md. + LONG_TTL_CACHE_WRITE_MULTIPLIER = 2.0 + @staticmethod def calculate_message_cost( usage: Dict[str, int], - pricing: Dict[str, float] + pricing: Dict[str, float], + *, + long_ttl_static_prefix_tokens: Optional[int] = None, ) -> Tuple[float, CostBreakdown]: """ Calculate cost for a single message @@ -24,6 +32,15 @@ def calculate_message_cost( Args: usage: Token usage dict with inputTokens, outputTokens, etc. pricing: Pricing dict with inputPricePerMtok, etc. + long_ttl_static_prefix_tokens: When the turn ran with the 1h TTL on + the static (tools + system) cachePoints, the size of that static + segment. Cache writes that cover it are billed at the 1h premium + (2x base) instead of the 5m one. Which part of the write was + static is inferred from the read: a static segment that was + read (cacheRead >= static) was not written, so the write is all + history at 5m; otherwise the unread remainder of the static + segment was written at 1h. ``None`` (the default) prices every + cache write at the catalog's 5m rate — today's behavior. Returns: Tuple of (total_cost, cost_breakdown) @@ -79,7 +96,15 @@ def calculate_message_cost( input_cost = (input_tokens / 1_000_000) * input_price output_cost = (output_tokens / 1_000_000) * output_price cache_read_cost = (cache_read_tokens / 1_000_000) * cache_read_price - cache_write_cost = (cache_write_tokens / 1_000_000) * cache_write_price + + long_ttl_written = 0 + if long_ttl_static_prefix_tokens and long_ttl_static_prefix_tokens > 0 and cache_write_tokens > 0: + long_ttl_written = max(0, min(cache_write_tokens, int(long_ttl_static_prefix_tokens) - cache_read_tokens)) + long_ttl_write_price = input_price * CostCalculator.LONG_TTL_CACHE_WRITE_MULTIPLIER + cache_write_cost = ( + (long_ttl_written / 1_000_000) * long_ttl_write_price + + ((cache_write_tokens - long_ttl_written) / 1_000_000) * cache_write_price + ) total_cost = input_cost + output_cost + cache_read_cost + cache_write_cost diff --git a/backend/tests/agents/main_agent/core/test_bedrock_cache_points.py b/backend/tests/agents/main_agent/core/test_bedrock_cache_points.py index 16d2caeee..39986fbf5 100644 --- a/backend/tests/agents/main_agent/core/test_bedrock_cache_points.py +++ b/backend/tests/agents/main_agent/core/test_bedrock_cache_points.py @@ -358,3 +358,64 @@ def test_the_static_points_are_untouched(self, model): assert steered["system"] == plain["system"] # Everything before the mixed message is identical too. assert steered["messages"][:-1] == plain["messages"][:-1] + + +# --------------------------------------------------------------------------- +# PR-5: selective 1h TTL on the STATIC prefix (thresholds spec §3.6) +# --------------------------------------------------------------------------- + +class TestStaticPrefixLongTtl: + """AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL=1h puts a 1h TTL on the tools + and system points only; the message point stays at the 5m default, and + the flag off emits exactly today's bytes.""" + + def _model(self, monkeypatch, value): + monkeypatch.setenv("AWS_REGION", "us-west-2") + if value is None: + monkeypatch.delenv("AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL", raising=False) + else: + monkeypatch.setenv("AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL", value) + from agents.main_agent.core.bedrock_count_tokens import CountTokensBedrockModel + + config = ModelConfig(model_id=CLAUDE_MODEL_ID, caching_enabled=True) + return config, CountTokensBedrockModel(**config.to_bedrock_config()) + + def _request(self, model): + return model.format_request( + [{"role": "user", "content": [{"text": "hi"}]}], + [{"name": "t", "description": "d", "inputSchema": {"json": {"type": "object", "properties": {}}}}], + system_prompt_content=[{"text": "sys"}, {"cachePoint": {"type": "default"}}], + ) + + def test_flag_on_sets_1h_on_tools_and_system_only(self, monkeypatch): + config, model = self._model(monkeypatch, "1h") + cc = config.to_bedrock_config()["cache_config"] + assert cc.tools_ttl == "1h" and cc.system_prompt_ttl == "1h" and cc.ttl is None + assert config.long_ttl_static_prefix() is True + req = self._request(model) + assert req["toolConfig"]["tools"][-1] == {"cachePoint": {"type": "default", "ttl": "1h"}} + assert req["system"][-1] == {"cachePoint": {"type": "default", "ttl": "1h"}} + assert req["messages"][-1]["content"][-1] == {"cachePoint": {"type": "default"}} + assert _count_cache_points(req) == 3 + + @pytest.mark.parametrize("value", [None, "", "5m", "2h", "true"]) + def test_anything_but_1h_is_todays_bytes(self, monkeypatch, value): + config, model = self._model(monkeypatch, value) + cc = config.to_bedrock_config()["cache_config"] + assert cc.tools_ttl is True and cc.system_prompt_ttl is True + assert config.long_ttl_static_prefix() is False + req = self._request(model) + assert req["toolConfig"]["tools"][-1] == {"cachePoint": {"type": "default"}} + assert req["system"][-1] == {"cachePoint": {"type": "default"}} + assert _count_cache_points(req) == 3 + + def test_non_anthropic_model_is_unaffected(self, monkeypatch): + monkeypatch.setenv("AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL", "1h") + config = ModelConfig(model_id="us.amazon.nova-micro-v1:0", caching_enabled=True) + cc = config.to_bedrock_config()["cache_config"] + assert cc.tools_ttl is False + assert config.long_ttl_static_prefix() is False + + def test_caching_disabled_means_no_long_ttl(self, monkeypatch): + monkeypatch.setenv("AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL", "1h") + assert ModelConfig(model_id=CLAUDE_MODEL_ID, caching_enabled=False).long_ttl_static_prefix() is False diff --git a/backend/tests/agents/main_agent/core/test_tool_result_offload.py b/backend/tests/agents/main_agent/core/test_tool_result_offload.py new file mode 100644 index 000000000..4426bdff5 --- /dev/null +++ b/backend/tests/agents/main_agent/core/test_tool_result_offload.py @@ -0,0 +1,245 @@ +"""Tool-result offload at intake — thresholds spec §3.6 (PR-4).""" + +import sys +import types +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from agents.main_agent.core import tool_result_offload as tro +from agents.main_agent.core.tool_result_offload import ( + PREFILTER_RATIO, + build_tool_result_offloader, + offload_prefix, + tool_result_offload_enabled, +) + + +class _FakeStorage: + """Unified-Storage-shaped in-memory backend (write/read), like strands.storage.""" + + def __init__(self): + self.objects = {} + + async def write(self, key, data): + self.objects[key] = data + + async def read(self, key): + return self.objects.get(key) + + async def delete(self, key): + self.objects.pop(key, None) + + async def list(self, query=""): + return [k for k in self.objects if k.startswith(query)] + + +class _Agent: + """Weak-referenceable stand-in (the plugin keys WeakKeyDictionaries on the agent).""" + + def __init__(self, model=None): + self.model = model + self.event_loop_metrics = SimpleNamespace(cycle_count=1) + self.storage = None + self.sandbox = None + + +def _event(text, tool_name="gmail_search", count_tokens=None): + result = {"toolUseId": "t1", "status": "success", "content": [{"text": text}]} + model = SimpleNamespace(count_tokens=count_tokens or AsyncMock(return_value=len(text) // 4)) + agent = _Agent(model) + return SimpleNamespace( + result=result, + tool_use={"toolUseId": "t1", "name": tool_name}, + selected_tool=None, + cancel_message=None, + agent=agent, + ) + + +@pytest.fixture +def offloader(monkeypatch): + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_MAX_TOKENS", "1000") + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS", "100") + storage = _FakeStorage() + plugin = build_tool_result_offloader(session_id="s1", user_id="u1", storage=storage) + assert plugin is not None + # Bind storage the way init_agent would. + plugin.init_agent(_Agent()) + return plugin, storage + + +class TestBuilder: + def test_flag_default_on_only_literal_false_off(self, monkeypatch): + monkeypatch.delenv("AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED", raising=False) + assert tool_result_offload_enabled() is True + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED", "false") + assert tool_result_offload_enabled() is False + assert build_tool_result_offloader(session_id="s", user_id="u", storage=_FakeStorage()) is None + + def test_no_bucket_means_no_plugin(self, monkeypatch): + monkeypatch.delenv("S3_USER_FILES_BUCKET_NAME", raising=False) + assert build_tool_result_offloader(session_id="s", user_id="u") is None + + def test_s3_backend_is_scoped_per_user_and_session(self, monkeypatch): + monkeypatch.setenv("S3_USER_FILES_BUCKET_NAME", "files-bucket") + captured = {} + + class FakeS3Storage: + def __init__(self, bucket, *, prefix="", region_name=None, **kw): + captured.update(bucket=bucket, prefix=prefix, region=region_name) + + async def write(self, k, d): ... + async def read(self, k): ... + async def delete(self, k): ... + async def list(self, q=""): return [] + + monkeypatch.setitem(sys.modules, "strands.storage", types.SimpleNamespace(S3Storage=FakeS3Storage)) + plugin = build_tool_result_offloader(session_id="sess", user_id="usr", region="us-west-2") + assert plugin is not None + assert captured == {"bucket": "files-bucket", "prefix": "compaction-offload/usr/sess", "region": "us-west-2"} + assert offload_prefix("usr", "sess") == "compaction-offload/usr/sess" + + def test_configuration(self, offloader): + plugin, _ = offloader + assert plugin._max_result_tokens == 1000 + assert plugin._preview_tokens == 100 + assert plugin._evict_after_cycles is None # no eviction from the model path + assert plugin._include_retrieval_tool is True + assert type(plugin).__name__ == "BoundedToolResultOffloader" + + def test_preview_is_clamped_below_gate(self, monkeypatch): + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_MAX_TOKENS", "500") + monkeypatch.setenv("AGENTCORE_TOOL_RESULT_OFFLOAD_PREVIEW_TOKENS", "9000") + plugin = build_tool_result_offloader(session_id="s", user_id="u", storage=_FakeStorage()) + assert plugin._preview_tokens == 499 + + +class TestPrefilterAndOffload: + @pytest.mark.asyncio + async def test_small_result_skips_count_tokens_and_is_untouched(self, offloader): + plugin, storage = offloader + counter = AsyncMock(return_value=10) + ev = _event("short result", count_tokens=counter) + before = ev.result + await plugin._handle_tool_result(ev) + counter.assert_not_called() + assert ev.result is before and storage.objects == {} + + @pytest.mark.asyncio + async def test_borderline_result_is_measured_by_count_tokens(self, offloader): + plugin, _ = offloader + # ~600 estimated tokens: over the pre-filter (500) but under the gate (1000). + counter = AsyncMock(return_value=600) + ev = _event("x" * 2400, count_tokens=counter) + before = ev.result + await plugin._handle_tool_result(ev) + counter.assert_called_once() + assert ev.result is before + + @pytest.mark.asyncio + async def test_oversized_result_is_offloaded_with_preview_and_reference(self, offloader, monkeypatch): + plugin, storage = offloader + import apis.shared.observability.emf as emf + emitted = [] + monkeypatch.setattr(emf, "emit_emf_metrics", lambda ns, metrics, properties=None, units=None: emitted.append((ns, metrics, properties))) + monkeypatch.delenv("PROMPT_CACHE_OBSERVABILITY_ENABLED", raising=False) + + body = "header line\n" + ("row data " * 2000) + ev = _event(body, count_tokens=AsyncMock(return_value=5000)) + await plugin._handle_tool_result(ev) + + text = ev.result["content"][0]["text"] + assert text.startswith("[Offloaded:") + assert "retrieve_offloaded_content" in text + assert "header line" in text # preview keeps the head + assert len(text) < len(body) // 4 + assert len(storage.objects) == 1 # the full block landed in storage + assert ev.result["toolUseId"] == "t1" and ev.result["status"] == "success" + assert emitted and emitted[0][0] == "AgentCoreStack/Compaction" + assert emitted[0][1]["ToolResultOffloaded"] == 1 + assert emitted[0][2]["toolName"] == "gmail_search" + assert "row data" not in str(emitted[0][2]) # content-free + + @pytest.mark.asyncio + async def test_storage_failure_keeps_original_result(self, offloader): + plugin, storage = offloader + + async def boom(key, data): + raise RuntimeError("s3 down") + + storage.write = boom + ev = _event("y" * 20000, count_tokens=AsyncMock(return_value=5000)) + before = ev.result + await plugin._handle_tool_result(ev) + assert ev.result is before + + @pytest.mark.asyncio + async def test_count_tokens_failure_keeps_original_result(self, offloader): + plugin, _ = offloader + ev = _event("z" * 20000, count_tokens=AsyncMock(side_effect=RuntimeError("AccessDenied"))) + before = ev.result + await plugin._handle_tool_result(ev) + assert ev.result is before + + def test_prefilter_ratio_is_half_the_gate(self): + assert PREFILTER_RATIO == 0.5 + + +class TestChatAgentWiring: + def test_chat_agent_adds_the_offloader_plugin(self, monkeypatch): + from unittest.mock import MagicMock + from agents.main_agent.chat_agent import ChatAgent + from agents.main_agent.core import AgentFactory + + monkeypatch.setenv("S3_USER_FILES_BUCKET_NAME", "files-bucket") + + class FakeS3Storage: + def __init__(self, bucket, *, prefix="", region_name=None, **kw): + self.prefix = prefix + + async def write(self, k, d): ... + async def read(self, k): ... + async def delete(self, k): ... + async def list(self, q=""): return [] + + monkeypatch.setitem(sys.modules, "strands.storage", types.SimpleNamespace(S3Storage=FakeS3Storage)) + captured = {} + monkeypatch.setattr(AgentFactory, "create_agent", staticmethod(lambda **kw: captured.update(kw) or MagicMock())) + + agent = ChatAgent.__new__(ChatAgent) + agent.system_prompt = "BASE" + agent.model_config = MagicMock() + agent.session_manager = MagicMock() + agent.session_id = "sess" + agent.user_id = "usr" + agent._accessible_skill_ids = None + monkeypatch.setattr(ChatAgent, "_build_filtered_tools", lambda self: [], raising=False) + monkeypatch.setattr(ChatAgent, "_create_hooks", lambda self: [], raising=False) + + agent._create_agent() + + plugins = captured["plugins"] + assert plugins and type(plugins[0]).__name__ == "BoundedToolResultOffloader" + assert plugins[0]._storage._prefix if hasattr(plugins[0]._storage, "_prefix") else True + + def test_chat_agent_without_bucket_passes_no_plugins(self, monkeypatch): + from unittest.mock import MagicMock + from agents.main_agent.chat_agent import ChatAgent + from agents.main_agent.core import AgentFactory + + monkeypatch.delenv("S3_USER_FILES_BUCKET_NAME", raising=False) + captured = {} + monkeypatch.setattr(AgentFactory, "create_agent", staticmethod(lambda **kw: captured.update(kw) or MagicMock())) + agent = ChatAgent.__new__(ChatAgent) + agent.system_prompt = "BASE" + agent.model_config = MagicMock() + agent.session_manager = MagicMock() + agent.session_id = "sess" + agent.user_id = "usr" + agent._accessible_skill_ids = None + monkeypatch.setattr(ChatAgent, "_build_filtered_tools", lambda self: [], raising=False) + monkeypatch.setattr(ChatAgent, "_create_hooks", lambda self: [], raising=False) + agent._create_agent() + assert captured["plugins"] is None diff --git a/backend/tests/agents/main_agent/session/conftest.py b/backend/tests/agents/main_agent/session/conftest.py index 57aa832ee..d232ba130 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 000000000..b977dc526 --- /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 e7cc1c3eb..25ce5075b 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 36f6a0190..eb8b5aeab 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 000000000..08681ad85 --- /dev/null +++ b/backend/tests/agents/main_agent/session/test_compaction_summary.py @@ -0,0 +1,213 @@ +"""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 + # Tiny 5-turn conversation calibrated to 2000 tokens against a 250 + # floor with 3 protected turns: the tail cannot fit → unreachable. + assert metrics["CompactionFloorUnreachable"] == 1 + assert props["summaryOutcome"] == "within_budget" and props["policySource"] == "fixed" + # Content-free: no summary text, no message text in the record. + assert "tiny" not in str(props) + + @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/agents/main_agent/streaming/test_static_prefix_ttl_cost.py b/backend/tests/agents/main_agent/streaming/test_static_prefix_ttl_cost.py new file mode 100644 index 000000000..49dac86c3 --- /dev/null +++ b/backend/tests/agents/main_agent/streaming/test_static_prefix_ttl_cost.py @@ -0,0 +1,41 @@ +"""The coordinator sizes the 1h static prefix from the context breakdown (PR-5).""" + +from types import SimpleNamespace + +from agents.main_agent.streaming.stream_coordinator import StreamCoordinator + + +class _Strands: + pass + + +def _wrapper(long_ttl): + return SimpleNamespace(model_config=SimpleNamespace(long_ttl_static_prefix=lambda: long_ttl)) + + +def _with_breakdown(system, tools): + from agents.main_agent.session.hooks import context_attribution as ca + agent = _Strands() + setattr(agent, ca._BREAKDOWN_ATTR, { + "total": system + tools + 500, + "partitions": [ + {"key": "system", "label": "System prompt", "tokens": system}, + {"key": "tools", "label": "Tools", "tokens": tools}, + {"key": "messages", "label": "Messages", "tokens": 500}, + ], + }) + return agent + + +def test_returns_system_plus_tools_when_arm_is_on(): + assert StreamCoordinator._long_ttl_static_prefix_tokens(_wrapper(True), _with_breakdown(8_000, 3_000)) == 11_000 + + +def test_none_when_arm_is_off(): + assert StreamCoordinator._long_ttl_static_prefix_tokens(_wrapper(False), _with_breakdown(8_000, 3_000)) is None + + +def test_none_without_breakdown_or_wrapper(): + assert StreamCoordinator._long_ttl_static_prefix_tokens(_wrapper(True), _Strands()) is None + assert StreamCoordinator._long_ttl_static_prefix_tokens(None, _with_breakdown(1, 1)) is None + assert StreamCoordinator._long_ttl_static_prefix_tokens(_wrapper(True), None) is None diff --git a/backend/tests/apis/inference_api/test_chat_service.py b/backend/tests/apis/inference_api/test_chat_service.py index 27e0534ec..7eca46194 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/backend/tests/apis/shared/costs/__init__.py b/backend/tests/apis/shared/costs/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/tests/apis/shared/costs/test_calculator_long_ttl.py b/backend/tests/apis/shared/costs/test_calculator_long_ttl.py new file mode 100644 index 000000000..78ed45bcc --- /dev/null +++ b/backend/tests/apis/shared/costs/test_calculator_long_ttl.py @@ -0,0 +1,53 @@ +"""1h static-prefix cache writes are billed at 2x base (thresholds spec §3.6, PR-5).""" + +import pytest + +from apis.shared.costs.calculator import CostCalculator + +PRICING = { + "inputPricePerMtok": 2.0, # base + "outputPricePerMtok": 10.0, + "cacheReadPricePerMtok": 0.2, # 0.1x + "cacheWritePricePerMtok": 2.5, # 1.25x (5m) +} + + +def _write_cost(read, write, static=None): + usage = {"inputTokens": 0, "outputTokens": 0, "cacheReadInputTokens": read, "cacheWriteInputTokens": write} + _, b = CostCalculator.calculate_message_cost(usage, PRICING, long_ttl_static_prefix_tokens=static) + return round(b.cache_write_cost, 9) + + +def test_multiplier_is_bedrocks_1h_premium(): + assert CostCalculator.LONG_TTL_CACHE_WRITE_MULTIPLIER == 2.0 + + +def test_default_prices_every_write_at_5m(): + assert _write_cost(read=0, write=30_000) == pytest.approx(30_000 / 1e6 * 2.5) + assert _write_cost(read=0, write=30_000, static=None) == pytest.approx(30_000 / 1e6 * 2.5) + + +def test_cold_everything_bills_static_at_2x_and_history_at_5m(): + # 10k static + 20k history all written, nothing read. + expected = 10_000 / 1e6 * 4.0 + 20_000 / 1e6 * 2.5 + assert _write_cost(read=0, write=30_000, static=10_000) == pytest.approx(expected) + + +def test_static_read_means_history_only_write_at_5m(): + # Static segment was read (1h entry alive); only history re-wrote. + assert _write_cost(read=10_000, write=20_000, static=10_000) == pytest.approx(20_000 / 1e6 * 2.5) + assert _write_cost(read=12_000, write=20_000, static=10_000) == pytest.approx(20_000 / 1e6 * 2.5) + + +def test_partially_read_static_bills_the_unread_remainder_at_2x(): + # Read 4k of a 10k static segment (tools hit, system missed) → 6k at 2x. + expected = 6_000 / 1e6 * 4.0 + 14_000 / 1e6 * 2.5 + assert _write_cost(read=4_000, write=20_000, static=10_000) == pytest.approx(expected) + + +def test_write_smaller_than_static_is_capped_by_the_write(): + assert _write_cost(read=0, write=3_000, static=10_000) == pytest.approx(3_000 / 1e6 * 4.0) + + +def test_no_write_no_cost(): + assert _write_cost(read=10_000, write=0, static=10_000) == 0.0 diff --git a/docs/kaizen/review-queue.md b/docs/kaizen/review-queue.md index bbadbe4b7..b84ede988 100644 --- a/docs/kaizen/review-queue.md +++ b/docs/kaizen/review-queue.md @@ -210,7 +210,7 @@ Items added by `kaizen-research`, consumed by `kaizen-review-prep`. - **Surface**: backend (`core/model_config.py` cache-point placement + `CacheConfig(strategy="auto")` at L389; `session/turn_based_session_manager.py` truncation anchor + repeated-compaction path; `session/compaction_models.py` `cache_ttl_seconds`; system-prompt assembly) - **Effort × Impact**: L × M–H - **Subtracts**: yes — every item is "find waste and delete it"; no new abstraction, no new dependency -- **Status**: open — **highest confidence-per-hour item in the scan.** Four checks: (1) **byte-stable prefix** — grep system-prompt assembly for time/random/env-derived values (the cookbook measured a **44% swing from one `datetime.now()`**; this also tests whether we're *reading* `systemPromptHash`); (2) **layered mixed-TTL breakpoints** — 54% cheaper upstream; ✅ **the "blocked by Strands #3758" caveat is STRUCK as of 2026-08-28** — the Python-side fix shipped in **1.53.0 via #3858**, so the technique becomes available on the [2026-08-28] Strands 1.51 → 1.54 bump. Sequence it *behind* that bump, since #3681 changes the cache-point placement this would tune; (3) **repeated-compaction audit** — does pass 2 preserve tool pairing and avoid re-writing the prefix? (our death-spiral incident says this is where the money went); (4) **retry cap with jitter** — an uncapped retry on a 424/throttle re-writes the whole prefix per attempt. +- **Status**: open — **(2) layered mixed-TTL is now BUILT behind a flag (2026-09-16, compaction PR-5: `AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL=1h`, default off; gate = `scripts/probe_static_prefix_ttl.py` + a dev week of cost rows, which bill the 1h premium honestly). (3) repeated-compaction audit answered by the compaction stack (#1125→#1131: hysteresis, bounded summary, paid-when-free apply). (1) and (4) still open. Original: **highest confidence-per-hour item in the scan.** Four checks: (1) **byte-stable prefix** — grep system-prompt assembly for time/random/env-derived values (the cookbook measured a **44% swing from one `datetime.now()`**; this also tests whether we're *reading* `systemPromptHash`); (2) **layered mixed-TTL breakpoints** — 54% cheaper upstream; ✅ **the "blocked by Strands #3758" caveat is STRUCK as of 2026-08-28** — the Python-side fix shipped in **1.53.0 via #3858**, so the technique becomes available on the [2026-08-28] Strands 1.51 → 1.54 bump. Sequence it *behind* that bump, since #3681 changes the cache-point placement this would tune; (3) **repeated-compaction audit** — does pass 2 preserve tool pairing and avoid re-writing the prefix? (our death-spiral incident says this is where the money went); (4) **retry cap with jitter** — an uncapped retry on a 424/throttle re-writes the whole prefix per attempt. ### [2026-08-14] Guard against `bedrock-agentcore` #564 — the one failure class the 1.21.0 bump did NOT close - **Source**: research/2026-08-14.md ▸ Community + GitHub issues; the carried-forward residue of the four now-resolved `bedrock-agentcore` bump entries — https://github.com/aws/bedrock-agentcore-sdk-python/issues/564 (**still open**; #482 and #571 closed, #564 did not) @@ -247,7 +247,7 @@ Items added by `kaizen-research`, consumed by `kaizen-review-prep`. - **Surface**: backend (`agent_factory.py` — `plugins=` already plumbed; a custom S3 `Storage` backend is the real build — the plugin ships InMemory/File only, and references must survive AgentCore Runtime restores across turns) - **Effort × Impact**: M–H × M–H (payoff directly measurable by the PR #697 cache/cost metrics) - **Subtracts**: partial — bounds MCP/tool payload growth at the source instead of relying solely on reactive below-anchor truncation in `TurnBasedSessionManager` (which stays for legacy history) -- **Status**: open — spike before commitment; four known gotchas: (1) `evict_after_cycles=20` runs on `BeforeModelCallEvent` and touches *prior* messages — potential byte-stability cache-buster, verify semantics or set `None` and expire via S3 lifecycle; (2) `model.count_tokens` per tool result adds latency, and Bedrock CountTokens rejects `us.*` inference-profile ids (de-prefix precedent in context attribution); (3) adoption flips `toolConfigHash` once (expected; the new tool must land in the deterministically-ordered tool list); (4) check SPA tool-result rendering against placeholder content. +- **Status**: **ADOPTED 2026-09-15** as compaction PR-4 (branch `feature/compaction-offload-escalation`, stacked on #1129; spec `compaction-model-relative-thresholds.md` §3.6). Gotchas as resolved: (1) `evict_after_cycles=None`, expiry via a 90-day S3 lifecycle rule on `compaction-offload/`; (2) a chars/4 pre-filter skips CountTokens for results under half the gate, and `CountTokensBedrockModel` already de-prefixes `us.*`; (3) `retrieve_offloaded_content` lands once in the tool list; (4) the preview is plain text in the tool result. Original entry: spike before commitment; four known gotchas: (1) `evict_after_cycles=20` runs on `BeforeModelCallEvent` and touches *prior* messages — potential byte-stability cache-buster, verify semantics or set `None` and expire via S3 lifecycle; (2) `model.count_tokens` per tool result adds latency, and Bedrock CountTokens rejects `us.*` inference-profile ids (de-prefix precedent in context attribution); (3) adoption flips `toolConfigHash` once (expected; the new tool must land in the deterministically-ordered tool list); (4) check SPA tool-result rendering against placeholder content. ### [2026-07-17] Adopt Strands `Limits` on the unattended Scheduled Runs / headless lane - **Source**: research/2026-07-17.md ▸ Top 5 #3 — convergent harness rail (Claude Code 2.1.212 spawn cap + opencode 1.18.2 `subagent_depth`); Strands `Limits` now available (we're on 1.47). diff --git a/docs/specs/compaction-model-relative-thresholds.md b/docs/specs/compaction-model-relative-thresholds.md index 31eb4d7f5..5a5068003 100644 --- a/docs/specs/compaction-model-relative-thresholds.md +++ b/docs/specs/compaction-model-relative-thresholds.md @@ -1,7 +1,9 @@ # Compaction relative to the model window — a trigger ceiling, a target floor, and paying the rewrite only when it is free -**Status:** PR-1 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), PR-3 (paid-when-free apply, #1129), +PR-4 (tool-result offload at intake, #1131) and PR-5 (selective 1h TTL, +flag off, built 2026-09-16) stacked on top of it. **Owner:** Phil Merrell **Related:** `compaction-over-threshold-cache-spiral.md` (#833 — the incident and the summary-cap PR this spec depends on) · @@ -220,9 +222,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. @@ -245,21 +284,59 @@ Everything above decides *what* to cut. PR-3 decides *when the bytes change*: is unreachable — a 40k-token summary is larger than the 25k floor. PR-1's `compaction_forced` metric will show exactly how often this bites until it lands. -- **Content-class eviction + offload** (PR-4): when the protected tail alone - exceeds the floor, move the oversized tool result or document to the - session workspace behind the retrieval tool and reference it, rather than - cutting deeper or giving up. +- **Offload escalation** (PR-4) — BUILT for tool results, at intake. The + case that defeats the floor is a huge tool result inside the protected + tail. Rather than escalate at cut time (which would mutate a protected turn + and need a restore-replay of every edit), oversized tool results are bounded + the moment they are produced, before they enter the prefix: Strands 1.55's + vended `ContextOffloader` on `AfterToolCallEvent`, S3 storage in the + user-files bucket under `compaction-offload/{userId}/{sessionId}/` (one + namespaced storage per agent, so references are session-scoped by + construction and an `@`-mention agent resolves the same ones), no eviction + from the model path (`evict_after_cycles=None`; a 90-day S3 lifecycle rule + expires the objects), gate 4,000 tokens / preview 1,000 (env-backed; + `AGENTCORE_TOOL_RESULT_OFFLOAD_ENABLED=false` removes the plugin). The + persisted message already carries the bounded form, so restore reproduces + it byte-for-byte. Our subclass adds a chars/4 pre-filter so the plugin's + per-result `CountTokens` round trip only runs for results near or over the + gate, and a content-free `ToolResultOffloaded` record per offload. The + model keeps a preview and pulls spans back with + `retrieve_offloaded_content` (pattern / line range / full) — one stable + spec in `toolConfig`, not an RBAC-gated tool, on the `read_skill_file` + precedent (the `workspace_files` catalog key is granted to no prod role, + so escalating into the workspace tools would have shipped dark). **User + attachments are not touched**: the digest + page-range read in + `document-context-offload.md` is the right shape for those and stays that + spec's PRs 1–4. What remains of the floor-unreachable case after this is + measured by `CompactionFloorUnreachable` on the cut record. - **`context_window_limit` on the model** (PR-3, small): plumb `maxInputTokens` into `ModelConfig` and set `context_window_limit` in `to_bedrock_config` (a valid `BedrockConfig` key in 1.55), so Strands' own `estimate_utilization` and our policy agree on the window, and the eventual v2 engine swap inherits it. -- **Per-section cache TTL** (PR-5, experiment): tools + system on `1h`, - messages on `5m`, via `CacheConfig(system_prompt_ttl="1h", tools_ttl="1h")`. - The 2026-07-27 model said a *blanket* 1h TTL was a wash (2× write premium - ate the saving) and a *selective* one was the variant worth testing. - Sequence behind the 1.55 cache-point invariant test; measure on - `cacheStatus` before and after. +- **Per-section cache TTL** (PR-5) — BUILT, **default off**. `AGENTCORE_PROMPT_CACHE_STATIC_PREFIX_TTL=1h` + makes `ModelConfig.to_bedrock_config` emit + `CacheConfig(system_prompt_ttl="1h", tools_ttl="1h")`; upstream rewrites + the hand-placed TTL-less system point and gives the tools point its own, + the message point stays at 5m (`cache_config.ttl` unset), so the order is + tools(1h) → system(1h) → messages(5m), which Bedrock requires + non-increasing. Anything but the literal `1h` emits exactly today's bytes. + **Why off by default, against the flags-default-on house style:** the + 2026-07-27 model found a *blanket* 1h TTL a wash — the 2× write premium ate + the saving — and only the selective variant worth testing; and the repo's + standing rule since #954/#956 is that a caching default is never adopted on + inspection alone. The gate is `scripts/probe_static_prefix_ttl.py` (two + arms, same static prefix, a configurable gap) plus a week of cost rows in + dev with the flag on. **The experiment arm's rows are priced honestly:** + `CostCalculator.calculate_message_cost` bills the unread remainder of the + static segment at the 1h premium (2× base) and the rest of the write at the + catalog's 5m rate, using the context-attribution breakdown for the static + size; every such row carries `staticPrefixTtl: "1h"` so the anatomy can + split arms. Economics to expect: +0.75× base on every static write, −1.15× + base on every return inside the hour but past five minutes; it pays when + the second is more frequent than the first (the audit's 36% cold re-writes + after a >5 min pause say it might, on the 28k static prefix; the hourly + system-prompt tick would bust it hourly until that fix lands). ## 4. Cost model @@ -321,7 +398,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 +426,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 @@ -347,12 +439,105 @@ on the first turn after a >300s gap and not before (unless hard ceiling); `partial_miss` on the cut turn only; `test_second_cache_key_for_a_session_shares_the_conversation` still passes with the in-place apply. -### PR-4 — content-class eviction and offload escalation (§3.6) - -### PR-5 — selective 1h TTL experiment (§3.6) +### PR-4 — offload escalation (§3.6) — BUILT as tool-result offload at intake + +**Cohort split (2026-09-15 audit, content-free, September prod).** Of the 95 +sessions that peaked over 100k: 52 (55%) had a single tool result ≥4k tokens +in their last three turns (43 only that, 9 also an attachment) — PR-4's +target; 25 (26%) had an attachment there (16 only that) — the document-offload +spec's; 27 (28%) had neither — long sessions whose bulk is old history plus a +23–40k summary, which the cut and the summary cap reach. The biggest single +intra-turn writes are squarely tool results (123k, 149k, 131k in one call); +the attachment-only cases include 1–3-turn sessions at 280–530k that are one +huge upload. Method: turns split at `cacheGapSeconds ≥ 10s`; "big tool result" += an intra-turn call that read the prior prefix and wrote ≥4,000 tokens +(slightly overstated on long `tool_use` blocks); "attachment in the last 3 +turns" = an upload row between the first call of those turns minus 5 min and +the last call. + +### PR-5 — selective 1h TTL experiment (§3.6) — BUILT, flag off; enable per the gate above + +**Probe run 2026-09-16, dev-ai us-west-2, Haiku 4.5, 420 s gap +(`scripts/probe_static_prefix_ttl.py --gap-seconds 420`):** + +| arm | call | read | write | $ (base $1.10/MTok) | +|---|---|---|---|---| +| 5m | first | 0 | 6,251 | 0.008598 | +| 5m | second | 0 | 6,251 | 0.008598 | +| 1h | first | 0 | 6,251 | 0.013756 | +| 1h | second | **5,924** | **327** | 0.001374 | + +Bedrock honors `ttl: "1h"` on the tools and system points: after the 5m +entry expired, the 1h arm **read** the static segment (5,924 tokens) and +re-wrote only the message segment (327), while the 5m arm re-wrote all +6,251. Pair cost $0.01513 vs $0.01720 — the 1h arm was **12% cheaper at this +gap**, having recovered its 60% dearer first write on one return. The +arithmetic that generalizes: the premium is 0.75× base per static write +(~$0.0052 on this prefix), the saving 1.15× base per return inside the hour +but past five minutes (~$0.0075); the arm pays when such returns outnumber +static writes by more than ~0.7 : 1. Static writes also happen on every +prefix *change* (a model or tool-set switch, and today the hourly +system-prompt tick), so the dev-week measurement should run **after** the +tick fix lands. + +**60 s gap, same day:** both arms read the full 6,251 on the second call +(5m: $0.008598 + $0.000691; 1h: $0.013756 + $0.000691). The 1h arm was +**$0.005157 more expensive** — exactly the 0.75× base premium on the first +write, with nothing to recover inside five minutes. So the arm is a fixed +surcharge per static write, ~$0.0052 on this 6.3k prefix (≈$0.023 on a 28k +prod prefix at Haiku rates, ≈$0.042 on Sonnet 5), recovered at ~$0.0075 / +$0.034 / $0.061 per return that lands between five and sixty minutes. The +dev week decides whether prod sessions return in that window often enough; +the probe cannot. ## 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 | +| `CompactionFloorUnreachable` (PR-4) | how often the protected tail alone still exceeds the floor after intake offload — the residual document/attachment case | +| `ToolResultOffloaded`, `ToolResultOffloadedTokens` (PR-4) | how much tool payload was kept out of the prefix, per tool (`toolName` property) | + +Properties (queryable in Logs Insights, not dimensions): `policySource`, +`contextWindow`, `ceiling`, `floor`, `summaryOutcome`, `summaryTokensBefore`. +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`). Partly covered + by PR-4's `ToolResultOffloadedTokens`; the sub-gate long tail is not. +- **Outcome signal joined to compaction** — a down-thumb rate keyed to + "turns since last cut" is the first evidence that a cut costs anything + besides dollars (the kaizen review-queue already lists this as the + 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 799d83068..3f1a0e1f2 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). diff --git a/infrastructure/lib/constructs/data/file-upload-construct.ts b/infrastructure/lib/constructs/data/file-upload-construct.ts index 7e9179b6f..dc1b55686 100644 --- a/infrastructure/lib/constructs/data/file-upload-construct.ts +++ b/infrastructure/lib/constructs/data/file-upload-construct.ts @@ -103,6 +103,20 @@ export class FileUploadConstruct extends Construct { id: 'expire-objects', expiration: cdk.Duration.days(365), }, + { + // Compaction's tool-result offload (backend + // agents/main_agent/core/tool_result_offload.py) parks oversized + // tool results under `compaction-offload/{userId}/{sessionId}/` and + // keeps a retrieval reference in the conversation. The plugin never + // evicts from the model path (that would turn a retrieval into a + // miss mid-turn), so objects expire here instead — 90 days, matching + // AgentCore Memory's conversation retention. The shortest matching + // expiration wins, so this overrides the 365-day rule for the + // prefix only; user uploads are untouched. + id: 'expire-compaction-offload', + prefix: 'compaction-offload/', + expiration: cdk.Duration.days(90), + }, { id: 'abort-incomplete-multipart', abortIncompleteMultipartUploadAfter: cdk.Duration.days(1),