From dbee30a04895cb0488cfa150d6e4c8bc35ffc589 Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Sat, 5 Sep 2026 10:39:58 +0100
Subject: [PATCH] fix(compression,llm): keep the summary across mid-execution
 compression, ignore empty saved points, commit the head hash on success

Three review findings on the bounded-chain change.

Mid-execution compression rebuilt the conversation from the in-flight
messages, which after a turn-start reuse hold only the recent turns: the
summary living in the system prompt never reached the compressor, so the
new summary replaced the old one, and the persisted point's query_index
was relative to that shortened list. The summary the agent is running
under now rides into the synthetic conversation as its latest point
(query_index -1, so every in-flight query is new), for both the database
and the in-memory path, and the database path persists the index of the
saved conversation's last row.

Saved points with an empty summary, which earlier versions wrote, were
treated as reusable: get_compressed_context sliced the raw history away
and the effective token count made the conversation look small. Point
selection everywhere now takes the latest usable point (non-blank
summary, positive token count) and falls back to the raw history when
there is none.

The chained system-head hash was committed while building the request,
so a transport failure followed by the same-primary retry omitted a
changed system message. The hash is now staged per request and committed
only when the provider records the response.
---
 .../services/compression/orchestrator.py      |  47 +++--
 .../answer/services/compression/service.py    |  39 ++++-
 .../services/compression/token_counter.py     |  11 +-
 .../api/answer/services/compression/types.py  |  30 ++++
 application/llm/handlers/base.py              |  68 +++++++-
 application/llm/openai.py                     |  26 ++-
 .../compression/test_compression_reuse.py     | 161 ++++++++++++++++++
 .../services/compression/test_orchestrator.py |   2 +-
 .../llm/handlers/test_mid_execution_chain.py  | 109 ++++++++++++
 tests/llm/test_responses_chain_budget.py      |  42 +++++
 10 files changed, 500 insertions(+), 35 deletions(-)

diff --git a/application/api/answer/services/compression/orchestrator.py b/application/api/answer/services/compression/orchestrator.py
index ad117559..8429af7d 100644
--- a/application/api/answer/services/compression/orchestrator.py
+++ b/application/api/answer/services/compression/orchestrator.py
@@ -10,6 +10,7 @@ from application.api.answer.services.compression.threshold_checker import (
 from application.api.answer.services.compression.types import (
     CompressionResult,
     is_compression_summary_row,
+    latest_usable_compression_point,
 )
 from application.core.model_utils import (
     get_api_key_for_provider,
@@ -96,10 +97,13 @@ class CompressionOrchestrator:
             ):
                 compression_meta = conversation.get("compression_metadata") or {}
                 points = compression_meta.get("compression_points") or []
-                if compression_meta.get("is_compressed") and points:
+                usable_point = latest_usable_compression_point(points)
+                if compression_meta.get("is_compressed") and usable_point:
                     # A saved summary applies to every later turn, not only
                     # the turn that made it (measured in prod: the full raw
-                    # history was replayed on the very next turn).
+                    # history was replayed on the very next turn). A point
+                    # whose summary is empty is not one: the raw history
+                    # stays.
                     summary, recent = self._read_only_service(
                         model_id
                     ).get_compressed_context(conversation)
@@ -108,7 +112,7 @@ class CompressionOrchestrator:
                         recent,
                         last_compression_at=(
                             compression_meta.get("last_compression_at")
-                            or points[-1].get("timestamp")
+                            or usable_point.get("timestamp")
                         ),
                     )
                 # No compression needed, return full history
@@ -145,6 +149,7 @@ class CompressionOrchestrator:
         decoded_token: Dict[str, Any],
         user_id: Optional[str] = None,
         model_user_id: Optional[str] = None,
