fix: retain active reminders and human boundaries during compaction

.github/workflows/ci.yml

@@ -62,7 +62,11 @@ jobs:
           python-version: ${{ matrix.python-version }}
 
       - name: Install dependencies
-        run: uv sync --all-extras --dev
+        run: uv sync --locked --all-extras --dev
+
+      - name: Verify released Core API
+        run: |
+          uv run --locked python -c "from importlib.metadata import version; from amplifier_core.llm_errors import ContextLengthError; assert version('amplifier-core') == '1.6.1'"
 
       # No API key placeholder here, unlike provider-openai's CI. Measured:
       # with ANTHROPIC_API_KEY/OPENAI_API_KEY unset, the suite is
@@ -72,4 +76,4 @@ jobs:
       # No `-m "not live"` deselection either: this repo registers no `live`
       # marker and has no network-dependent tests.
       - name: Run test suite
-        run: uv run pytest -q
+        run: uv run --locked pytest -q

README.md

@@ -99,11 +99,29 @@ Compaction triggers when token usage reaches the configured threshold (default:
 ### Protected Messages (Never Removed)
 
 - **System messages**: All system messages are always preserved
-- **First user message**: The original task/request is always protected (prevents losing context about what was originally asked)
-- **Last user message**: The most recent user input is always preserved
+- **First human prompt**: The original human task/request is protected, using message metadata rather than treating every user-role message as human input
+- **Last human prompt**: The most recent human input is protected by the same metadata-based classification
 - **Recent messages**: Last N% of messages (configurable via `protected_recent`)
+- **Recent tool results**: The last `protected_tool_results` results (default 5) are protected from both truncation and removal; a protected sibling also prevents removal of its owning call group
 - **Tool pairs**: Tool_use and tool_result messages are treated as atomic units
 
+### Request-scoped retention
+
+The optional `context.request_retention` capability lets an orchestrator name
+the exact persisted reminder bodies required for its next request. The newest
+matching admitted `ephemeral=True, persisted=True` user-role envelope is kept
+complete through compaction, along with the first/latest human prompts.
+Quoting reminder XML in an ordinary human prompt does not change its identity.
+
+This protects delivery in the request view; it does not change message roles,
+pin every historical reminder, or rewrite canonical history. A missing required
+body or an irreducible required set that cannot fit raises `ContextLengthError`
+instead of silently dropping instructions. Failed assembly restores the prior
+compaction state.
+
+Protection takes precedence over the compaction target. A protected tool cohort
+can leave a view above that target; it is not a strict native-token ceiling.
+
 ### Compaction Phases
 
 1. **Phase 1 - Tool Result Truncation**: Older tool results are truncated to reduce token usage
@@ -279,7 +297,18 @@ a corresponding quality regression.
 
 ## Dependencies
 
-- `amplifier-core>=1.0.0`
+- Host-provided `amplifier-core>=1.6.1`, including
+  `amplifier_core.llm_errors.ContextLengthError` for fail-loud request retention.
+  Core remains provided by the host, not installed as a runtime module dependency.
+
+Development and CI pin the released `amplifier-core==1.6.1` package in `uv.lock`.
+A Git `main` source mapping can retain an older commit in the lockfile; the
+release pin ensures tests exercise the Core API required by this module.
+
+```bash
+uv sync --locked --all-extras --dev
+uv run --locked pytest -q
+```
 
 ## Contributing
 

amplifier_module_context_simple/__init__.py

@@ -46,6 +46,7 @@
 from typing import Any
 
 from amplifier_core import ModuleCoordinator, TextBlock
+from amplifier_core.llm_errors import ContextLengthError
 
 logger = logging.getLogger(__name__)
 
@@ -107,6 +108,16 @@ def _carries_loaded_tool_state(msg: dict[str, Any]) -> bool:
     return any(meta.get(key) for key in LOADED_TOOL_STATE_METADATA_KEYS)
 
 
+def _is_human_message(msg: dict[str, Any]) -> bool:
+    """Wire role alone does not distinguish a prompt from an injection."""
+    meta = msg.get("metadata") or {}
+    return (
+        msg.get("role") == "user"
+        and not meta.get("ephemeral")
+        and meta.get("source") not in ("hook", "context-compaction")
+    )
+
+
 async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None = None):
     """
     Mount the simple context manager.
@@ -190,6 +201,12 @@ async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None =
         )
 
     await coordinator.mount("context", context)
+    # Optional module capability; the core Context protocol is unchanged.
+    register_capability = getattr(coordinator, "register_capability", None)
+    if callable(register_capability):
+        register_capability(
+            "context.request_retention", context.get_messages_for_request_retaining
+        )
     logger.info(f"Mounted SimpleContextManager (token_meter={token_meter!r})")
 
     async def cleanup() -> None:
@@ -224,14 +241,14 @@ class SimpleContextManager:
     Level 5: Remove more messages (60% of configured protection)
     Level 6: Truncate remaining tool results (except last N)
     Level 7: Remove more messages (30% of configured protection - last resort)
-    Level 8: Stub first user message + remove old stubs (extreme pressure)
+    Level 8: Stub unprotected machine prefix + remove old stubs (extreme pressure)
 
     This interleaved approach ensures minimal data loss by:
     - Preferring truncation (preserves structure) over removal (loses context)
     - Progressively relaxing protection as pressure increases
     - Respecting configured protected_recent as baseline, only relaxing under pressure
-    - Always protecting: system messages, last user message, last N tool results, tool pairs
-    - First user message: stubbable at Level 8, but never fully removed
+    - Always protecting: system messages, first/last human prompts, last N tool results, tool pairs
+    - Requested active persisted injections remain complete through all levels
     """
 
     def __init__(
@@ -328,6 +345,8 @@ def __init__(
         self._last_measured_prompt_tokens: int | None = None
         self._last_token_meter_stats: dict[str, Any] | None = None
         self._system_prompt_factory: Callable[[], Awaitable[str]] | None = None
+        self._request_retained_contents: frozenset[str] = frozenset()
+        self._request_protected_seqs: set[int] = set()
 
         # --- Sticky compaction decision state ---
         # Compaction decisions (remove / truncate / stub) are keyed by a
@@ -553,6 +572,82 @@ async def set_system_prompt_factory(
         self._system_prompt_factory = factory
         logger.info("System prompt factory registered - will refresh on each request")
 
+    async def get_messages_for_request_retaining(
+        self,
+        *,
+        retain_contents: list[str],
+        provider: Any | None = None,
+        token_budget: int | None = None,
+    ) -> list[dict[str, Any]]:
+        """Optional capability: retain current persisted injections for one view.
+
+        Contents must exactly match admitted user-role messages marked
+        ephemeral=True and persisted=True. Only the newest matching copy is
+        protected. This neither admits messages nor changes their lifetime;
+        the caller supplies the current delivery requirements on every call.
+        """
+        previous_contents = self._request_retained_contents
+        previous_seqs = self._request_protected_seqs
+        decisions = (
+            self._removed_seqs.copy(),
+            self._truncated_seqs.copy(),
+            self._stubbed_seqs.copy(),
+            self._sticky_level,
+            self._last_compaction_stats,
+            self._last_token_meter_stats,
+        )
+        try:
+            self._request_retained_contents = frozenset(retain_contents)
+            return await self.get_messages_for_request(token_budget, provider)
+        except BaseException:
+            # A failed/cancelled assembly must not commit a reduction that was
+            # never delivered. Canonical history is unchanged throughout.
+            (
+                self._removed_seqs,
+                self._truncated_seqs,
+                self._stubbed_seqs,
+                self._sticky_level,
+                self._last_compaction_stats,
+                self._last_token_meter_stats,
+            ) = decisions
+            raise
+        finally:
+            self._request_retained_contents = previous_contents
+            self._request_protected_seqs = previous_seqs
+
+    def _protected_sequences(self, messages: list[dict[str, Any]]) -> set[int]:
+        humans = [msg for msg in messages if _is_human_message(msg)]
+        protected = [humans[0], humans[-1]] if humans else []
+        remaining = set(self._request_retained_contents)
+        for msg in reversed(messages):
+            meta = msg.get("metadata") or {}
+            content = msg.get("content")
+            if (
+                msg.get("role") == "user"
+                and meta.get("ephemeral") is True
+                and meta.get("persisted") is True
+                and isinstance(content, str)
+                and content in remaining
+            ):
+                protected.append(msg)
+                remaining.remove(content)
+        if remaining:
+            raise ValueError("Requested retained injection is not in admitted history")
+        return {seq for msg in protected if (seq := self._extract_seq(msg)) is not None}
+
+    def _is_request_protected(self, msg: dict[str, Any]) -> bool:
+        return self._extract_seq(msg) in self._request_protected_seqs
+
+    def _check_retained_budget(
+        self, messages: list[dict[str, Any]], budget: int
+    ) -> None:
+        if self._request_retained_contents and self._estimate_tokens(messages) > budget:
+            raise ContextLengthError(
+                "Context cannot fit the current injections and protected conversation "
+                "within the estimated input budget; shorten the active instructions "
+                "or use a larger context window. Required content was not discarded."
+            )
+
     async def get_messages_for_request(
         self,
         token_budget: int | None = None,
@@ -629,6 +724,16 @@ async def get_messages_for_request(
             # Static mode: use messages as-is (may include stored system messages)
             working_messages = list(self.messages)
 
+        self._request_protected_seqs = self._protected_sequences(working_messages)
+        self._check_retained_budget(
+            [
+                m
+                for m in working_messages
+                if m.get("role") == "system" or self._is_request_protected(m)
+            ],
+            effective_budget,
+        )
+
         token_count, meter_source, estimated_tokens = self._measure_working_tokens(
             working_messages
         )
@@ -643,7 +748,11 @@ async def get_messages_for_request(
         }
 
         # Check if compaction needed (using effective budget with notice reserve deducted)
-        if self._should_compact(token_count, effective_budget):
+        retained_view_over_budget = (
+            bool(self._request_retained_contents)
+            and estimated_tokens > effective_budget
+        )
+        if self._should_compact(token_count, effective_budget) or retained_view_over_budget:
             # Compact EPHEMERALLY - returns new list, working_messages unchanged
             compacted = await self._compact_ephemeral(
                 effective_budget, working_messages
@@ -731,8 +840,10 @@ async def get_messages_for_request(
             # Strip internal bookkeeping at the module boundary -- everything
             # above this point (sticky decisions, token accounting) still runs
             # on messages carrying `_seq`; only what leaves has it removed.
+            self._check_retained_budget(compacted, budget)
             return self._strip_internal_metadata(compacted)
 
+        self._check_retained_budget(working_messages, budget)
         return self._strip_internal_metadata(working_messages)
 
     # Metadata keys that are internal bookkeeping only and must never cross
@@ -876,6 +987,8 @@ def _exceeds_threshold(self, estimated_tokens: int, budget: int) -> bool:
         """
         if budget <= 0:
             return False
+        if self._request_retained_contents and estimated_tokens > budget:
+            return True
         if (
             self.token_meter == TOKEN_METER_ACTUAL
             and self._last_measured_prompt_tokens is not None
@@ -1037,6 +1150,9 @@ def _apply_sticky_decisions(
         result: list[dict[str, Any]] = []
         for msg in messages:
             seq = self._extract_seq(msg)
+            if self._is_request_protected(msg):
+                result.append(dict(msg))
+                continue
             if seq is not None and seq in self._removed_seqs:
                 continue
             if seq is not None and seq in self._truncated_seqs:
@@ -1079,6 +1195,7 @@ async def _compact_ephemeral(
         messages_to_compact = (
             source_messages if source_messages is not None else self.messages
         )
+        self._request_protected_seqs = self._protected_sequences(messages_to_compact)
         target_tokens = int(budget * self.target_usage)
         old_count = len(messages_to_compact)
         old_tokens = self._estimate_tokens(messages_to_compact)
@@ -1420,10 +1537,10 @@ async def _compact_ephemeral(
 
         # Check if we still need more space
         if current_tokens > target_tokens:
-            # === LEVEL 8: Stub first user message + remove old stubs (extreme pressure) ===
+            # === LEVEL 8: Stub unprotected machine prefix + remove old stubs (extreme pressure) ===
             max_level_reached = 8
 
-            # Find first user message and stub it if not already stubbed
+            # Find first and last user messages.
             first_user_idx = None
             last_user_idx = None
             for i, msg in enumerate(working_messages):
@@ -1432,9 +1549,14 @@ async def _compact_ephemeral(
                         first_user_idx = i
                     last_user_idx = i
 
-            # Stub first user message (previously protected) - but NEVER if it's also the last
-            # The last user message is the current intent and must always be preserved
-            if first_user_idx is not None and first_user_idx != last_user_idx:
+            # An unprotected machine prefix can still be reduced at extreme pressure.
+            # Human boundary prompts and current retained injections stay complete.
+            if (
+                first_user_idx is not None
+                and first_user_idx != last_user_idx
+                and not self._is_request_protected(working_messages[first_user_idx])
+                and not _is_human_message(working_messages[first_user_idx])
+            ):
                 first_msg = working_messages[first_user_idx]
                 if not first_msg.get("_stubbed"):
                     content = first_msg.get("content", "")
@@ -1462,6 +1584,7 @@ async def _compact_ephemeral(
                     if msg.get("_stubbed")
                     and i < protected_boundary  # Outside protected recent zone
                     and i != last_user_idx  # Never remove last user message
+                    and not self._is_request_protected(msg)
                 ]
 
                 stubs_removed = 0
@@ -1625,11 +1748,11 @@ def _remove_messages_with_protection(
             i for i, msg in enumerate(messages) if msg.get("role") == "user"
         }
 
-        # Find first and last user message indices (always fully protected from stubbing too)
+        # Human boundaries must not be displaced by machine user-role messages.
         first_user_idx = None
         last_user_idx = None
         for i, msg in enumerate(messages):
-            if msg.get("role") == "user":
+            if _is_human_message(msg):
                 if first_user_idx is None:
                     first_user_idx = i
                 last_user_idx = i
@@ -1654,11 +1777,11 @@ def _remove_messages_with_protection(
                 len(loaded_tool_state_indices),
             )
 
-        # First user message is stubbable at extreme pressure (Level 8), but never fully removed
-        # (It's excluded from removal_candidates via user_message_indices, but can be stubbed)
-        # We don't add it to protected_indices so it can be stubbed at Level 8
-
-        # Always protect the LAST user message (current context)
+        protected_indices.update(
+            i for i, msg in enumerate(messages) if self._is_request_protected(msg)
+        )
+        if first_user_idx is not None:
+            protected_indices.add(first_user_idx)
         if last_user_idx is not None:
             protected_indices.add(last_user_idx)
 
@@ -1667,6 +1790,13 @@ def _remove_messages_with_protection(
         for i in range(protected_boundary, len(messages)):
             protected_indices.add(i)
 
+        # The last N tool results are protected from removal as well as truncation.
+        # Their owning assistant and sibling results are vetoed atomically below.
+        tool_result_indices = [
+            i for i, msg in enumerate(messages) if msg.get("role") == "tool"
+        ]
+        protected_indices |= self._protected_tool_indices(tool_result_indices)
+
         # Removal candidates exclude ALL user messages (they can only be stubbed, not removed)
         removal_candidates = [
             i

pyproject.toml

@@ -30,11 +30,8 @@ packages = [
 [tool.hatch.metadata]
 allow-direct-references = true
 
-[tool.uv.sources]
-amplifier-core = { git = "https://github.com/microsoft/amplifier-core", branch = "main" }
-
 [dependency-groups]
-dev = ["amplifier-core", "pytest>=9.0.3", "pytest-asyncio>=0.23.0"]
+dev = ["amplifier-core==1.6.1", "pytest>=9.0.3", "pytest-asyncio>=0.23.0"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]