test(compaction): trim typed-retention tests to two invariants

Keep the converter->pruner checkpoint/no-checkpoint case (the issue's
headline: an older completed ask must not survive while the newest
oversized typed ask is dropped) and one boundary case covering multipart
clipping, the token-estimator budget for CJK text, and an oversized
mixed text+image boundary that now retains no older ask (behaviour change:
previously the oversized boundary was skipped and an older ask kept).

Also drop the provenance-footer disclaimer from the docstring.

Parallel fix in #131819 by liuhao1024 credited here.

Co-authored-by: liuhao1024 <sunsky.lau@gmail.com>

agent/native_compaction.py

@@ -249,8 +249,6 @@ def prune_pre_checkpoint_items(
       ``input_text`` parts, preserving part metadata). An oversized unsupported/mixed-content
       boundary stops older user retention rather than substituting an older ask. A recognized
       image-only user message is retained whole at one-token cost.
-      Head truncation can still omit a trailing internal provenance footer; no footer recovery
-      or semantic classification is attempted.
     - Summaries are retained whole within ``retained_summary_token_budget``, never sliced
       (framing would corrupt) and never duplicated.
     - ``item_sources`` (parallel to ``items``) is the raw chat message each item came from.

tests/agent/test_native_compaction_typed_user_retention.py

@@ -1,8 +1,4 @@
-"""Retain the newest plaintext ask on the actual Codex typed-content path.
-
-These synthetic oversized asks require a native checkpoint; they do not model
-normal bounded tool receipts or establish a historical session failure.
-"""
+"""Retain the newest typed (input_text) ask at a native checkpoint boundary."""
 
 from copy import deepcopy
 