+        persist_query_index: Optional[int] = None,
     ) -> CompressionResult:
         """
         Perform the actual compression operation.
@@ -211,18 +216,22 @@ class CompressionOrchestrator:
             points = (conversation.get("compression_metadata") or {}).get(
                 "compression_points"
             ) or []
+            # Build on the latest point that can stand in for the history it
+            # covers; an empty saved summary does not move the start.
+            usable_point = latest_usable_compression_point(points)
             start_index = 0
-            if points:
+            if usable_point:
                 try:
-                    start_index = int(points[-1].get("query_index", -1)) + 1
+                    start_index = int(usable_point.get("query_index", -1)) + 1
                 except (TypeError, ValueError):
                     start_index = 0
+            start_index = max(start_index, 0)
             new_queries = [
                 q
                 for q in conversation.get("queries", [])[start_index:]
                 if not is_compression_summary_row(q)
             ]
-            if start_index > compress_up_to or not new_queries:
+            if usable_point and (start_index > compress_up_to or not new_queries):
                 logger.info(
                     f"No new queries since the last compression point for "
                     f"conversation {conversation_id}; reusing its summary"
@@ -231,23 +240,31 @@ class CompressionOrchestrator:
                     conversation
                 )
                 return CompressionResult.success_from_existing(
-                    summary, recent, last_compression_at=points[-1].get("timestamp")
+                    summary, recent, last_compression_at=usable_point.get("timestamp")
                 )
 
+            if usable_point:
+                covered = usable_point.get("query_index")
+                on_top = (
+                    " on top of the summary carried into this turn"
+                    if isinstance(covered, int) and covered < 0
+                    else f" on top of the summary at query {covered}"
+                )
+            else:
+                on_top = ""
             logger.info(
                 f"Initiating compression for conversation {conversation_id}: "
                 f"compressing queries {start_index}-{compress_up_to} of "
-                f"{queries_count}"
-                + (
-                    f" on top of the summary at query {points[-1].get('query_index')}"
-                    if points
-                    else ""
-                )
+                f"{queries_count}{on_top}"
             )
 
             # Perform compression and save to DB
             metadata = compression_service.compress_and_save(
-                conversation_id, conversation, compress_up_to, start_index=start_index
+                conversation_id,
+                conversation,
+                compress_up_to,
+                start_index=start_index,
+                persist_query_index=persist_query_index,
             )
 
             logger.info(
@@ -284,6 +301,7 @@ class CompressionOrchestrator:
         decoded_token: Dict[str, Any],
         current_conversation: Optional[Dict[str, Any]] = None,
         model_user_id: Optional[str] = None,
+        persist_query_index: Optional[int] = None,
     ) -> CompressionResult:
         """
         Perform compression during tool execution.
@@ -325,6 +343,7 @@ class CompressionOrchestrator:
                 decoded_token,
                 user_id=user_id,
                 model_user_id=model_user_id,
+                persist_query_index=persist_query_index,
             )
 
         except Exception as e:
diff --git a/application/api/answer/services/compression/service.py b/application/api/answer/services/compression/service.py
index ef1f494d..7419cbcb 100644
--- a/application/api/answer/services/compression/service.py
+++ b/application/api/answer/services/compression/service.py
@@ -12,6 +12,7 @@ from application.api.answer.services.compression.token_counter import TokenCount
 from application.api.answer.services.compression.types import (
     CompressionMetadata,
     is_compression_summary_row,
+    latest_usable_compression_point,
 )
 from application.core.settings import settings
 
@@ -53,6 +54,7 @@ class CompressionService:
         conversation: Dict[str, Any],
         compress_up_to_index: int,
         start_index: int = 0,
+        persist_query_index: Optional[int] = None,
     ) -> CompressionMetadata:
         """
         Compress conversation history up to specified index.
@@ -60,6 +62,12 @@ class CompressionService:
         Args:
             conversation: Full conversation document
             compress_up_to_index: Last query index to include in compression
+            start_index: First query index that is new since the previous
+                compression point (earlier queries are already summarised).
+            persist_query_index: Index to record on the point when the
+                conversation passed in is a shortened, synthetic one (the
+                mid-execution path) and its positions do not match the
+                saved conversation's. Defaults to ``compress_up_to_index``.
 
         Returns:
             CompressionMetadata with compression details
@@ -104,11 +112,12 @@ class CompressionService:
             )
 
             previous_summary_tokens = 0
+            usable_point = latest_usable_compression_point(existing_compressions)
+            existing_compressions = [usable_point] if usable_point else []
             if existing_compressions:
                 # Each point already folds in the ones before it, so only
-                # the latest matters — and it is part of what the new
-                # summary replaces.
-                existing_compressions = [existing_compressions[-1]]
+                # the latest usable one matters — and it is part of what
+                # the new summary replaces.
                 previous_summary_tokens = TokenCounter.count_message_tokens(
                     [{"content": existing_compressions[0].get("compressed_summary", "")}]
                 )
@@ -188,7 +197,11 @@ class CompressionService:
             # Build compression metadata
             compression_metadata = CompressionMetadata(
                 timestamp=datetime.now(timezone.utc),
-                query_index=compress_up_to_index,
+                query_index=(
+                    persist_query_index
+                    if persist_query_index is not None
+                    else compress_up_to_index
+                ),
                 compressed_summary=compressed_summary,
                 original_token_count=original_tokens,
                 compressed_token_count=compressed_tokens,
@@ -209,6 +222,7 @@ class CompressionService:
         conversation: Dict[str, Any],
         compress_up_to_index: int,
         start_index: int = 0,
+        persist_query_index: Optional[int] = None,
     ) -> CompressionMetadata:
         """
         Compress conversation and save to database.
@@ -231,7 +245,10 @@ class CompressionService:
 
         # Perform compression
         metadata = self.compress_conversation(
-            conversation, compress_up_to_index, start_index=start_index
+            conversation,
+            compress_up_to_index,
+            start_index=start_index,
+            persist_query_index=persist_query_index,
         )
 
         # Save to database
@@ -278,8 +295,16 @@ class CompressionService:
                     return None, []
                 return None, queries
 
