Exclude canonical public history from shared request views

README.md

@@ -462,3 +462,15 @@ trademarks or logos is subject to and must follow
 [Microsoft's Trademark & Brand Guidelines](https://www.microsoft.com/legal/intellectualproperty/trademarks/usage/general).
 Use of Microsoft trademarks or logos in modified versions of this project must not cause confusion or imply Microsoft sponsorship.
 Any use of third-party trademarks or logos are subject to those third-party's policies.
+
+### Canonical public history and execution views
+
+`amplifier_module_context_simple.request_view.request_view` is the shared
+projection for request, summary and token-estimate consumers. It excludes
+`amplifier_public_message`, `amplifier_public_copy_source` and
+`amplifier_public_reference_only` metadata from copies. A row whose
+`amplifier_public_reference_only` is exactly `True` is omitted from execution.
+Canonical `get_messages()` history is retained unchanged. Sequence IDs remain
+available while fitting and are stripped only at the existing final request
+boundary; unrelated metadata retains its previous behavior. This helper does
+not validate originals, store messages, negotiate transport or grant access.

amplifier_module_context_simple/__init__.py

@@ -49,6 +49,7 @@
 from amplifier_core.llm_errors import ContextLengthError
 
 from ._text_estimate import estimate_messages
+from .request_view import request_view
 
 logger = logging.getLogger(__name__)
 
@@ -1156,6 +1157,8 @@ async def get_measured_request_view(
             else:
                 working = list(self.messages)
 
+            working = request_view(working)
+
             developer_seqs = {
                 seq
                 for message in working
@@ -1610,6 +1613,7 @@ async def get_messages_for_request(
             # Static mode: use messages as-is (may include stored system messages)
             working_messages = list(self.messages)
 
+        working_messages = request_view(working_messages)
         self._request_protected_seqs = self._protected_sequences(working_messages)
         self._check_retained_budget(
             [
@@ -1760,44 +1764,11 @@ async def get_messages_for_request(
     # compaction identity (see _extract_seq): meaningless to a provider, and --
     # because _estimate_tokens stringifies the whole message dict -- it also
     # inflates the token estimate of every message carrying it.
-    _INTERNAL_METADATA_KEYS = frozenset({"_seq"})
-
     def _strip_internal_metadata(
         self, messages: list[dict[str, Any]]
     ) -> list[dict[str, Any]]:
-        """Return a provider-facing view with internal-only metadata removed.
-
-        CRITICAL: stored history must KEEP `_seq` -- the sticky decision store
-        is keyed on it, so losing it would silently break stickiness (and with
-        it, prefix stability). The returned view can share dict objects with
-        `self.messages`: the no-compaction path returns stored dicts directly,
-        and even the compacted path's `dict(msg)` shallow copies share the SAME
-        nested metadata dict. So this NEVER mutates in place -- any message
-        needing a strip is rebuilt as a new dict with a new metadata dict, and
-        the stored original is left untouched.
-
-        Messages with nothing to strip pass through by identity (no copy),
-        which keeps this deterministic and byte-stable call over call.
-        """
-        result: list[dict[str, Any]] = []
-        for msg in messages:
-            meta = msg.get("metadata")
-            if not isinstance(meta, dict) or self._INTERNAL_METADATA_KEYS.isdisjoint(
-                meta
-            ):
-                result.append(msg)
-                continue
-            result.append(
-                {
-                    **msg,
-                    "metadata": {
-                        k: v
-                        for k, v in meta.items()
-                        if k not in self._INTERNAL_METADATA_KEYS
-                    },
-                }
-            )
-        return result
+        """Final execution projection, after sequence-based fitting decisions."""
+        return request_view(messages, strip_sequence=True)
 
     async def get_messages(self) -> list[dict[str, Any]]:
         """

amplifier_module_context_simple/_text_estimate.py

@@ -8,6 +8,8 @@
 
 from typing import Any, NamedTuple
 
+from .request_view import request_view
+
 
 class TextEstimate(NamedTuple):
     tokens: int
@@ -17,7 +19,7 @@ class TextEstimate(NamedTuple):
 def estimate_messages(messages: list[dict[str, Any]]) -> TextEstimate:
     tokens = 0
     has_images = False
-    for message in messages:
+    for message in request_view(messages):
         content = message.get("content")
         if isinstance(content, list):
             content, found = _content_without_image_payloads(content)

amplifier_module_context_simple/request_view.py

@@ -0,0 +1,40 @@
+"""Execution views of canonical history, shared with summary/count consumers.
+
+Public originals and copy lineage are history facts, never model context. The
+history-only flag omits its row from execution; it does not delete the original.
+Projection copies changed metadata and leaves canonical rows untouched. Other
+metadata retains its existing behavior. No envelope parsing or storage lives here.
+"""
+
+from typing import Any
+
+
+_HISTORY_METADATA = frozenset({
+    "amplifier_public_message",
+    "amplifier_public_copy_source",
+    "amplifier_public_reference_only",
+})
+
+
+def request_view(
+    messages: list[dict[str, Any]], *, strip_sequence: bool = False
+) -> list[dict[str, Any]]:
+    """Project history for execution; keep sequence IDs until fitting is done."""
+    excluded = _HISTORY_METADATA | {"_seq"} if strip_sequence else _HISTORY_METADATA
+    result = []
+    for message in messages:
+        metadata = message.get("metadata")
+        if not isinstance(metadata, dict):
+            result.append(message)
+            continue
+        if metadata.get("amplifier_public_reference_only") is True:
+            continue
+        if excluded.isdisjoint(metadata):
+            result.append(message)
+            continue
+        result.append({
+            **message,
+            "metadata": {key: value for key, value in metadata.items()
+                         if key not in excluded},
+        })
+    return result

tests/test_public_history_request_view.py

@@ -0,0 +1,90 @@
+"""Canonical public originals never become execution context or token pressure."""
+
+import copy
+
+import pytest
+
+from amplifier_module_context_simple import SimpleContextManager
+from amplifier_module_context_simple._text_estimate import estimate_messages
+
+
+ORIGINAL = "PUBLIC_ORIGINAL_SENTINEL" * 10000
+HISTORY_ONLY = "HISTORY_ONLY_EXECUTION_SENTINEL" * 10000
+
+
+def rows():
+    return [
+        {"role": "user", "content": "expanded execution", "metadata": {
+            "amplifier_public_message": {"version": 1, "blocks": [ORIGINAL]},
+            "amplifier_public_copy_source": {"source": ORIGINAL},
+            "legacy": {"keep": "unrelated metadata"},
+        }},
+        {"role": "assistant", "content": "execution response"},
+        {"role": "user", "content": HISTORY_ONLY, "metadata": {
+            "amplifier_public_reference_only": True,
+            "amplifier_public_message": {"blocks": [ORIGINAL]},
+        }},
+    ]
+
+
+def assert_execution(view):
+    assert ORIGINAL not in str(view)
+    assert HISTORY_ONLY not in str(view)
+    assert "amplifier_public_" not in str(view)
+    assert view[0]["content"] == "expanded execution"
+    assert view[0]["metadata"]["legacy"] == {"keep": "unrelated metadata"}
+    assert len(view) == 2
+
+
+def test_one_projection_preserves_legacy_identity_and_canonical_metadata():
+    from amplifier_module_context_simple.request_view import request_view
+    original = rows()
+    before = copy.deepcopy(original)
+    projected = request_view(original)
+    assert_execution(projected)
+    assert original == before
+    projected[0]["metadata"]["new"] = "view only"
+    assert "new" not in original[0]["metadata"]
+    legacy = {"role": "user", "content": "plain", "metadata": {"unrelated": 7}}
+    assert request_view([legacy])[0] is legacy
+    false_flag = {"role": "user", "content": "keep", "metadata": {
+        "amplifier_public_reference_only": False}}
+    assert request_view([false_flag])[0]["content"] == "keep"
+
+
+def test_text_estimate_ignores_history_facts_and_history_only_rows():
+    original = rows()
+    expected = [
+        {"role": "user", "content": "expanded execution", "metadata": {
+            "legacy": {"keep": "unrelated metadata"}}},
+        {"role": "assistant", "content": "execution response"},
+    ]
+    assert estimate_messages(original) == estimate_messages(expected)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("route", ["ordinary", "retaining", "measured"])
+async def test_all_request_paths_exclude_originals_before_fitting_and_counting(route):
+    context = SimpleContextManager(max_tokens=300, token_meter="actual",
+                                   compaction_notice_enabled=False)
+    await context.set_messages(rows())
+    canonical = copy.deepcopy(await context.get_messages())
+    counted = []
+    async def count(view):
+        assert_execution(view)
+        counted.append(view)
+        return {"dispatch": object(), "budget_decision": {
+            "estimated_input_tokens": 30, "input_limit_tokens": 300,
+            "measurement": {"kind": "provider_count", "source": "fixture", "input_tokens": 30}}}
+    if route == "ordinary":
+        view = await context.get_messages_for_request()
+    elif route == "retaining":
+        view = await context.get_messages_for_request_retaining(retain_contents=[])
+    else:
+        result = await context.get_measured_request_view(
+            provider=None, retain_contents=[], count_view=count)
+        view = result["base_view"]
+        assert counted
+    assert_execution(view)
+    assert context._last_compaction_stats is None
+    assert await context.get_messages() == canonical