Fix Responses prompt caching in long conversations

Retain each completed local Responses turn's masked extras in result
metadata and replay them at their original boundary. This keeps growing
tool and user turns from discarding cacheable prompt prefixes.

Preserve replay eligibility, secret masking and history summaries, and
include retained extras in Responses compression accounting. Chat
Completions and fallback requests retain their existing behavior.

Validation: 238 targeted tests passed in the framework runtime. A live
GPT-6.1 Sol conversation reached 131,099 input tokens with 130,176 cached
(99.3%) on a user follow-up.

AGENTS.md

@@ -17,7 +17,7 @@
 - `agent.py` owns `Agent`, `AgentContext`, and loop data.
 - `Agent.hist_add_message` updates `AgentContext.last_message`, the activity timestamp used by snapshots and chat persistence; `created_at` remains unchanged. Context initialization preserves an explicitly supplied activity timestamp across greeting/history hooks.
 - `Agent.hist_add_ai_response` owns Responses-API state advancement: it calls `_remember_llm_result_state` internally. Model turns pass an `LLMResult`; omitted results and legacy positional string IDs use the non-LLM sentinel. Callers must not invoke `_remember_llm_result_state` manually.
-- Prepared local Responses input may project intact native history through `helpers/responses_history.py`; retain original Chat/fallback messages, current protocol/extras, summaries and masking. Durable capability metadata stores only a stable-prefix digest for eligibility.
+- Prepared local Responses input may project intact native history through `helpers/responses_history.py`; retain original Chat/fallback messages, current protocol/extras, summaries and masking. Durable capability metadata stores only a stable-prefix digest for eligibility; per-turn extras snapshots live separately in result metadata and count toward Responses history compression.
 - `Agent.call_chat_model_turn` builds native function tools only when the chat model's `a0_api_mode` selects Responses; Chat Completions turns pass an empty tool list and name map, since that transport drops them.
 - Native function calls use canonical call content for history and combine public response text with calls for repeat comparison. The canonical history content passes through the normal history template/masking hook; provider output metadata remains intact.
 - `initialize.py` owns framework initialization.

helpers/history.py

@@ -180,7 +180,23 @@ class Topic(Record):
         if self.summary:
             return tokens.approximate_tokens(self.summary)
         else:
-            return sum(msg.get_tokens() for msg in self.messages)
+            return sum(self._message_tokens())
+
+    def _message_tokens(self) -> list[int]:
+        from helpers.llm_result import result_from_metadata
+        from helpers.litellm_transport import TransportMode
+
+        extras = []
+        for msg in self.messages:
+            result = result_from_metadata(msg.metadata) if msg.ai and not msg.summary else None
+            extras.append(result.history_extras if result else "")
+        include_extras = any(extras) and TransportMode.from_value(
+            self.history.agent.get_chat_model().kwargs.get("a0_api_mode")
+        ) is TransportMode.RESPONSES
+        return [
+            msg.get_tokens() + (tokens.approximate_tokens(extra) if include_extras and extra else 0)
+            for msg, extra in zip(self.messages, extras)
+        ]
 
     def add_message(
         self,
@@ -224,11 +240,12 @@ class Topic(Record):
             * message_ratio
         )
         large_msgs = []
-        for m in (m for m in self.messages if not m.summary):
+        for m, tok in zip(self.messages, self._message_tokens()):
+            if m.summary:
+                continue
             # TODO refactor this
             out = m.output()
             text = output_text(out)
-            tok = m.get_tokens()
             leng = len(text)
             if tok > msg_max_size:
                 large_msgs.append((m, tok, leng, out))

helpers/history.py.dox.md

@@ -80,6 +80,7 @@
 - Update this file whenever public functions, classes, persistence behavior, path/security assumptions, side effects, or cross-module contracts change.
 - `clear_responses_provider_state(agent)` removes the active provider continuation IDs after local history rewrites while preserving stored response ID lists for later cleanup.
 - `Message.from_dict()` normalizes legacy AI Responses metadata through `LLMResult.metadata()` so loaded chats shed transient payloads while unrelated metadata and non-AI tool-result inputs remain intact.