-            # Get the most recent compression point
-            latest_compression = compression_points[-1]
+            # The most recent point that can actually stand in for the
+            # history it covers. An empty saved summary must not slice the
+            # raw history away and replace it with nothing.
+            latest_compression = latest_usable_compression_point(compression_points)
+            if latest_compression is None:
+                logger.warning(
+                    "No usable compression point (saved summaries are empty) - "
+                    "using full history"
+                )
+                return None, conversation.get("queries", []) or []
             compressed_summary = latest_compression.get("compressed_summary")
             last_compressed_index = latest_compression.get("query_index")
             compressed_tokens = latest_compression.get("compressed_token_count", 0)
diff --git a/application/api/answer/services/compression/token_counter.py b/application/api/answer/services/compression/token_counter.py
index 462abb72..ad082879 100644
--- a/application/api/answer/services/compression/token_counter.py
+++ b/application/api/answer/services/compression/token_counter.py
@@ -5,7 +5,10 @@ from typing import Any, Dict, List
 
 from application.utils import num_tokens_from_string
 from application.core.settings import settings
-from application.api.answer.services.compression.types import is_compression_summary_row
+from application.api.answer.services.compression.types import (
+    is_compression_summary_row,
+    latest_usable_compression_point,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -148,7 +151,11 @@ class TokenCounter:
             points = metadata.get("compression_points") or []
             if not (metadata.get("is_compressed") and points):
                 return TokenCounter.count_query_tokens(queries)
-            latest = points[-1] or {}
+            latest = latest_usable_compression_point(points)
+            if latest is None:
+                # Only unusable (empty) points: the raw history is what the
+                # next turn will replay.
+                return TokenCounter.count_query_tokens(queries)
             try:
                 last_index = int(latest.get("query_index", -1))
             except (TypeError, ValueError):
diff --git a/application/api/answer/services/compression/types.py b/application/api/answer/services/compression/types.py
index b7c20ba8..186b6cbb 100644
--- a/application/api/answer/services/compression/types.py
+++ b/application/api/answer/services/compression/types.py
@@ -12,6 +12,36 @@ COMPRESSION_SUMMARY_PROMPT = "[Context Compression Summary]"
 COMPRESSION_SUMMARY_MARKER = "compression_summary"
 
 
+def is_usable_compression_point(point: Any) -> bool:
+    """True when a saved compression point can stand in for the history it
+    covers: a non-blank summary and, when recorded, a positive token count.
+
+    Earlier versions persisted points with an empty summary (a 494k-token
+    conversation was "compressed" to 0 tokens in production); reusing such a
+    point would drop the history it covers and replace it with nothing.
+    """
+    if not isinstance(point, dict):
+        return False
+    summary = point.get("compressed_summary")
+    if not isinstance(summary, str) or not summary.strip():
+        return False
+    count = point.get("compressed_token_count")
+    if count is None:
+        return True
+    try:
+        return int(count) > 0
+    except (TypeError, ValueError):
+        return False
+
+
+def latest_usable_compression_point(points: Any) -> Optional[Dict[str, Any]]:
+    """The most recent point that ``is_usable_compression_point``, or None."""
+    for point in reversed(list(points or [])):
+        if is_usable_compression_point(point):
+            return point
+    return None
+
+
 def is_compression_summary_row(query: Any) -> bool:
     """True for the visible summary row written by ``append_compression_message``.
 
diff --git a/application/llm/handlers/base.py b/application/llm/handlers/base.py
index 6de612d0..661e39bf 100644
--- a/application/llm/handlers/base.py
+++ b/application/llm/handlers/base.py
@@ -592,6 +592,42 @@ class LLMHandler(ABC):
             include_tool_calls=include_tool_calls,
         )
 
+    @staticmethod
+    def _carry_current_summary(conversation: Dict, agent) -> Dict:
+        """Attach the summary the agent is already running under as the
+        synthetic conversation's latest compression point.
+
+        The synthetic conversation is rebuilt from the in-flight messages,
+        which hold only the turns replayed this turn plus the current one;
+        the earlier history exists solely as the summary in the system
+        prompt. Without this point the compressor would summarise the recent
+        turns alone and the new summary would replace, not extend, the old
+        one. ``query_index`` is -1 because every synthetic query is newer
+        than the summary.
+        """
+        summary = getattr(agent, "compressed_summary", None)
+        if not conversation or not summary or not str(summary).strip():
+            return conversation
+        from application.api.answer.services.compression.token_counter import (
+            TokenCounter,
+        )
+
+        tokens = TokenCounter.count_message_tokens([{"content": summary}])
+        conversation["compression_metadata"] = {
+            "is_compressed": True,
+            "compression_points": [
+                {
+                    "timestamp": getattr(agent, "last_compression_at", None),
+                    "query_index": -1,
+                    "compressed_summary": summary,
+                    "compressed_token_count": tokens,
+                    "original_token_count": tokens,
+                    "carried": True,
+                }
+            ],
+        }
+        return conversation
+
     @staticmethod
     def _reset_responses_chain(agent, metadata) -> None:
         """After a compression the rebuilt local messages ARE the context.
@@ -639,11 +675,22 @@ class LLMHandler(ABC):
                 agent.conversation_id, agent.initial_user_id
             )
 
