feat(context): let the host word a Tier 1 card's recovery footer

CHANGELOG.md

@@ -7,6 +7,46 @@ the GitHub Release body, so a release with no entry here fails.
 
 Versioning follows [docs/versioning.md](docs/versioning.md).
 
+## [0.8.0] - 2026-09-05
+
+### Added
+
+- `KeepLastNToolResultsCompactor` takes a `recovery_footer` callable that renders
+  the last line of a Tier 1 mini card — the handle that fetches the discarded body
+  back. The default, `default_recovery_footer`, is the existing
+  `[Full text] <handle>`, so no consumer changes behaviour by upgrading.
+
+  Rationale: a Tier 1 card *replaces* the whole tool message, so unlike the
+  loop's own truncation footer (`_spill_footer`, which appends to a surviving
+  body) there is nothing left beside the handle to tell the model what it is for.
+  On a Tier1-only turn — Tier 2 fires only when Tier 1 did not free enough — the
+  model therefore sees bare handles and is never told they are recoverable at all.
+
+  The default cannot fix that itself: this module knows neither what a host calls
+  its recovery tool nor whether that tool is bound for the agent whose history it
+  is compacting, and a footer naming a tool the agent cannot call is worse than no
+  footer (the same reason `_spill_footer` gates its prose on the tool map). A host
+  that does know both should pass `recovery_footer`.
+
+  Write host prose as prose. `recover_result(spill_id="...")` renders to the model
+  as source code and it answers in kind: on a live run a model reproduced such a
+  footer inside a ```bash block instead of emitting a tool call, and
+  `LeakedToolCallRetryObserver` fired twice. Name the tool and its argument in
+  words.
+
+  Idempotency is unaffected — the already-carded check keys on
+  `OMITTED_TOOL_RESULT_PLACEHOLDER`, not on the footer's wording — and the footer
+  is called only when a body actually spilled, so a host renderer is never asked
+  to point at nothing.
+
+- `TieredCompactor` takes and forwards the same `recovery_footer` to the Tier 1
+  compactor it builds internally. Tier 1 is a winning candidate in its own right,
+  so its card reaches the model from the tiered path exactly as from the
+  standalone one. Without the forward, a workflow whose primary path is tiered and
+  whose fallback is standalone would render the handle-only default on every
+  production turn and the host's prose only in the fallback — the harder of the
+  two failures to notice.
+
 ## [0.7.0] - 2026-09-04
 
 ### Changed

agent_core/runtime/loop/compact.py

@@ -30,6 +30,7 @@
     "StringSliceCompactor",
     "compact_messages",
     "compress_tool_results",
+    "default_recovery_footer",
     "estimate_tokens",
     "partition_for_compaction",
     "tool_names_by_call_id",
@@ -77,6 +78,28 @@
 URL_RE = re.compile(r'https?://[^\s\)>"\'<]+')
 _TOOL_RESULT_COMPACT_MAX_CHARS = 1_200
 
+
+def default_recovery_footer(spill_path: str) -> str:
+    """Render the last line of a mini card: the handle that fetches the body back.
+
+    Deliberately says nothing about HOW to fetch it. A host's recovery tool is
+    the host's own — it may be named anything, take different arguments, or not
+    be bound for this agent at all — and a footer naming a tool the agent cannot
+    call is worse than no footer (see ``_spill_footer``, which gates its own
+    prose on the tool being in the tool map for exactly that reason). This
+    default therefore carries only the handle, which is correct everywhere.
+
+    Hosts that DO bind a recovery tool, and know it is bound for this agent,
+    should pass ``recovery_footer`` to say so in prose: a card is the only thing
+    left on the message, so unlike ``_spill_footer``'s site there is no
+    surviving instruction next to it telling the model what the handle is for.
+    Write that prose as prose. ``recover_result(spill_id="...")`` renders to the
+    model as source code, and it responds in kind — on a live run it reproduced
+    such a footer inside a ```bash block instead of emitting a tool call, and
+    ``LeakedToolCallRetryObserver`` fired twice.
+    """
+    return f"[Full text] {spill_path}"
+
 # Header of the spill recovery index. This is presentation only — the text the
 # MODEL reads above the paths — since the index is identified by
 # ``Message.spill_refs``. The two remaining substring checks against it
@@ -574,6 +597,11 @@ class KeepLastNToolResultsCompactor:
 
     ``keep_tool_result == -1`` disables filtering entirely.
 
+    ``recovery_footer`` renders that last line. The default carries the handle
+    and nothing else, because this module cannot know what a host's recovery tool
+    is called or whether it is bound; a host that does know should pass prose
+    naming it. See :func:`default_recovery_footer`.
+
     Caveat: only ``ToolMessage`` content is redacted. Workflows that
     inject large content as ``HumanMessage`` (e.g. an observer that
     splices fan-in reports between turns) bypass this compactor; pair
@@ -586,6 +614,7 @@ def __init__(
         keep_tool_result: int,
         protect_tool_names: frozenset[str] = frozenset(),
         spill: Callable[[str, str], str | None] | None = None,
+        recovery_footer: Callable[[str], str] = default_recovery_footer,
     ) -> None:
         if keep_tool_result < -1:
             raise ValueError(f"keep_tool_result must be >= -1 (got {keep_tool_result})")
@@ -597,6 +626,7 @@ def __init__(
         # age only, so existing callers are unaffected.
         self._protect = frozenset(protect_tool_names)
         self._spill = spill
+        self._recovery_footer = recovery_footer
 
     def compact(
         self,
@@ -655,7 +685,7 @@ def compact(
             if card:
                 placeholder += "\n" + card
             if spill_path:
-                placeholder += f"\n[Full text] {spill_path}"
+                placeholder += "\n" + self._recovery_footer(spill_path)
             # Without a pointer, replacing the body DESTROYS it, so a card that
             # is not even shorter is a pure loss and we keep the body. With one
             # we always replace, even when the card is longer: the pointer only

agent_core/runtime/loop/tiered_compact.py

@@ -18,6 +18,7 @@
     INPUT_ESTIMATE_KEY,
     KeepLastNToolResultsCompactor,
     compress_tool_results,
+    default_recovery_footer,
     estimate_tokens,
     tool_names_by_call_id,
 )
@@ -215,6 +216,7 @@ def __init__(
         summary_retries: int = 2,
         summary_retry_timeout_s: float | None = None,
         prompt_builder: SummaryPromptBuilder = compaction_prompt,
+        recovery_footer: Callable[[str], str] = default_recovery_footer,
         manifest_max_paths: int | None = _SPILL_MANIFEST_MAX_PATHS,
         manifest_max_chars: int | None = _SPILL_MANIFEST_MAX_CHARS,
     ) -> None:
@@ -233,10 +235,16 @@ def __init__(
         spill_callback = spill or (
             spill_store.spill_compacted_body if spill_store is not None else None
         )
+        # Forwarded, not defaulted here: Tier 1's card is the ONLY thing left on
+        # a message it rewrites, and Tier 1 is a winning candidate in its own
+        # right (``best_label == "tier1"``), so a host that worded the footer for
+        # the standalone compactor must get the same wording through this one or
+        # the same run renders two different footers.
         self._tier1 = KeepLastNToolResultsCompactor(
             keep_tool_result=keep_tool_result,
             protect_tool_names=protect_tool_names,
             spill=spill_callback,
+            recovery_footer=recovery_footer,
         )
         # Tier 1 is not the only candidate that can win, and the others rewrite
         # the SAME protected results — which Tier 1 also keeps out of its spill

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "apodex-agent-core"
-version = "0.7.0"
+version = "0.8.0"
 description = "Shared, product-neutral runtime primitives for Apodex agents"
 readme = "README.md"
 license = "Apache-2.0"

tests/test_keep_last_n_compactor.py

@@ -18,6 +18,7 @@
     OMITTED_TOOL_RESULT_PLACEHOLDER,
     KeepLastNToolResultsCompactor,
     _args_preview,
+    default_recovery_footer,
 )
 
 
@@ -218,3 +219,85 @@ def test_loop_cap_handle_is_used_when_no_spill_ref_is_pinned():
     )
     tool_out = next(m for m in out if m.get("role") == "tool")
     assert tool_out["spill_refs"] == ["/spill/loop-cap"]
+
+
+# --- the recovery footer hook ----------------------------------------------
+
+
+def test_the_default_footer_names_no_tool():
+    """A card is the whole message, so a wrong tool name here has no antidote.
+
+    This module cannot know what a host calls its recovery tool, or whether that
+    tool is bound for the agent whose history this is. Carrying only the handle
+    is the one rendering that is correct in every host, which is why it is the
+    default rather than a guess at the common case.
+    """
+    body = "see https://example.com/a " + "x" * 2_000
+    content = _blanked(
+        _one_call("web_search", '{"query": "x"}', body),
+        spill=lambda _n, _c: "/spill/xyz",
+    )
+    assert content.splitlines()[-1] == default_recovery_footer("/spill/xyz")
+    assert "recover" not in content.splitlines()[-1].lower()
+
+
+def test_a_host_footer_replaces_the_last_line_and_nothing_else():
+    body = "see https://example.com/a " + "x" * 2_000
+    messages = _one_call("web_search", '{"query": "x"}', body)
+    default = _blanked(messages, spill=lambda _n, _c: "/spill/xyz")
+    hosted = _blanked(
+        messages,
+        spill=lambda _n, _c: "/spill/xyz",
+        recovery_footer=lambda ref: f"[Saved. Fetch it with fetch_body id {ref}.]",
+    )
+
+    assert hosted.splitlines()[-1] == "[Saved. Fetch it with fetch_body id /spill/xyz.]"
+    # The card above the footer — call line and source URLs — is untouched, so a
+    # host swapping the footer cannot silently change the card's token budget.
+    assert hosted.splitlines()[:-1] == default.splitlines()[:-1]
+
+
+def test_a_host_footer_still_reaches_the_model_only_when_a_body_spilled():
+    """No handle, no footer — a host renderer must not invent one.
+
+    The card is emitted for unspilled bodies too (that is the whole point of the
+    args + URLs lines). Calling the footer there would have it render a pointer
+    to nothing.
+    """
+    calls: list[str] = []
+
+    def footer(ref: str) -> str:
+        calls.append(ref)
+        return f"[Saved: {ref}]"
+
+    content = _blanked(
+        _one_call("web_search", '{"query": "x"}', "see https://example.com/a " + "x" * 2_000),
+        recovery_footer=footer,
+    )
+    assert calls == []
+    assert "[Saved:" not in content
+
+
+def test_a_host_footer_survives_a_second_pass_unnested():
+    """Idempotency is anchored on the placeholder, not the footer's wording.
+
+    A host footer is free to be longer than the default, which is exactly the
+    case where a second pass re-carding the message would compound. The
+    already-placeheld check has to catch it regardless of what the last line says.
+    """
+    body = "see https://example.com/a " + "x" * 2_000
+    messages = _one_call("web_search", '{"query": "x"}', body)
+    compactor = KeepLastNToolResultsCompactor(
+        keep_tool_result=0,
+        spill=lambda _n, _c: "/spill/xyz",
+        recovery_footer=lambda ref: (
+            "[Full text saved. Recovery id: " + ref + " — use the recover_result tool "
+            "(a tool call, not a shell command) with that spill id.]"
+        ),
+    )
+    once = compactor.compact(messages, 0)
+    twice = compactor.compact(once, 0)
+    first = next(m["content"] for m in once if m.get("role") == "tool")
+    second = next(m["content"] for m in twice if m.get("role") == "tool")
+    assert first == second
+    assert second.count("Recovery id:") == 1

tests/test_tiered_compact.py

@@ -15,6 +15,7 @@
     OMITTED_TOOL_RESULT_PLACEHOLDER,
     SPILL_MANIFEST_HEADER,
     KeepLastNToolResultsCompactor,
+    default_recovery_footer,
     estimate_tokens,
 )
 from agent_core.runtime.loop.tiered_compact import (
@@ -268,3 +269,56 @@ def test_overlong_handle_removes_an_old_index_without_inserting_an_empty_one():
         for m in twice
     )
     assert not any(SPILL_MANIFEST_HEADER in str(m.get("content")) for m in twice)
+
+
+def test_tiered_forwards_the_recovery_footer_to_its_tier1():
+    """A host that worded the footer must not get two wordings in one run.
+
+    Tier 1 is a winning candidate in its own right, so its card reaches the model
+    from inside ``TieredCompactor`` exactly as it does from the standalone
+    compactor. If this constructor swallowed ``recovery_footer``, a workflow whose
+    primary path is tiered and whose fallback path is standalone would render the
+    handle-only default on every production turn and the host's prose only in the
+    fallback — the harder failure to notice of the two.
+    """
+    async def run():
+        footer = lambda ref: f"[Saved. Ask fetch_body for {ref}.]"  # noqa: E731
+        tiered = TieredCompactor(
+            keep_tool_result=1,
+            summary_llm=_FakeLLM(),
+            relief_target=10**9,
+            spill=lambda _n, _c: "/spill/xyz",
+            recovery_footer=footer,
+        )
+        out = await tiered.compact(_msgs(), 1)
+        carded = [
+            m.get("content") or ""
+            for m in out
+            if (m.get("content") or "").startswith(OMITTED_TOOL_RESULT_PLACEHOLDER)
+        ]
+        assert carded, "expected Tier 1 to card at least one tool body"
+        for content in carded:
+            assert content.splitlines()[-1] == "[Saved. Ask fetch_body for /spill/xyz.]"
+
+    asyncio.run(run())
+
+
+def test_tiered_default_footer_is_unchanged():
+    async def run():
+        tiered = TieredCompactor(
+            keep_tool_result=1,
+            summary_llm=_FakeLLM(),
+            relief_target=10**9,
+            spill=lambda _n, _c: "/spill/xyz",
+        )
+        out = await tiered.compact(_msgs(), 1)
+        carded = [
+            m.get("content") or ""
+            for m in out
+            if (m.get("content") or "").startswith(OMITTED_TOOL_RESULT_PLACEHOLDER)
+        ]
+        assert carded
+        for content in carded:
+            assert content.splitlines()[-1] == default_recovery_footer("/spill/xyz")
+
+    asyncio.run(run())