+- Topic token totals and large-message compression include retained extras snapshots when the selected model uses Responses. Chat Completions keeps text-only accounting. Summarized messages/topics no longer contribute snapshot overhead; unchanged records conservatively count it even when replay eligibility later changes.
 - `output_langchain()` removes leading assistant messages after grouping so provider histories always begin with a user turn; the WebUI greeting remains persisted and displayed but is not sent as an orphaned assistant message.
 - `_json_dumps()` emits compact JSON (`","`, `":"` separators) for serialized history and generated non-string user-turn content.
 - Observed side-effect areas: filesystem writes, filesystem deletion, model calls, plugin state, settings/state persistence, secret handling.

helpers/litellm_transport.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from copy import deepcopy
 from dataclasses import dataclass, field
 from enum import Enum
 import hashlib
@@ -184,6 +185,7 @@ class LiteLLMTransport:
     last_request_state: str = field(init=False, default=RESPONSES_STATE_PROVIDER)
     explicit_prompt_caching: bool = field(init=False, default=False)
     history_prefix_hash: str = field(init=False, default="")
+    history_extras: str = field(init=False, default="")
     stream_usage_retried: bool = field(init=False, default=False)
 
     def __post_init__(self) -> None:
@@ -411,6 +413,7 @@ class LiteLLMTransport:
             self.kwargs.get("responses_state")
         )
         self.history_prefix_hash = ""
+        self.history_extras = ""
         context = self.kwargs.get("responses_history_context")
         if (
             self.last_request_state == RESPONSES_STATE_LOCAL
@@ -424,6 +427,11 @@ class LiteLLMTransport:
                 prefix = project_system_prompt(context.get("prefix", []), replacements)
                 scope = {"affinity": self.policy.cache_key, "tools": response_kwargs.get("tools")}
                 self.history_prefix_hash = responses_history.prefix_hashes(prefix, scope)[-1]
+                self.history_extras = responses_history.prompt_extras(prompt, prefix)
+                if self.history_extras and isinstance(context.get("extras"), str):
+                    self.history_extras = context["extras"]
+                    prompt = deepcopy(prefix)
+                    responses_history.append_extras(prompt, self.history_extras)
                 response_kwargs["input"] = responses_history.project_history(
                     prompt, context.get("groups", []), scope,
                 )
@@ -459,6 +467,7 @@ class LiteLLMTransport:
             capability=self._capability_metadata(),
         )
         result.usage = _reported_usage(response)
+        result.history_extras = self.history_extras
         return result
 
     def _stream_result_from_parser(

helpers/litellm_transport.py.dox.md

@@ -25,6 +25,7 @@
 - Keep provider selection and provider-specific defaults outside this helper; callers pass a resolved LiteLLM model name and kwargs.
 - Strip Agent Zero internal kwargs before sending requests to LiteLLM.
 - Validate internal `responses_history_context` against selected prepared local input, bind its stable-prefix digest to actual affinity/tool schemas, and project eligible groups through `responses_history`. Strip the control on every provider path and omit its digest from Chat/fallback result metadata.
+- Capture only the validated current extras suffix in completed local Responses results. Historical snapshots restore the growing request prefix; Chat/fallback and provider-managed continuation never acquire these snapshots.
 - Apply `responses_prompt_replacements` only to Responses input when generated A0 functions are present; preserve original Chat/fallback messages and strip this internal control before either provider call.
 - Do not send orphan tool controls when no tools are present; strict OpenAI-compatible servers can reject empty `tools` arrays.
 - When Agent Zero function tools are present, default Responses requests to one required native call; explicit request-level `tool_choice` and `parallel_tool_calls` values still win.

helpers/llm_result.py

@@ -63,11 +63,13 @@ class LLMResult:
     usage: dict[str, Any] = field(default_factory=dict)
     raw: dict[str, Any] = field(default_factory=dict)
     capability: dict[str, Any] = field(default_factory=dict)
+    history_extras: str = ""
 
     @classmethod
     def from_dict(cls, data: dict[str, Any] | None) -> "LLMResult":
         data = data or {}
         mode = data.get("mode")
+        extras = data.get("history_extras")
         return cls(
             response=str(data.get("response") or ""),
             reasoning=str(data.get("reasoning") or ""),
@@ -83,6 +85,7 @@ class LLMResult:
             usage=object_to_dict(data.get("usage") or {}),
             raw=object_to_dict(data.get("raw") or {}),
             capability=object_to_dict(data.get("capability") or {}),
+            history_extras=extras if isinstance(extras, str) else "",
         )
 
     @classmethod
@@ -229,6 +232,7 @@ class LLMResult:
             "usage": self.usage,
             "raw": self.raw,
             "capability": self.capability,
+            "history_extras": self.history_extras,
         }
 
     def metadata(self) -> dict[str, Any]:
@@ -242,6 +246,7 @@ class LLMResult:
                 "state": self.state,
                 "usage": self.usage,
                 "capability": self.capability,
+                **({"history_extras": self.history_extras} if self.history_extras else {}),
             }
         }
 

helpers/llm_result.py.dox.md

@@ -17,7 +17,7 @@
 
 ## Runtime Contracts
 
-- `LLMResult.metadata()` stores only durable provider state under `RESPONSE_METADATA_KEY`: response IDs, structured output items, provider/mode/state, usage, and capability data. Runtime prompt inputs, raw responses, and duplicated response/reasoning text are not persisted in history.
+- `LLMResult.metadata()` stores durable provider state under `RESPONSE_METADATA_KEY`: response IDs, structured output items, provider/mode/state, usage, capability data, and a nonempty `history_extras` snapshot for eligible local Responses turns. Accumulated prompt inputs, raw responses, and duplicated response/reasoning text are not persisted in history. Missing or non-string extras deserialize as empty for legacy chats.
 - `from_response(...)` must preserve provider `response_id`, `previous_response_id`, raw output items, usage, and capability metadata.
 - `from_dict(...)` preserves explicit modes, including the empty non-LLM sentinel; only missing or null modes use the legacy `responses` default.
 - `from_chat(...)` must produce an equivalent chat-completions result with `mode="chat_completions"` and `state="off"`, preserving optional function-call output items and provider usage when the chat transport supplies them.

helpers/responses_history.py

@@ -51,13 +51,13 @@ def prepare_call(agent: Any, call_data: dict) -> dict:
         )
         return rendered[0].get("content") if len(rendered) == 1 else None
 
+    mask = get_secrets_manager(agent.context).mask_values
+    prompt = ResponsesTransport.input_from_model_messages(model, messages)
+    prefix = ResponsesTransport.input_from_model_messages(model, prepared["prefix"])
     kwargs["responses_history_context"] = {
-        "prompt": ResponsesTransport.input_from_model_messages(model, messages),
-        "prefix": ResponsesTransport.input_from_model_messages(model, prepared["prefix"]),
-        "groups": prepare_groups(
-            agent.loop_data.history_output, render,
-            get_secrets_manager(agent.context).mask_values,
-        ),
+        "prompt": prompt, "prefix": prefix,
+        "extras": mask(prompt_extras(prompt, prefix)),
+        "groups": prepare_groups(agent.loop_data.history_output, render, mask),
     }
     return kwargs
 
@@ -75,6 +75,38 @@ def prefix_hashes(items: list[dict], scope: dict) -> list[str]:
     return hashes
 
 
+def prompt_extras(items: list[dict], prefix: list[dict]) -> str:
+    """Extract only the prepared extras suffix, never a copy of prior input."""
+    if len(items) == len(prefix) + 1 and items[:-1] == prefix:
+        content = items[-1].get("content")
+        return content if items[-1].get("role") == "user" and isinstance(content, str) else ""
+    if not prefix or len(items) != len(prefix) or items[:-1] != prefix[:-1]:
+        return ""
+    before, after = prefix[-1], items[-1]
+    if before.get("role") != "user" or {**after, "content": before.get("content")} != before:
+        return ""
+    old, new = before.get("content"), after.get("content")
+    if isinstance(old, str) and isinstance(new, str) and new.startswith(old + "\n"):
+        return new[len(old) + 1:]
+    if isinstance(old, list) and isinstance(new, list) and len(new) == len(old) + 1 and new[:-1] == old:
+        block = new[-1]
+        if isinstance(block, dict) and block.get("type") == "input_text" and isinstance(block.get("text"), str):
+            return block["text"]
+    return ""
+
+
+def append_extras(items: list[dict], extras: str) -> None:
+    if items and items[-1].get("role") == "user":
+        content = items[-1].get("content")
+        if isinstance(content, str):
+            items[-1]["content"] = content + "\n" + extras
+            return
+        if isinstance(content, list):
+            content.append({"type": "input_text", "text": extras})
+            return
+    items.append({"role": "user", "content": extras})
+
+
 def _json_object(text: Any) -> dict | None:
     if not isinstance(text, str):
         return None