+            persist_query_index = None
             if conversation:
-                # Merge current in-flight messages (including tool calls)
+                # Merge current in-flight messages (including tool calls).
+                # The synthetic conversation holds only the turns replayed
+                # this turn plus the in-flight one, so the summary already in
+                # the system prompt rides along as its latest point, and the
+                # persisted point is indexed against the SAVED conversation,
+                # not this shortened list.
+                saved_count = len(conversation.get("queries") or [])
+                if saved_count:
+                    persist_query_index = saved_count - 1
                 conversation_from_msgs = self._build_conversation_from_messages(messages)
                 if conversation_from_msgs:
-                    conversation = conversation_from_msgs
+                    conversation = self._carry_current_summary(
+                        conversation_from_msgs, agent
+                    )
             else:
                 logger.warning(
                     "Could not load conversation for compression; attempting in-memory compression"
@@ -661,6 +708,7 @@ class LLMHandler(ABC):
                 model_id=agent.model_id,
                 decoded_token=getattr(agent, "decoded_token", {}),
                 current_conversation=conversation,
+                persist_query_index=persist_query_index,
             )
 
             if not result.success:
@@ -767,6 +815,8 @@ class LLMHandler(ABC):
                     "Cannot perform in-memory compression: no user/assistant turns found"
                 )
                 return False, None