@@ -19,18 +15,13 @@ from agent.native_compaction import (
 _CHECKPOINT = {"type": "compaction", "encrypted_content": "synthetic_checkpoint"}
 
 
-@pytest.mark.parametrize("issuer", ["codex_backend", None], ids=["typed-codex", "string-control"])
 @pytest.mark.parametrize("checkpoint", [True, False], ids=["checkpoint", "no-checkpoint"])
-@pytest.mark.parametrize("multipart", [False, True], ids=["single-text", "multipart"])
-def test_converter_retains_newest_oversized_ask_not_completed_older_ask(issuer, checkpoint, multipart):
-    prefix = "CURRENT ASK: inspect this input. "
-    body = "x" * (RETAINED_USER_MESSAGE_TOKEN_BUDGET * 4)
-    current = prefix + body
-    content = [{"type": "text", "text": prefix}, {"type": "text", "text": body}] if multipart else current
+def test_converter_retains_newest_oversized_typed_ask_not_older_ask(checkpoint):
+    current = "CURRENT ASK: inspect this input. " + "x" * (RETAINED_USER_MESSAGE_TOKEN_BUDGET * 4)
     messages = [
         {"role": "user", "content": "OLDER ASK: already completed"},
         {"role": "assistant", "content": "Completed the older ask."},
-        {"role": "user", "content": content},
+        {"role": "user", "content": current},
         {
             "role": "assistant",
             "content": "Continuing the current ask.",
@@ -40,98 +31,52 @@ def test_converter_retains_newest_oversized_ask_not_completed_older_ask(issuer,
     original = deepcopy(messages)
 
     converted = _chat_messages_to_responses_input(
-        messages, current_issuer_kind=issuer, native_compaction_eligible=True,
+        messages, current_issuer_kind="codex_backend", native_compaction_eligible=True,
     )
     users = [item for item in converted if item.get("role") == "user"]
-    texts = [
-        "".join(part["text"] for part in item["content"])
-        if isinstance(item["content"], list) else item["content"]
-        for item in users
-    ]
+    assert all(part["type"] == "input_text" for item in users for part in item["content"])
+    texts = ["".join(part["text"] for part in item["content"]) for item in users]
 
     if checkpoint:
         assert converted[0] == _CHECKPOINT
-        assert len(users) == 1  # An older completed ask must not become the sole anchor.
-        assert texts[0].startswith("CURRENT ASK:")
+        assert len(texts) == 1 and texts[0].startswith("CURRENT ASK:")
         assert current.startswith(texts[0]) and texts[0] != current
         assert 0 < estimate_tokens_rough(texts[0]) <= RETAINED_USER_MESSAGE_TOKEN_BUDGET
-        assert converted[-1]["role"] == "assistant"
     else:
         assert texts == [messages[0]["content"], current]
-    for item, text in zip(users, texts):
-        if issuer == "codex_backend" or (multipart and text.startswith("CURRENT ASK:")):
-            assert isinstance(item["content"], list)
-            assert all(part["type"] == "input_text" for part in item["content"])
-        else:
-            assert isinstance(item["content"], str)
     assert messages == original
 
 
+_IMAGE = {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}
+
+
 @pytest.mark.parametrize(
     "content,budget,expected",
     [
-        ([{"type": "input_text", "text": "abcd"}], 2, "whole"),
-        ([{"type": "input_text", "text": "abcdefgh"}], 1, ("abcd",)),
         ([{"type": "input_text", "text": "abcd"},
           {"type": "input_text", "text": "efghijkl"},
           {"type": "input_text", "text": "trailing"}], 2, ("abcd", "efgh")),
         ([{"type": "input_text", "text": "你好世界"}], 2, ("你好",)),
-        ([{"type": "input_text", "text": "абвгдежз"}], 2, ("абвг",)),
-        ([{"type": "input_text", "text": "🌍🌍🌍"}], 2, ("🌍🌍",)),
-        ([{"type": "input_text", "text": "abcd" + " " * 40}], 2, ("abcd    ",)),
-        ([{"type": "input_text", "text": "a"},
-          {"type": "input_text", "text": "b"},
-          {"type": "input_text", "text": "c"}], 2, ("a", "b")),
-        ([{"type": "input_text", "text": ""},
-          {"type": "input_text", "text": "abcdefgh"}], 2, ("", "abcd")),
-        ([], 1, "older"),
-        ([{"type": "input_text", "text": ""}], 1, "older"),
-        ([{"type": "input_text", "text": None}], 1, "older"),
-        ([{"type": "unknown", "value": "not text"}], 1, "older"),
-        ([{"type": "unknown", "text": "CURRENT" * 20}], 2, "none"),
-        ([{"type": "input_text", "text": "CURRENT" * 20},
-          {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}], 2, "none"),
-        ([{"type": "input_text", "text": "abcd"},
-          {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}], 1, "whole"),
-        ([{"type": "input_text", "text": "CURRENT"}], 0, "none"),
-    ],
-    ids=[
-        "fits-whole", "single-boundary", "multipart-boundary", "cjk", "cyrillic", "emoji",
-        "whitespace-budget", "part-rounding", "empty-leading-part", "empty-list", "empty-text",
-        "malformed-text", "unknown-part", "oversized-unsupported", "oversized-mixed-image",
-        "mixed-image-fits", "zero-budget",
+        ([{"type": "input_text", "text": "CURRENT" * 20}, _IMAGE], 2, None),
     ],
+    ids=["multipart-boundary", "cjk-estimator-budget", "oversized-mixed-image"],
 )
-def test_typed_retention_is_budgeted_copy_safe_and_conservative(content, budget, expected):
+def test_typed_boundary_is_budgeted_and_never_substitutes_older_ask(content, budget, expected):
     older = {"role": "user", "content": "old"}
-    current = {
-        "type": "message", "role": "user", "id": "current-user",
-        "metadata": {"source": "synthetic"}, "content": deepcopy(content),
-    }
+    current = {"type": "message", "role": "user", "id": "current-user", "content": deepcopy(content)}
     for index, part in enumerate(current["content"]):
-        part["metadata"] = {"index": index, "annotations": ["preserve"]}
-    tail = {"role": "assistant", "content": "untouched tail"}
-    items = [older, current, _CHECKPOINT, tail]
+        part["metadata"] = {"index": index}
+    items = [older, current, _CHECKPOINT]
     original = deepcopy(items)
 
-    pruned = prune_pre_checkpoint_items(items, retained_user_token_budget=budget)
+    users = [i for i in prune_pre_checkpoint_items(items, retained_user_token_budget=budget)
+             if i.get("role") == "user"]
 
-    assert pruned[0] is _CHECKPOINT and pruned[-1] is tail
-    users = [item for item in pruned if item.get("role") == "user"]
-    if expected == "older":
-        assert users == [older]  # Unmeasurable placeholders retain the existing skip policy.
-    elif expected == "none":
-        assert users == []  # Never substitute an older ask for an oversized unsupported one.
+    if expected is None:
+        assert users == []  # Never substitute an older ask for an oversized mixed one.
     else:
-        assert users and users[-1].get("id") == current["id"]
-        if expected == "whole":
-            assert users[-1] is current
-        else:
-            assert users == [{**current, "content": [
-                {**current["content"][index], "text": text}
-                for index, text in enumerate(expected)
-            ]}]
-            assert users[0] is not current
-            assert sum(max(1, estimate_tokens_rough(part["text"]))
-                       for part in users[0]["content"]) <= budget
+        assert users == [{**current, "content": [
+            {**current["content"][index], "text": text} for index, text in enumerate(expected)
+        ]}]
+        assert sum(max(1, estimate_tokens_rough(p["text"])) for p in users[0]["content"]) <= budget
     assert items == original