@@ -128,7 +160,10 @@ def prepare_groups(
         if not all(_replayable_output(item) for item in output):
             continue
         try:
-            if _contains_secret(output, mask) or _contains_secret([call.arguments for call in calls], mask):
+            if (
+                _contains_secret(output, mask) or _contains_secret([call.arguments for call in calls], mask)
+                or _contains_secret(result.history_extras, mask)
+            ):
                 continue
         except Exception:
             continue  # Keep prepared text when secret masking cannot be checked.
@@ -151,10 +186,12 @@ def prepare_groups(
             ):
                 break
             results.append({"type": "function_call_output", "call_id": call_id, "output": text})
-        if len(results) == len(calls):
+        paired = len(results) == len(calls)
+        if paired or result.history_extras:
             groups.append({
                 PREFIX_HASH: prefix_hash, "assistant": assistant,
-                "output": deepcopy(output), "results": results,
+                "extras": result.history_extras,
+                "output": deepcopy(output) if paired else [], "results": results if paired else [],
             })
     return groups
 
@@ -162,14 +199,20 @@ def prepare_groups(
 def project_history(items: list[dict], groups: list[dict], scope: dict) -> list[dict]:
     indices = {value: index for index, value in enumerate(prefix_hashes(items, scope))}
     replacements = {}
+    extras = {}
     used_calls, used_items = set(), set()
     for group in groups:
         index = indices.get(group.get(PREFIX_HASH))
-        if index is None or index + 1 >= len(items):
+        if index is None or index >= len(items):
             continue
-        assistant, following = items[index:index + 2]
+        assistant = items[index]
         if assistant.get("role") != "assistant" or assistant.get("content") != group["assistant"]:
             continue
+        if group.get("extras"):
+            extras[index] = group["extras"]
+        if not group["results"] or index + 1 >= len(items):
+            continue
+        following = items[index + 1]
         content = following.get("content")
         expected = "\n".join(result["output"] for result in group["results"])
         if following.get("role") != "user" or not isinstance(content, str):
@@ -190,6 +233,8 @@ def project_history(items: list[dict], groups: list[dict], scope: dict) -> list[
     result = []
     index = 0
     while index < len(items):
+        if index in extras:
+            append_extras(result, extras[index])
         if index in replacements:
             result.extend(deepcopy(replacements[index]))
             index += 2

helpers/responses_history.py.dox.md

@@ -9,13 +9,14 @@
 
 ## Contracts
 
-- Persist only `history_prefix_hash` in existing result capability metadata, never copies of old prompt inputs or a diagnostic replay counter. Hash system, protocol and prepared history; retain freshly prepared extras in each request.
-- Require canonical visible arguments, original unique call IDs, immediately paired result records, compatible local Responses state and unchanged prefix/scope. Legacy records without digests stay as text.
-- Replay function calls, encrypted reasoning and explicitly tagged assistant commentary with text-only content, preserving original provider items and phase. Reject final/unmarked messages, refusals, incomplete commentary, known unmasked secrets, unsupported items, incomplete pairs and non-text result layouts.
+- Persist only `history_prefix_hash` in capability metadata. `LLMResult.history_extras` separately retains that turn's prepared text suffix, never copies of accumulated input. Restore eligible snapshots before their assistant records so growing requests preserve their original prefix; append freshly prepared extras for the current turn.
+- Require canonical visible arguments, compatible local Responses state and unchanged prefix/scope. Native output replay additionally requires original unique call IDs and immediately paired result records. An unpaired final response may retain its extras snapshot but stays as text; never fabricate a tool acknowledgement. Legacy records without digests stay as text.
+- Replay function calls, encrypted reasoning and explicitly tagged assistant commentary with text-only content, preserving original provider items and phase. Reject final/unmarked messages, refusals, incomplete commentary, known unmasked secrets (including extras snapshots), unsupported items, incomplete pairs and non-text result layouts.
 - This helper is a transport adapter over generic rendered records and metadata. Shared history storage/rendering remains API-agnostic; replay only changes copied provider input.
 - Build tool outputs from visible rendered result content, preserving additional fields. Never restore raw tool output metadata or invent a final response-tool result.
 - Preserve summaries, unmatched/custom layouts, attachments and all remaining prepared input. No execution or policy authorization occurs here.
 - Internal history context is removed before every provider request; Chat and fallback messages remain untouched.
+- Mask the current extras suffix before sending and retaining it, using the same secret manager as replay validation. This keeps safe snapshots eligible and preserves identical request boundaries even when a configured secret matches extras text.
 
 - Skip replay preparation for Chat mode after model-call hooks. Reuse the transport conversion primitive rather than private Agent methods; the actual request retains final fallback and input-validation checks.
 

tests/test_responses_history.py

@@ -66,12 +66,12 @@ def test_native_replay_uses_visible_results_and_retains_current_extras():
     prompt = prompt_for(records)
     replay = transport(prompt, prompt, records)
     request = replay._responses_request(stream=False)
-    assert request["input"][:2] == BASE
+    assert request["input"][:2] == source.messages
     assert request["input"][2:4] == [item.to_dict() for item in result.output_items]
     assert request["input"][4] == {"type": "function_call_output", "call_id": "call_1", "output": render(records[1])}
     assert request["input"][5] == {"role": "user", "content": "Current extras"}
     assert "DO NOT REPLAY RAW OUTPUT" not in json.dumps(request)
-    assert "Old extras" not in json.dumps(request)
+    assert result.history_extras == "Old extras"
     assert "responses_history_context" not in request
     assert records == original
     assert prompt == prompt_for(records)
@@ -152,6 +152,36 @@ def test_prepared_context_lifecycle_and_post_hook_changes(monkeypatch):
     assert history.prepare_call(agent, {"model": model, "messages": prompt}) == {"responses_prompt_replacements": {}}
 
 
+def test_current_and_retained_extras_use_the_same_secret_mask(monkeypatch):
+    from types import SimpleNamespace
+    from langchain_core.messages import SystemMessage, HumanMessage
+    from langchain_core.prompts import ChatPromptTemplate
+    from agent import LoopData
+    from models import LiteLLMChatWrapper
+    from helpers import secrets
+
+    agent = SimpleNamespace(context=SimpleNamespace(), loop_data=LoopData())
+    agent.loop_data.history_output = [{"ai": False, "content": "Question"}]
+    model = LiteLLMChatWrapper(model="test", provider="openai", model_config=None, a0_api_mode="responses")
+    prompt = [SystemMessage("Rules"), HumanMessage("Question\nProvider: private-value")]
+    mask = lambda text: text.replace("private-value", "MASKED_KEY")
+    monkeypatch.setattr(secrets, "get_secrets_manager", lambda _: SimpleNamespace(mask_values=mask))
+    history.remember_prompt(agent.loop_data, ChatPromptTemplate.from_messages(prompt).format(), prompt[0], [])
+    prepared = history.prepare_call(agent, {"model": model, "messages": prompt})
+    original = prepared["responses_history_context"]["prompt"]
+    source = transport(original, BASE)
+    source.kwargs.update(prepared)
+    result = completed_call(source)
+    assert result.history_extras == "Provider: MASKED_KEY"
+    assert result.input_items[-1]["content"] == "Question\nProvider: MASKED_KEY"
+    records = records_for(result)
+    replay_prompt = prompt_for(records)
+    replay = transport(replay_prompt, replay_prompt)
+    replay.kwargs["responses_history_context"]["groups"] = history.prepare_groups(records, render, mask)
+    assert replay._responses_request(stream=False)["input"][:len(result.input_items)] == result.input_items
+    assert source._chat_request(stream=False)["messages"] == original
+
+
 @pytest.mark.parametrize("change", ["summary", "arguments", "missing_output", "wrong_id", "attachments", "commentary", "unencrypted", "legacy", "secret"])
 def test_ineligible_history_stays_in_prepared_text(change):
     result = completed_call(transport(BASE, BASE))
@@ -276,3 +306,103 @@ def test_tool_result_summary_is_replayed_as_visible_summary_only():
     output = next(item for item in request["input"] if item.get("type") == "function_call_output")
     assert output["output"] == "Condensed tool result"
     assert "DO NOT REPLAY RAW OUTPUT" not in json.dumps(request)
+
+
+def test_growing_requests_keep_extras_across_tools_reload_and_user_followups():
+    prefix = deepcopy(BASE)
+    records = []
+    previous = []
+    for turn in range(4):
+        prompt = deepcopy(prefix)
+        prompt[-1]["content"] += f"\nExtras {turn}"
+        source = transport(prompt, prefix, records)
+        result = completed_call(source, call_id=f"call_{turn}", name="response" if turn == 2 else "lookup")
+        assert result.input_items[:len(previous)] == previous
+        for old in range(turn + 1):
+            assert json.dumps(result.input_items).count(f"Extras {old}") == 1
+        previous = deepcopy(result.input_items)
+        persisted = json.loads(json.dumps(result.metadata()))["responses"]
+        assert "input_items" not in persisted and persisted["history_extras"] == f"Extras {turn}"
+        result = LLMResult.from_dict(persisted)
+        added = records_for(result)
+        if turn == 2:
+            added[1] = {"ai": False, "content": "Here is a real user follow-up."}
+        records.extend(added)
+        prefix.extend([
+            {"role": "assistant", "content": render(added[0])},
+            {"role": "user", "content": render(added[1])},
+        ])
+    # A terminating response call has no result; never invent an acknowledgement.
+    assert not any(item.get("call_id") == "call_2" for item in previous)
+
+
+@pytest.mark.parametrize("tail", [
+    {"role": "user", "content": "Question"},
+    {"role": "user", "content": [{"type": "input_text", "text": "Question"},
+                                   {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}]},
+    {"role": "assistant", "content": "Prior response"},
+])
+def test_extras_snapshot_preserves_prepared_message_boundaries(tail):
+    prefix = [BASE[0], deepcopy(tail)]
+    prompt = deepcopy(prefix)
+    if tail["role"] == "assistant":
+        prompt.append({"role": "user", "content": "Old extras"})
+    elif isinstance(tail["content"], str):
+        prompt[-1]["content"] += "\nOld extras"
+    else:
+        prompt[-1]["content"].append({"type": "input_text", "text": "Old extras"})
+    result = completed_call(transport(prompt, prefix))
+    assert result.history_extras == "Old extras"
+    records = records_for(result)
+    replay_prompt = prompt_for(records, prefix)
+    request = transport(replay_prompt, replay_prompt, records)._responses_request(stream=False)
+    assert request["input"][:len(prompt)] == prompt
+    assert prefix[-1] == tail
+
+
+@pytest.mark.parametrize("change", ["summary", "arguments", "secret", "mask_error", "tools"])
+def test_extras_replay_respects_history_and_masking_changes(change):
+    prompt = [BASE[0], {"role": "user", "content": "Question\nOld extras"}]
+    records = records_for(completed_call(transport(prompt, BASE)))
+    prefix = deepcopy(BASE)
+    mask = lambda text: text
+    kwargs = {}
+    if change == "summary":
+        prefix[-1]["content"] = "Summarized question"
+    elif change == "arguments":
+        records[0]["content"] = "Summarized answer"
+    elif change == "secret":
+        mask = lambda text: text.replace("Old extras", "MASKED")
+    elif change == "mask_error":
+        def mask(text):
+            raise OSError("masking unavailable")
+    else:
+        kwargs["a0_responses_function_tools"] = [{**TOOLS[0], "description": "changed"}]
+    prompt = prompt_for(records, prefix)
+    source = transport(prompt, prompt, **kwargs)
+    source.kwargs["responses_history_context"]["groups"] = history.prepare_groups(records, render, mask)
+    assert source._responses_request(stream=False)["input"] == prompt
+
+
+def test_retained_extras_count_toward_responses_compression_only(monkeypatch):
+    from types import SimpleNamespace
+    from helpers import history as stored_history, tokens
+
+    model = SimpleNamespace(kwargs={"a0_api_mode": "responses"})
+    agent = SimpleNamespace(get_chat_model=lambda: model)
+    hist = stored_history.History(agent)
+    result = completed_call(transport([BASE[0], {"role": "user", "content": "Question\n" + "large extras " * 100}], BASE))
+    msg = hist.add_message(True, result.function_calls_text(), metadata=result.metadata())
+    expected = msg.get_tokens() + tokens.approximate_tokens(result.history_extras)
+    assert hist.get_tokens() == expected
+    restored = stored_history.deserialize_history(hist.serialize(), agent)
+    assert restored.get_tokens() == expected
+    monkeypatch.setattr(hist, "_get_ctx_size_for_history", lambda: expected - 1)
+    assert hist.is_over_limit()
+    model.kwargs["a0_api_mode"] = "chat"
+    assert hist.get_tokens() == msg.get_tokens() and not hist.is_over_limit()
+    model.kwargs["a0_api_mode"] = "responses"
+    msg.set_summary("Condensed answer")
+    assert hist.get_tokens() == msg.get_tokens()
+    restored.current.summary = "Condensed topic"
+    assert restored.get_tokens() == tokens.approximate_tokens("Condensed topic")