+            # The summary already in play must survive this compression too.
+            conversation = self._carry_current_summary(conversation, agent)
 
             compression_model = (
                 settings.COMPRESSION_MODEL_OVERRIDE
@@ -843,14 +893,18 @@ class LLMHandler(ABC):
                     return True, pruned
                 return False, None
 
-            # Attach metadata to synthetic conversation
-            conversation["compression_metadata"] = {
-                "is_compressed": True,
-                "compression_points": [metadata.to_dict()],
+            # Attach metadata to a copy of the synthetic conversation (the
+            # one handed to the compressor keeps its carried point).
+            context_conversation = {
+                **conversation,
+                "compression_metadata": {
+                    "is_compressed": True,
+                    "compression_points": [metadata.to_dict()],
+                },
             }
 
             compressed_summary, recent_queries = (
-                compression_service.get_compressed_context(conversation)
+                compression_service.get_compressed_context(context_conversation)
             )
 
             agent.compressed_summary = compressed_summary
diff --git a/application/llm/openai.py b/application/llm/openai.py
index a983ba6e..d0aa7ab9 100644
--- a/application/llm/openai.py
+++ b/application/llm/openai.py
@@ -264,6 +264,11 @@ class OpenAILLM(BaseLLM):
         # transcript instead of deduping it, so a chained round only carries
         # the head when it changed.
         self._chain_system_hash = None
+        # The head hash a request in flight would establish; it becomes
+        # ``_chain_system_hash`` only once the provider records the response,
+        # so a failed request leaves the chain's head unchanged and a retry
+        # re-sends the head.
+        self._pending_system_hash = None
         # Opaque per-user prompt-cache routing key, set by the agent per call.
         self._prompt_cache_key = None
         # Files-API ids for inline ``file_data`` content parts already
@@ -327,6 +332,7 @@ class OpenAILLM(BaseLLM):
         self._last_reasoning_items = list(state.get("reasoning_items") or [])
         self._reasoning_for_calls = dict(state.get("reasoning_for_calls") or {})
         self._chain_system_hash = state.get("system_hash")
+        self._pending_system_hash = None
         return True
 
     def start_responses_turn(self) -> None:
@@ -337,6 +343,7 @@ class OpenAILLM(BaseLLM):
         self._last_response_call_ids = set()
         self._imported_response_id = None
         self._chain_system_hash = None
+        self._pending_system_hash = None
         self._last_finish_reason = None
 
     def _resolve_file_part(self, item):
@@ -1142,7 +1149,9 @@ class OpenAILLM(BaseLLM):
         """
         chained = bool(previous_response_id and settings.OPENAI_RESPONSES_STORE)
         if not chained:
-            self._chain_system_hash = self._system_fingerprint(messages)
+            # The full head goes out; it becomes the chain's head only once
+            # the provider records the response.
+            self._pending_system_hash = self._system_fingerprint(messages)
             return self._to_responses_input(messages, chained=False), None
 
         trimmed = self._trim_for_previous_response(messages)
@@ -1154,8 +1163,12 @@ class OpenAILLM(BaseLLM):
             # unchanged head stays home. After a compression the head also
             # carries the summary, which made every tool round re-append it.
             trimmed = [m for m in trimmed if m.get("role") != "system"]
-        elif head_hash:
-            self._chain_system_hash = head_hash
+            self._pending_system_hash = self._chain_system_hash
+        else:
+            # A changed (or first) head is sent; only a recorded response
+            # makes it the chain's head, so a failed request and its retry
+            # still carry it.
+            self._pending_system_hash = head_hash or self._chain_system_hash
         input_items = self._to_responses_input(trimmed, chained=True)
         expected = set(self._last_response_call_ids or ())
         if expected:
@@ -1189,7 +1202,7 @@ class OpenAILLM(BaseLLM):
                     "request the provider would reject",
                     sorted(missing),
                 )
-                self._chain_system_hash = self._system_fingerprint(messages)
+                self._pending_system_hash = self._system_fingerprint(messages)
                 return self._to_responses_input(messages, chained=False), None
         return input_items, previous_response_id
 
@@ -1409,6 +1422,11 @@ class OpenAILLM(BaseLLM):
         rid = getattr(response, "id", None)
         if rid:
             self._last_response_id = rid
+        # The provider recorded this request, so the head it carried (or
+        # kept) is now the chain's head.
+        if self._pending_system_hash is not None:
+            self._chain_system_hash = self._pending_system_hash
+            self._pending_system_hash = None
         self._last_response_call_ids = self._function_call_ids(response)
         usage = getattr(response, "usage", None)
         if usage is not None:
diff --git a/tests/api/answer/services/compression/test_compression_reuse.py b/tests/api/answer/services/compression/test_compression_reuse.py
index b3124a81..fa11dab3 100644
--- a/tests/api/answer/services/compression/test_compression_reuse.py
+++ b/tests/api/answer/services/compression/test_compression_reuse.py
@@ -338,3 +338,164 @@ class TestIncrementalTailExcludesSummaryRows:
         assert result.success and not result.compression_performed
         assert result.compressed_summary == "S"
         svc.compress_and_save.assert_not_called()
+
+
+@pytest.mark.unit
+class TestAbsolutePersistIndex:
+    def test_compress_conversation_persists_the_given_absolute_index(self):
+        llm = MagicMock()
+        llm.gen.return_value = "<summary>new</summary>"
+        svc = CompressionService(llm=llm, model_id="m")
+        svc.prompt_builder = MagicMock(version="v1.0")
+        svc.prompt_builder.build_prompt.return_value = [{"role": "user", "content": "p"}]
+        conv = {"queries": [{"prompt": "q18", "response": "r"}, {"prompt": "q19", "response": ""}]}
+
+        metadata = svc.compress_conversation(conv, compress_up_to_index=1, persist_query_index=19)
+
+        assert metadata.query_index == 19
+
+    @patch(
+        "application.api.answer.services.compression.orchestrator.get_provider_from_model_id",
+        return_value="openai",
+    )
+    @patch(
+        "application.api.answer.services.compression.orchestrator.get_api_key_for_provider",
+        return_value="sk",
+    )
+    @patch("application.api.answer.services.compression.orchestrator.LLMCreator")
+    @patch("application.api.answer.services.compression.orchestrator.CompressionService")
+    @patch("application.api.answer.services.compression.orchestrator.settings")
+    def test_mid_execution_builds_on_the_carried_summary_and_persists_the_absolute_index(
+        self, mock_settings, MockCompressionService, MockLLMCreator, _key, _provider,
+        orchestrator, conversation_service,
+    ):
+        mock_settings.COMPRESSION_MODEL_OVERRIDE = None
+        MockLLMCreator.create_llm.return_value = MagicMock()
+        metadata = MagicMock()
+        metadata.compression_ratio = 3.0
+        metadata.original_token_count = 30
+        metadata.compressed_token_count = 10
+        metadata.timestamp = "2026-09-03T10:00:00+00:00"
+        svc = MagicMock()
+        svc.compress_and_save.return_value = metadata
+        svc.get_compressed_context.return_value = ("new", [])
+        MockCompressionService.return_value = svc
+        conversation_service.get_conversation.return_value = {"queries": [], "compression_metadata": {}}
+        synthetic = {
+            "queries": [{"prompt": "q18", "response": "r"}, {"prompt": "q19", "response": ""}],
+            "compression_metadata": {
+                "is_compressed": True,
+                "compression_points": [{"query_index": -1, "compressed_summary": "prior",
+                                        "compressed_token_count": 5, "original_token_count": 5}],
+            },
+        }
+
+        result = orchestrator.compress_mid_execution(
+            "conv1", "user1", "m", {"sub": "user1"}, current_conversation=synthetic,
+            persist_query_index=19,
+        )
+
+        assert result.success and result.compression_performed
+        args, kwargs = svc.compress_and_save.call_args
+        assert kwargs["start_index"] == 0          # every synthetic query is newer than the summary
+        assert kwargs["persist_query_index"] == 19  # indexed against the database conversation
+
+
+@pytest.mark.unit
+class TestUnusableSavedPoints:
+    """A saved point with an empty summary (older versions wrote them) must
+    never make a turn drop history."""
+
+    def _conversation_with_empty_point(self):
+        return {
+            "queries": [
+                {"prompt": "q0", "response": "first saved turn"},
+                {"prompt": "q1", "response": "second saved turn"},
+            ],
+            "compression_metadata": {
+                "is_compressed": True,
+                "last_compression_at": EPOCH,
+                "compression_points": [{
+                    "timestamp": EPOCH, "query_index": 1, "compressed_summary": "",
+                    "original_token_count": 493541, "compressed_token_count": 0,
+                }],
+            },
+            "agent_id": "agent-1",
+        }
+
+    def test_reuse_ignores_an_empty_point_and_keeps_history(
+        self, orchestrator, conversation_service, threshold_checker
+    ):
+        conversation_service.get_conversation.return_value = self._conversation_with_empty_point()
+        threshold_checker.should_compress.return_value = False
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.success is True
+        assert result.compressed_summary is None
+        assert [q["prompt"] for q in result.recent_queries] == ["q0", "q1"]
+        assert [h["prompt"] for h in result.as_history()] == ["q0", "q1"]
+
+    def test_effective_count_ignores_an_empty_point(self):
+        conv = self._conversation_with_empty_point()
+        assert TokenCounter.count_effective_conversation_tokens(conv) == (
+            TokenCounter.count_conversation_tokens(conv)
+        )
+
+    def test_get_compressed_context_ignores_an_empty_point(self):
+        summary, recent = CompressionService(llm=None, model_id="m").get_compressed_context(
+            self._conversation_with_empty_point()
+        )
+        assert summary is None
+        assert [q["prompt"] for q in recent] == ["q0", "q1"]
+
+    def test_get_compressed_context_falls_back_to_the_latest_usable_point(self):
+        conv = _compressed_conversation()
+        conv["queries"].append({"prompt": "q4", "response": "r4"})
+        conv["compression_metadata"]["compression_points"].append(
+            {"timestamp": "2026-09-04T09:00:00+00:00", "query_index": 3,
+             "compressed_summary": "   ", "compressed_token_count": 0}
+        )
+        summary, recent = CompressionService(llm=None, model_id="m").get_compressed_context(conv)
+        assert summary == "S"
+        assert [q["prompt"] for q in recent] == ["q2", "q3", "q4"]
+
+    @patch(
+        "application.api.answer.services.compression.orchestrator.get_provider_from_model_id",
+        return_value="openai",
+    )
+    @patch(
+        "application.api.answer.services.compression.orchestrator.get_api_key_for_provider",
+        return_value="sk",
+    )
+    @patch("application.api.answer.services.compression.orchestrator.LLMCreator")
+    @patch("application.api.answer.services.compression.orchestrator.CompressionService")
+    @patch("application.api.answer.services.compression.orchestrator.settings")
+    def test_recompression_starts_after_the_latest_usable_point(
+        self, mock_settings, MockCompressionService, MockLLMCreator, _key, _provider,
+        orchestrator, conversation_service, threshold_checker,
+    ):
+        mock_settings.COMPRESSION_MODEL_OVERRIDE = None
+        conv = _compressed_conversation()
+        conv["queries"].append({"prompt": "q4", "response": "r4"})
+        conv["compression_metadata"]["compression_points"].append(
+            {"timestamp": "2026-09-04T09:00:00+00:00", "query_index": 3,
+             "compressed_summary": "", "compressed_token_count": 0}
+        )
+        conversation_service.get_conversation.return_value = conv
+        threshold_checker.should_compress.return_value = True
+        MockLLMCreator.create_llm.return_value = MagicMock()
+        metadata = MagicMock()
+        metadata.compression_ratio = 3.0
+        metadata.original_token_count = 30
+        metadata.compressed_token_count = 10
+        metadata.timestamp = "2026-09-05T10:00:00+00:00"
+        svc = MagicMock()
+        svc.compress_and_save.return_value = metadata
+        svc.get_compressed_context.return_value = ("S2", [])
+        MockCompressionService.return_value = svc
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.compression_performed
+        assert svc.compress_and_save.call_args.kwargs["start_index"] == 2
diff --git a/tests/api/answer/services/compression/test_orchestrator.py b/tests/api/answer/services/compression/test_orchestrator.py
index 5f9a154c..bbfc3a60 100644
--- a/tests/api/answer/services/compression/test_orchestrator.py
+++ b/tests/api/answer/services/compression/test_orchestrator.py
@@ -373,7 +373,7 @@ class TestCompressMidExecution:
                 "gpt-4",
                 decoded_token,
                 user_id="user1",
-                model_user_id=None,
+                model_user_id=None, persist_query_index=None,
             )
 
     def test_loads_conversation_when_not_provided(
diff --git a/tests/llm/handlers/test_mid_execution_chain.py b/tests/llm/handlers/test_mid_execution_chain.py
index eedfa796..f7c9c0a9 100644
--- a/tests/llm/handlers/test_mid_execution_chain.py
+++ b/tests/llm/handlers/test_mid_execution_chain.py
@@ -107,3 +107,112 @@ def test_empty_summary_is_a_failure_not_a_success():
     assert success is False and messages is None
     conv_service.append_compression_message.assert_not_called()
     agent.llm.start_responses_turn.assert_not_called()
+
+
+# ── the summary already in play must survive a mid-execution compression ────
+
+
+PRIOR = "Original delivery location: Warehouse Seven."
+
+
+def _run_with(handler, agent, result, db_conversation, synthetic):
+    conv_service = Mock()
+    conv_service.get_conversation.return_value = db_conversation
+    orchestrator = Mock()
+    orchestrator.compress_mid_execution.return_value = result
+    with patch(
+        "application.api.answer.services.compression.CompressionOrchestrator",
+        return_value=orchestrator,
+    ), patch(
+        "application.api.answer.services.conversation_service.ConversationService",
+        return_value=conv_service,
+    ), patch.object(
+        handler, "_build_conversation_from_messages", return_value=synthetic
+    ), patch.object(
+        handler,
+        "_rebuild_messages_after_compression",
+        return_value=[{"role": "system", "content": "rebuilt"}],
+    ), patch.object(handler, "_prune_messages_minimal", return_value=None):
+        outcome = handler._perform_mid_execution_compression(
+            agent, [{"role": "user", "content": "hi"}]
+        )
+    return outcome, orchestrator
+
+
+@pytest.mark.unit
+def test_db_path_carries_the_current_summary_and_persists_the_absolute_index():
+    agent = _agent()
+    agent.compressed_summary = PRIOR
+    agent.last_compression_at = "2026-09-03T09:00:00+00:00"
+    db_conversation = {
+        "queries": [{"prompt": f"q{i}", "response": "r"} for i in range(20)],
+        "compression_metadata": {
+            "is_compressed": True,
+            "compression_points": [{"query_index": 17, "compressed_summary": PRIOR,
+                                    "compressed_token_count": 12}],
+        },
+    }
+    # What the turn actually replayed: the two turns after the saved point
+    # plus the in-flight one. Their indexes are 0-2 here, 18-19 in the DB.
+    synthetic = {"queries": [{"prompt": "q18", "response": "r"}, {"prompt": "q19", "response": ""}],
+                 "compression_metadata": {"is_compressed": False, "compression_points": []}}
+
+    (success, _), orchestrator = _run_with(_Handler(), agent, _result(), db_conversation, synthetic)
+
+    assert success is True
+    kwargs = orchestrator.compress_mid_execution.call_args.kwargs
+    sent = kwargs["current_conversation"]
+    points = sent["compression_metadata"]["compression_points"]
+    assert sent["compression_metadata"]["is_compressed"] is True
+    # The summary the system prompt carries is what the compressor builds on...
+    assert points[-1]["compressed_summary"] == PRIOR
+    # ...and it predates every synthetic query, so all of them are "new".
+    assert points[-1]["query_index"] == -1
+    # The persisted point must index the DATABASE conversation, not the
+    # shortened synthetic list.
+    assert kwargs["persist_query_index"] == 19
+
+
+@pytest.mark.unit
+def test_db_path_without_a_summary_in_play_sends_no_carried_point():
+    agent = _agent()
+    agent.compressed_summary = None
+    db_conversation = {"queries": [{"prompt": "q0", "response": "r"}]}
+    synthetic = {"queries": [{"prompt": "q0", "response": ""}],
+                 "compression_metadata": {"is_compressed": False, "compression_points": []}}
+
+    _, orchestrator = _run_with(_Handler(), agent, _result(), db_conversation, synthetic)
+
+    sent = orchestrator.compress_mid_execution.call_args.kwargs["current_conversation"]
+    assert sent["compression_metadata"]["compression_points"] == []
+
+
+@pytest.mark.unit
+def test_in_memory_path_carries_the_current_summary():
+    agent = _agent()
+    agent.compressed_summary = PRIOR
+    agent.decoded_token = {"sub": "user1"}
+    metadata = Mock()
+    metadata.compressed_token_count = 10
+    metadata.original_token_count = 100
+    metadata.compression_ratio = 10.0
+    metadata.timestamp = "2026-09-03T10:00:00+00:00"
+    metadata.to_dict.return_value = {"query_index": 1, "compressed_summary": "new", "compressed_token_count": 10}
+    svc = Mock()
+    svc.compress_conversation.return_value = metadata
+    svc.get_compressed_context.return_value = ("new", [])
+    handler = _Handler()
+    with patch("application.api.answer.services.compression.service.CompressionService", return_value=svc), patch(
+        "application.llm.llm_creator.LLMCreator"
+    ), patch("application.core.model_utils.get_provider_from_model_id", return_value="openai"), patch(
+        "application.core.model_utils.get_api_key_for_provider", return_value="sk"
+    ), patch.object(
+        handler, "_build_conversation_from_messages",
+        return_value={"queries": [{"prompt": "q", "response": "r"}, {"prompt": "q2", "response": ""}],
+                      "compression_metadata": {"is_compressed": False, "compression_points": []}},
+    ), patch.object(handler, "_rebuild_messages_after_compression", return_value=[{"role": "system", "content": "x"}]):
+        success, _ = handler._perform_in_memory_compression(agent, [{"role": "user", "content": "hi"}])
+
+    assert success is True
+    sent = svc.compress_conversation.call_args[0][0]
+    assert sent["compression_metadata"]["compression_points"][-1]["compressed_summary"] == PRIOR
diff --git a/tests/llm/test_responses_chain_budget.py b/tests/llm/test_responses_chain_budget.py
index 4cae62a7..3f1a3e7e 100644
--- a/tests/llm/test_responses_chain_budget.py
+++ b/tests/llm/test_responses_chain_budget.py
@@ -58,6 +58,11 @@ def _roles(items):
     return [i.get("role") for i in items if isinstance(i, dict) and i.get("role")]
 
 
+def _accepted(llm, rid="resp_1"):
+    """The provider accepted the request just built: record its response."""
+    llm._record_responses_metadata(types.SimpleNamespace(id=rid, output=[], usage=None))
+
+
 # ── chained system head ──────────────────────────────────────────────────────
 
 
@@ -69,6 +74,7 @@ def test_chained_input_omits_unchanged_system_head(monkeypatch):
     first, prev = llm._build_responses_input(_messages(), None)
     assert prev is None
     assert "system" in _roles(first)
+    _accepted(llm)
 
     chained, prev = llm._build_responses_input(_messages(), "resp_1")
     assert prev == "resp_1"
@@ -79,9 +85,11 @@ def test_chained_input_omits_unchanged_system_head(monkeypatch):
 def test_chained_input_resends_changed_system_head(monkeypatch):
     llm = _make_llm(monkeypatch)
     llm._build_responses_input(_messages("sys v1"), None)
+    _accepted(llm)
 
     chained, _ = llm._build_responses_input(_messages("sys v2"), "resp_1")
     assert _roles(chained) == ["system", "user"]
+    _accepted(llm, "resp_2")
 
     # ...and the new head becomes the one the chain holds.
     again, _ = llm._build_responses_input(_messages("sys v2"), "resp_2")
@@ -92,6 +100,7 @@ def test_chained_input_resends_changed_system_head(monkeypatch):
 def test_chained_system_head_hash_roundtrips_through_state(monkeypatch):
     llm = _make_llm(monkeypatch)
     llm._build_responses_input(_messages(), None)
+    _accepted(llm)
     state = llm.export_responses_state()
     assert state.get("system_hash")
 
@@ -105,6 +114,7 @@ def test_chained_system_head_hash_roundtrips_through_state(monkeypatch):
 def test_start_responses_turn_forgets_system_head(monkeypatch):
     llm = _make_llm(monkeypatch)
     llm._build_responses_input(_messages(), None)
+    _accepted(llm)
     llm.start_responses_turn()
     chained, _ = llm._build_responses_input(_messages(), "resp_1")
     # A fresh chain has no head on the server yet, so the head is sent.
@@ -312,3 +322,35 @@ def test_cache_key_for_user_is_opaque_and_stable():
     assert key == _cache_key_for_user("user_2Vhzgd63RSgixvvbF8Z2nhtqnE9")
     assert key != _cache_key_for_user("someone-else")
     assert _cache_key_for_user(None) is None
+
+
+# ── the head hash commits only once the provider accepted the request ───────
+
+
+@pytest.mark.unit
+def test_system_hash_commits_only_when_the_response_is_recorded(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages(), None)
+    # Not recorded (the request failed before the provider stored it): a
+    # chained follow-up must still carry the head.
+    chained, _ = llm._build_responses_input(_messages(), "resp_1")
+    assert _roles(chained) == ["system", "user"]
+    _accepted(llm)
+    chained, _ = llm._build_responses_input(_messages(), "resp_1")
+    assert _roles(chained) == ["user"]
+
+
+@pytest.mark.unit
+def test_failed_chained_request_resends_changed_head_on_retry(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages("sys v1"), None)
+    _accepted(llm)
+    first, _ = llm._build_responses_input(_messages("sys v2"), "resp_1")
+    assert _roles(first) == ["system", "user"]
+    # Transport error before any chunk: nothing recorded. The same-primary
+    # retry chains onto resp_1 again, whose stored head is still "sys v1".
+    retry, _ = llm._build_responses_input(_messages("sys v2"), "resp_1")
+    assert _roles(retry) == ["system", "user"]
+    _accepted(llm, "resp_2")
+    after, _ = llm._build_responses_input(_messages("sys v2"), "resp_2")
+    assert _roles(after) == ["user"]