From 04d358ab6943e72a9312c73f7820a246f95c7902 Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Sat, 5 Sep 2026 00:28:51 +0100
Subject: [PATCH 1/6] fix(llm,compression): bound cross-turn Responses chaining
 and make compression stick

In store mode every user turn chained onto the previous response, so the
provider's stored transcript grew without bound (measured: 889k prompt
tokens for a 37k-token saved history) while every local guard, the
compression pipeline included, measured the saved history. Each chained
tool round also re-sent the system message, which the server appends rather
than dedupes, and a saved compression point was applied exactly once, in the
turn that made it.

Chaining is now bounded. A turn starts from the saved history when the
previous turn's reported prompt reached the chain budget (default: the
model's context window), when the conversation was compressed after that
turn was produced, or when OPENAI_RESPONSES_CHAIN_ACROSS_TURNS is off.
Chained rounds omit an unchanged system head (hash carried in the persisted
Responses state). truncation="auto" is available behind a setting as a
backstop against a chain that outgrows the model's window.

Compression: a saved point is applied at every turn start; the threshold
counts the summary plus the queries after the point instead of the raw
history; re-compression summarises only the tail on top of the last point;
the mid-execution path marks itself persisted and resets the provider chain
so the rebuilt messages are the context; an empty summary is rejected; the
visible "[Context Compression Summary]" rows are no longer replayed as
history; appending the same point twice is a no-op.

Cache hints: a per-user prompt_cache_key and an optional
prompt_cache_retention on Responses API calls.

Measured on Azure with the same client shape as production (stateless
OpenAI client, server-side tools, PDF part): tokens billed on the sixth turn
fell from 58k to 35k, tool rounds add tens of tokens instead of ~2.8k, the
turn after a compression reused the saved summary in under two seconds
instead of re-summarising, and the round after a mid-execution compression
started from the compressed context instead of the full stored transcript.
---
 application/agents/base.py                    | 109 ++++++-
 .../services/compression/orchestrator.py      |  65 +++-
 .../answer/services/compression/service.py    |  54 +++-
 .../services/compression/threshold_checker.py |   5 +-
 .../services/compression/token_counter.py     |  34 ++
 .../api/answer/services/compression/types.py  |  46 ++-
 .../answer/services/conversation_service.py   |   3 +-
 .../api/answer/services/stream_processor.py   |  17 +-
 application/core/settings.py                  |  15 +
 application/llm/handlers/base.py              |  40 ++-
 application/llm/openai.py                     |  52 ++-
 .../storage/db/repositories/conversations.py  |  15 +
 docs/content/Deploying/DocsGPT-Settings.mdx   |   5 +
 docs/content/Models/cloud-providers.mdx       |   3 +
 .../compression/test_compression_reuse.py     | 246 ++++++++++++++
 .../services/compression/test_service.py      |   8 +-
 .../llm/handlers/test_mid_execution_chain.py  | 109 +++++++
 tests/llm/test_base.py                        |  12 +-
 tests/llm/test_responses_chain_budget.py      | 299 ++++++++++++++++++
 19 files changed, 1099 insertions(+), 38 deletions(-)
 create mode 100644 tests/api/answer/services/compression/test_compression_reuse.py
 create mode 100644 tests/llm/handlers/test_mid_execution_chain.py
 create mode 100644 tests/llm/test_responses_chain_budget.py

diff --git a/application/agents/base.py b/application/agents/base.py
index 003f2c01..7af5cfcd 100644
--- a/application/agents/base.py
+++ b/application/agents/base.py
@@ -2,6 +2,7 @@ import json
 import logging
 import uuid
 from abc import ABC, abstractmethod
+from datetime import datetime, timezone
 from typing import Any, Dict, Generator, List, Optional
 
 from application.agents.tool_executor import (
@@ -32,6 +33,39 @@ from application.logging import build_stack_data, log_activity, LogContext
 logger = logging.getLogger(__name__)
 
 
+def _parse_epoch(value: Any) -> Optional[datetime]:
+    """Parse a compression timestamp as stored anywhere it lands.
+
+    ``compression_points[].timestamp`` is written as an ISO datetime,
+    ``compression_metadata.last_compression_at`` as ``str(datetime)`` (space
+    separator), and the per-turn ``compression_epoch`` as whichever of those
+    the agent carried. All must compare equal.
+    """
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
+    text = str(value).strip()
+    if not text:
+        return None
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def _epoch_text(value: Any) -> Optional[str]:
+    """Serialize a compression timestamp for message metadata."""
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return str(value)
+
+
 class BaseAgent(ABC):
     # Inert defaults: an instance built without __init__ still resolves these.
     _guardrail_engine = None
@@ -64,6 +98,7 @@ class BaseAgent(ABC):
         limited_request_mode: Optional[bool] = False,
         request_limit: Optional[int] = settings.DEFAULT_AGENT_LIMITS["request_limit"],
         compressed_summary: Optional[str] = None,
+        last_compression_at: Optional[Any] = None,
         llm=None,
         llm_handler=None,
         tool_executor: Optional[ToolExecutor] = None,
@@ -160,6 +195,10 @@ class BaseAgent(ABC):
         self.limited_request_mode = limited_request_mode
         self.request_limit = request_limit
         self.compressed_summary = compressed_summary
+        # When the conversation was last compressed (any path). Stamped onto
+        # each turn's metadata so a later turn never chains onto a Responses
+        # id produced before a compression.
+        self.last_compression_at = last_compression_at
         self.current_token_count = 0
         self.context_limit_reached = False
         self.conversation_id: Optional[str] = None
@@ -314,13 +353,25 @@ class BaseAgent(ABC):
             **stored_metadata,
             "responses_state": state,
             "usage": getattr(self.llm, "_last_usage", None),
+            "compression_epoch": _epoch_text(getattr(self, "last_compression_at", None)),
         }
         metadata = {key: value for key, value in metadata.items() if value is not None}
         if metadata:
             yield {"metadata": metadata}
 
     def _previous_response_id(self) -> Optional[str]:
-        """Return the immediately preceding compatible Responses API id."""
+        """Return the preceding turn's Responses id when chaining onto it is safe.
+
+        Chaining keeps the provider's stored transcript (and its prompt cache)
+        warm across turns, but that transcript is invisible to every local
+        guard, so it is bounded. No chaining across turns when the operator
+        turned it off, when the previous turn's reported prompt already
+        reached the chain budget (default: the model's context window), or
+        when the conversation was compressed after that turn was produced —
+        the compressed local history is the context then, not the server's.
+        """
+        if not getattr(settings, "OPENAI_RESPONSES_CHAIN_ACROSS_TURNS", True):
+            return None
         if not self.chat_history:
             return None
         turn = self.chat_history[-1]
@@ -333,13 +384,49 @@ class BaseAgent(ABC):
         current_chain_key = (
             chain_key_factory() if callable(chain_key_factory) else None
         )
-        if (
+        if not (
             current_chain_key
             and meta.get("response_chain_key") == current_chain_key
             and meta.get("response_id")
         ):
+            return None
+
+        current_epoch = _parse_epoch(getattr(self, "last_compression_at", None))
+        if current_epoch is not None:
+            turn_epoch = _parse_epoch(meta.get("compression_epoch"))
+            if turn_epoch is None or turn_epoch < current_epoch:
+                logger.info(
+                    "Responses chain reset: the conversation was compressed after "
+                    "the previous turn; starting from the compressed local history"
+                )
+                return None
+
+        usage = meta.get("usage") if isinstance(meta.get("usage"), dict) else {}
+        try:
+            prompt_tokens = int(usage.get("prompt_tokens") or 0)
+        except (TypeError, ValueError):
+            prompt_tokens = 0
+        if not prompt_tokens:
+            # No provider-reported usage on the previous turn (older rows,
+            # estimate-only providers): nothing to bound against.
             return meta["response_id"]
-        return None
+        budget = getattr(settings, "OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS", None)
+        if not budget:
+            from application.core.model_utils import get_token_limit
+
+            budget = get_token_limit(
+                getattr(self, "model_id", None),
+                user_id=getattr(self, "model_user_id", None) or getattr(self, "user", None),
+            )
+        if budget and prompt_tokens >= int(budget):
+            logger.info(
+                "Responses chain budget reached (%s >= %s prompt tokens on the "
+                "previous turn); starting this turn from the local history",
+                prompt_tokens,
+                budget,
+            )
+            return None
+        return meta["response_id"]
 
     def _previous_responses_state(self) -> Optional[Dict[str, Any]]:
         """Return continuity state from the immediately preceding turn."""
@@ -1238,6 +1325,22 @@ class BaseAgent(ABC):
             previous_response_id = self._previous_response_id()
             if previous_response_id:
                 gen_kwargs["previous_response_id"] = previous_response_id
+                if not preserve_responses_state and hasattr(
+                    self.llm, "_chain_system_hash"
+                ):
+                    # The system head the chain last saw, so the first
+                    # chained call of this turn skips an unchanged system
+                    # message instead of appending a copy server-side.
+                    state = self._previous_responses_state() or {}
+                    self.llm._chain_system_hash = state.get("system_hash")
+        if hasattr(self.llm, "_prompt_cache_key"):
+            # Route a user's calls to the same cache shard. Keyed by user, not
+            # conversation: a new conversation has no id until its first turn
+            # is saved, and a key that appears on turn two would look up a
+            # different shard from the one turn one populated.
+            self.llm._prompt_cache_key = (
+                getattr(self, "initial_user_id", None) or getattr(self, "user", None)
+            )
 
         # Forward OpenAI sampling params (temperature, max_tokens, top_p, ...).
         if self.llm_params:
diff --git a/application/api/answer/services/compression/orchestrator.py b/application/api/answer/services/compression/orchestrator.py
index 2a0c0cdd..c23e7e43 100644
--- a/application/api/answer/services/compression/orchestrator.py
+++ b/application/api/answer/services/compression/orchestrator.py
@@ -1,14 +1,13 @@
 """High-level compression orchestration."""
 
 import logging
-from typing import Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, Optional
 
 from application.api.answer.services.compression.service import CompressionService
 from application.api.answer.services.compression.threshold_checker import (
     CompressionThresholdChecker,
 )
 from application.api.answer.services.compression.types import CompressionResult
-from application.api.answer.services.conversation_service import ConversationService
 from application.core.model_utils import (
     get_api_key_for_provider,
     get_provider_from_model_id,
@@ -16,6 +15,9 @@ from application.core.model_utils import (
 from application.core.settings import settings
 from application.llm.llm_creator import LLMCreator
 
+if TYPE_CHECKING:  # pragma: no cover - annotation only
+    from application.api.answer.services.conversation_service import ConversationService
+
 logger = logging.getLogger(__name__)
 
 
@@ -29,7 +31,7 @@ class CompressionOrchestrator:
 
     def __init__(
         self,
-        conversation_service: ConversationService,
+        conversation_service: "ConversationService",
         threshold_checker: Optional[CompressionThresholdChecker] = None,
     ):
         """
@@ -89,6 +91,23 @@ class CompressionOrchestrator:
                 current_query_tokens,
                 user_id=registry_user_id,
             ):
+                compression_meta = conversation.get("compression_metadata") or {}
+                points = compression_meta.get("compression_points") or []
+                if compression_meta.get("is_compressed") and points:
+                    # 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).
+                    summary, recent = self._read_only_service(
+                        model_id
+                    ).get_compressed_context(conversation)
+                    return CompressionResult.success_from_existing(
+                        summary,
+                        recent,
+                        last_compression_at=(
+                            compression_meta.get("last_compression_at")
+                            or points[-1].get("timestamp")
+                        ),
+                    )
                 # No compression needed, return full history
                 queries = conversation.get("queries", [])
                 return CompressionResult.success_no_compression(queries)
@@ -109,6 +128,12 @@ class CompressionOrchestrator:
             )
             return CompressionResult.failure(str(e))
 
+    def _read_only_service(self, model_id: str) -> CompressionService:
+        """A service for applying saved compression points; no LLM is involved."""
+        return CompressionService(
+            llm=None, model_id=model_id, conversation_service=self.conversation_service
+        )
+
     def _perform_compression(
         self,
         conversation_id: str,
@@ -171,7 +196,8 @@ class CompressionOrchestrator:
                 conversation_service=self.conversation_service,
             )
 
-            # Compress all queries up to the latest
+            # Compress everything since the last compression point up to the
+            # latest query; the earlier queries are already in that point.
             queries_count = len(conversation.get("queries", []))
             compress_up_to = queries_count - 1
 
@@ -179,14 +205,41 @@ class CompressionOrchestrator:
                 logger.warning("No queries to compress")
                 return CompressionResult.success_no_compression([])
 
+            points = (conversation.get("compression_metadata") or {}).get(
+                "compression_points"
+            ) or []
+            start_index = 0
+            if points:
+                try:
+                    start_index = int(points[-1].get("query_index", -1)) + 1
+                except (TypeError, ValueError):
+                    start_index = 0
+            if start_index > compress_up_to:
+                logger.info(
+                    f"No new queries since the last compression point for "
+                    f"conversation {conversation_id}; reusing its summary"
+                )
+                summary, recent = compression_service.get_compressed_context(
+                    conversation
+                )
+                return CompressionResult.success_from_existing(
+                    summary, recent, last_compression_at=points[-1].get("timestamp")
+                )
+
             logger.info(
                 f"Initiating compression for conversation {conversation_id}: "
-                f"compressing all {queries_count} queries (0-{compress_up_to})"
+                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 ""
+                )
             )
 
             # Perform compression and save to DB
             metadata = compression_service.compress_and_save(
-                conversation_id, conversation, compress_up_to
+                conversation_id, conversation, compress_up_to, start_index=start_index
             )
 
             logger.info(
diff --git a/application/api/answer/services/compression/service.py b/application/api/answer/services/compression/service.py
index 2934d79e..0b51e401 100644
--- a/application/api/answer/services/compression/service.py
+++ b/application/api/answer/services/compression/service.py
@@ -11,6 +11,7 @@ from application.api.answer.services.compression.prompt_builder import (
 from application.api.answer.services.compression.token_counter import TokenCounter
 from application.api.answer.services.compression.types import (
     CompressionMetadata,
+    is_compression_summary_row,
 )
 from application.core.settings import settings
 
@@ -51,6 +52,7 @@ class CompressionService:
         self,
         conversation: Dict[str, Any],
         compress_up_to_index: int,
+        start_index: int = 0,
     ) -> CompressionMetadata:
         """
         Compress conversation history up to specified index.
@@ -74,8 +76,14 @@ class CompressionService:
                     f"(conversation has {len(queries)} queries)"
                 )
 
-            # Get queries to compress
-            queries_to_compress = queries[: compress_up_to_index + 1]
+            if start_index < 0 or start_index > compress_up_to_index:
+                raise ValueError(
+                    f"Nothing to compress: start_index {start_index} is past "
+                    f"compress_up_to_index {compress_up_to_index}"
+                )
+            # Only the queries after the previous compression point are new;
+            # the earlier ones are already inside that point's summary.
+            queries_to_compress = queries[start_index : compress_up_to_index + 1]
 
             # Check if there are existing compressions. ``compression_metadata``
             # is a nullable JSONB column, so a never-compressed conversation
@@ -85,14 +93,26 @@ class CompressionService:
                 "compression_points", []
             )
 
+            previous_summary_tokens = 0
             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]]
+                previous_summary_tokens = TokenCounter.count_message_tokens(
+                    [{"content": existing_compressions[0].get("compressed_summary", "")}]
+                )
                 logger.info(
-                    f"Found {len(existing_compressions)} previous compression(s) - "
-                    f"will incorporate into new summary"
+                    "Found a previous compression point (query %s) - "
+                    "the new summary builds on it",
+                    existing_compressions[0].get("query_index"),
                 )
 
-            # Calculate original token count
-            original_tokens = TokenCounter.count_query_tokens(queries_to_compress)
+            # Calculate original token count: everything the new summary replaces
+            original_tokens = (
+                TokenCounter.count_query_tokens(queries_to_compress)
+                + previous_summary_tokens
+            )
 
             # Log tool call stats
             self._log_tool_call_stats(queries_to_compress)
@@ -105,7 +125,7 @@ class CompressionService:
             # Call LLM to generate compression
             logger.info(
                 f"Starting compression: {len(queries_to_compress)} queries "
-                f"(messages 0-{compress_up_to_index}, {original_tokens} tokens) "
+                f"(messages {start_index}-{compress_up_to_index}, {original_tokens} tokens) "
                 f"using model {self.model_id}"
             )
 
@@ -126,6 +146,13 @@ class CompressionService:
                 [{"content": compressed_summary}]
             )
 
+            # An empty summary is not a compression: it replaced a 494k-token
+            # conversation with nothing in prod (2026-08-27) while reporting
+            # success.
+            if not compressed_summary.strip() or compressed_tokens <= 0:
+                raise ValueError(
+                    "Compression produced an empty summary; keeping original history"
+                )
             # Calculate compression ratio
             compression_ratio = (
                 original_tokens / compressed_tokens if compressed_tokens > 0 else 0
@@ -171,6 +198,7 @@ class CompressionService:
         conversation_id: str,
         conversation: Dict[str, Any],
         compress_up_to_index: int,
+        start_index: int = 0,
     ) -> CompressionMetadata:
         """
         Compress conversation and save to database.
@@ -192,7 +220,9 @@ class CompressionService:
             )
 
         # Perform compression
-        metadata = self.compress_conversation(conversation, compress_up_to_index)
+        metadata = self.compress_conversation(
+            conversation, compress_up_to_index, start_index=start_index
+        )
 
         # Save to database
         self.conversation_service.update_compression_metadata(
@@ -248,7 +278,13 @@ class CompressionService:
             # Get only messages after compression point
             queries = conversation.get("queries", [])
             total_queries = len(queries)
-            recent_queries = queries[last_compressed_index + 1 :]
+            # The visible summary rows appended after a compression are not
+            # history: their content already rides in the system prompt.
+            recent_queries = [
+                q
+                for q in queries[last_compressed_index + 1 :]
+                if not is_compression_summary_row(q)
+            ]
 
             logger.info(
                 f"Using compressed context: summary ({compressed_tokens} tokens, "
diff --git a/application/api/answer/services/compression/threshold_checker.py b/application/api/answer/services/compression/threshold_checker.py
index b885b642..69bdbf8b 100644
--- a/application/api/answer/services/compression/threshold_checker.py
+++ b/application/api/answer/services/compression/threshold_checker.py
@@ -46,8 +46,9 @@ class CompressionThresholdChecker:
             True if tokens >= threshold% of context window
         """
         try:
-            # Calculate total tokens in conversation
-            total_tokens = TokenCounter.count_conversation_tokens(conversation)
+            # What the next turn will replay: summary + queries after the
+            # last compression point, or the raw history when never compressed.
+            total_tokens = TokenCounter.count_effective_conversation_tokens(conversation)
             total_tokens += current_query_tokens
 
             # Get context window limit for model
diff --git a/application/api/answer/services/compression/token_counter.py b/application/api/answer/services/compression/token_counter.py
index d2cbfc34..462abb72 100644
--- a/application/api/answer/services/compression/token_counter.py
+++ b/application/api/answer/services/compression/token_counter.py
@@ -5,6 +5,7 @@ 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
 
 logger = logging.getLogger(__name__)
 
@@ -131,3 +132,36 @@ class TokenCounter:
         except Exception as e:
             logger.error(f"Error calculating conversation tokens: {str(e)}")
             return 0
+
+    @staticmethod
+    def count_effective_conversation_tokens(conversation: Dict[str, Any]) -> int:
+        """Tokens the next turn will actually replay.
+
+        The latest summary plus the queries after its compression point, or
+        everything when the conversation was never compressed.
+        ``count_conversation_tokens`` counts the raw history regardless, which
+        is what made every turn after a compression trigger it again.
+        """
+        try:
+            queries = conversation.get("queries", []) or []
+            metadata = conversation.get("compression_metadata") or {}
+            points = metadata.get("compression_points") or []
+            if not (metadata.get("is_compressed") and points):
+                return TokenCounter.count_query_tokens(queries)
+            latest = points[-1] or {}
+            try:
+                last_index = int(latest.get("query_index", -1))
+            except (TypeError, ValueError):
+                last_index = -1
+            recent = [
+                q
+                for q in queries[last_index + 1 :]
+                if not is_compression_summary_row(q)
+            ]
+            summary_tokens = TokenCounter.count_message_tokens(
+                [{"content": latest.get("compressed_summary") or ""}]
+            )
+            return summary_tokens + TokenCounter.count_query_tokens(recent)
+        except Exception as e:
+            logger.error(f"Error calculating effective conversation tokens: {str(e)}")
+            return 0
diff --git a/application/api/answer/services/compression/types.py b/application/api/answer/services/compression/types.py
index cfd77618..ff0ff4f3 100644
--- a/application/api/answer/services/compression/types.py
+++ b/application/api/answer/services/compression/types.py
@@ -4,6 +4,16 @@ from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
+# ``prompt`` of the visible summary row appended after a compression. These
+# rows are display-only: their content already reaches the model through the
+# system prompt (``compressed_summary``), so history replay skips them.
+COMPRESSION_SUMMARY_PROMPT = "[Context Compression Summary]"
+
+
+def is_compression_summary_row(query: Any) -> bool:
+    """True for the visible summary row written by ``append_compression_message``."""
+    return isinstance(query, dict) and query.get("prompt") == COMPRESSION_SUMMARY_PROMPT
+
 
 @dataclass
 class CompressionMetadata:
@@ -42,10 +52,17 @@ class CompressionResult:
     metadata: Optional[CompressionMetadata] = None
     error: Optional[str] = None
     compression_performed: bool = False
+    # When the conversation was last compressed — the point this turn's
+    # history reflects, whether it was made now or reused from the DB.
+    last_compression_at: Optional[Any] = None
 
     @classmethod
     def success_with_compression(
-        cls, summary: str, queries: List[Dict], metadata: CompressionMetadata
+        cls,
+        summary: str,
+        queries: List[Dict],
+        metadata: CompressionMetadata,
+        last_compression_at: Optional[Any] = None,
     ) -> "CompressionResult":
         """Create a successful result with compression."""
         return cls(
@@ -54,6 +71,31 @@ class CompressionResult:
             recent_queries=queries,
             metadata=metadata,
             compression_performed=True,
+            last_compression_at=(
+                last_compression_at
+                if last_compression_at is not None
+                else getattr(metadata, "timestamp", None)
+            ),
+        )
+
+    @classmethod
+    def success_from_existing(
+        cls,
+        summary: Optional[str],
+        queries: List[Dict],
+        last_compression_at: Optional[Any] = None,
+    ) -> "CompressionResult":
+        """A previously saved compression point applied to this turn.
+
+        No LLM call was made; the summary and the queries after its point
+        are what the turn replays instead of the raw history.
+        """
+        return cls(
+            success=True,
+            compressed_summary=summary,
+            recent_queries=queries,
+            compression_performed=False,
+            last_compression_at=last_compression_at,
         )
 
     @classmethod
@@ -81,6 +123,8 @@ class CompressionResult:
         """
         out: List[Dict[str, str]] = []
         for q in self.recent_queries:
+            if is_compression_summary_row(q):
+                continue
             entry: Dict[str, str] = {
                 "prompt": q["prompt"],
                 "response": q["response"],
diff --git a/application/api/answer/services/conversation_service.py b/application/api/answer/services/conversation_service.py
index 6a0075b9..f1522240 100644
--- a/application/api/answer/services/conversation_service.py
+++ b/application/api/answer/services/conversation_service.py
@@ -14,6 +14,7 @@ from typing import Any, Dict, List, Optional
 from sqlalchemy import text as sql_text
 
 from application.core.settings import settings
+from application.api.answer.services.compression.types import COMPRESSION_SUMMARY_PROMPT
 from application.storage.db.base_repository import looks_like_uuid
 from application.storage.db.repositories.agents import AgentsRepository
 from application.storage.db.repositories.conversations import (
@@ -603,7 +604,7 @@ class ConversationService:
                     str(conv["id"]) if conv is not None else conversation_id
                 )
                 repo.append_message(conv_pg_id, {
-                    "prompt": "[Context Compression Summary]",
+                    "prompt": COMPRESSION_SUMMARY_PROMPT,
                     "response": summary,
                     "thought": "",
                     "sources": [],
diff --git a/application/api/answer/services/stream_processor.py b/application/api/answer/services/stream_processor.py
index 44101f3e..764aacc0 100644
--- a/application/api/answer/services/stream_processor.py
+++ b/application/api/answer/services/stream_processor.py
@@ -8,6 +8,7 @@ from application.agents.agent_creator import AgentCreator
 from application.agents.default_tools import synthesized_default_tools
 from application.api.answer.services.compression import CompressionOrchestrator
 from application.api.answer.services.compression.token_counter import TokenCounter
+from application.api.answer.services.compression.types import is_compression_summary_row
 from application.api.answer.services.conversation_service import ConversationService
 from application.prompts.composer import compose_preset, is_composed_preset
 from application.api.answer.services.prompt_renderer import (
@@ -146,6 +147,9 @@ class StreamProcessor:
         self._required_tool_actions: Optional[Dict[str, Set[Optional[str]]]] = None
         self.compressed_summary: Optional[str] = None
         self.compressed_summary_tokens: int = 0
+        # When the conversation's history was last compressed (DB point or
+        # one made this turn); the agent stamps it on the turn's metadata.
+        self.last_compression_at: Optional[Any] = None
         self._agent_data: Optional[Dict[str, Any]] = None
 
     def initialize(self):
@@ -313,6 +317,7 @@ class StreamProcessor:
                         ),
                     }
                     for query in conversation.get("queries", [])
+                    if not is_compression_summary_row(query)
                 ]
         else:
             # model_user_id keeps history trim aligned with the BYOM's
@@ -355,10 +360,11 @@ class StreamProcessor:
                         ),
                     }
                     for query in conversation.get("queries", [])
+                    if not is_compression_summary_row(query)
                 ]
                 return
 
-            if result.compression_performed and result.compressed_summary:
+            if result.compressed_summary:
                 self.compressed_summary = result.compressed_summary
                 self.compressed_summary_tokens = TokenCounter.count_message_tokens(
                     [{"content": result.compressed_summary}]
@@ -366,11 +372,17 @@ class StreamProcessor:
                 logger.info(
                     f"Using compressed summary ({self.compressed_summary_tokens} tokens) "
                     f"+ {len(result.recent_queries)} recent messages"
+                    + ("" if result.compression_performed else " (saved compression point)")
                 )
+            self.last_compression_at = result.last_compression_at
 
             self.history = result.as_history()
             # Preserve metadata from recent queries (as_history only has prompt/response)
-            recent = result.recent_queries if result.recent_queries else conversation.get("queries", [])
+            recent = [
+                q
+                for q in (result.recent_queries or conversation.get("queries", []))
+                if not is_compression_summary_row(q)
+            ]
             for i, entry in enumerate(self.history):
                 # Match by index from the end of recent queries
                 offset = len(recent) - len(self.history)
@@ -1741,6 +1753,7 @@ class StreamProcessor:
             "llm_params": self.data.get("llm_params") or {},
             "multimodal_content": request_multimodal,
             "compressed_summary": self.compressed_summary,
+            "last_compression_at": self.last_compression_at,
             "llm": llm,
             "llm_handler": llm_handler,
             "tool_executor": tool_executor,
diff --git a/application/core/settings.py b/application/core/settings.py
index 4113171b..56cea673 100644
--- a/application/core/settings.py
+++ b/application/core/settings.py
@@ -369,6 +369,21 @@ class Settings(BaseSettings):
     # True persists Responses API calls server-side so previous_response_id can chain turns.
     # False keeps them stateless, carrying reasoning across the tool loop as encrypted items.
     OPENAI_RESPONSES_STORE: bool = False
+    # Cross-turn ``previous_response_id`` chaining (store mode only). The
+    # chained transcript lives on the provider and is invisible to every
+    # local guard, so it is bounded: a turn starts from the local history
+    # when the previous turn's reported prompt already reached the budget
+    # (default: the model's context window) or when the conversation was
+    # compressed after that turn was produced.
+    OPENAI_RESPONSES_CHAIN_ACROSS_TURNS: bool = True
+    OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS: Optional[int] = None
+    # ``truncation: "auto"`` lets the provider drop middle items instead of
+    # failing every request once a chain exceeds the model's native window.
+    OPENAI_RESPONSES_TRUNCATION_AUTO: bool = False
+    # Prompt-cache hints on the Responses API: route a user's calls to the
+    # same cache shard, and request extended retention where offered.
+    OPENAI_PROMPT_CACHE_KEY: bool = True
+    OPENAI_PROMPT_CACHE_RETENTION: Optional[str] = None
     OPENAI_REASONING_SUMMARY: str = "auto"
 
     # Lets OpenAI-compatible clients identify a logical chat by session header, which
diff --git a/application/llm/handlers/base.py b/application/llm/handlers/base.py
index 02f43b1b..6de612d0 100644
--- a/application/llm/handlers/base.py
+++ b/application/llm/handlers/base.py
@@ -592,6 +592,22 @@ class LLMHandler(ABC):
             include_tool_calls=include_tool_calls,
         )
 
+    @staticmethod
+    def _reset_responses_chain(agent, metadata) -> None:
+        """After a compression the rebuilt local messages ARE the context.
+
+        Forget the provider-side chain so the next call is not prepended with
+        the uncompressed transcript (prod: 236k input tokens billed for a
+        four-message rebuilt input), and stamp the compression epoch so later
+        turns do not chain onto pre-compression responses either.
+        """
+        starter = getattr(getattr(agent, "llm", None), "start_responses_turn", None)
+        if callable(starter):
+            starter()
+        timestamp = getattr(metadata, "timestamp", None)
+        if timestamp is not None:
+            agent.last_compression_at = timestamp
+
     def _perform_mid_execution_compression(
         self, agent, messages: List[Dict]
     ) -> tuple[bool, Optional[List[Dict]]]:
@@ -661,11 +677,18 @@ class LLMHandler(ABC):
                 logger.warning("Compression not performed")
                 return False, None
 
-            # Check if compression actually reduced tokens
+            # Check if compression actually reduced tokens — and produced
+            # anything at all: a 0-token "summary" replaced a 494k-token
+            # conversation with nothing in prod (2026-08-27).
             if result.metadata:
-                if result.metadata.compressed_token_count >= result.metadata.original_token_count:
+                if (
+                    result.metadata.compressed_token_count <= 0
+                    or result.metadata.compressed_token_count
+                    >= result.metadata.original_token_count
+                ):
                     logger.warning(
-                        "Compression did not reduce token count; falling back to minimal pruning"
+                        "Compression did not reduce token count (or produced an "
+                        "empty summary); falling back to minimal pruning"
                     )
                     pruned = self._prune_messages_minimal(messages)
                     if pruned:
@@ -685,10 +708,15 @@ class LLMHandler(ABC):
                     agent.conversation_id, result.metadata.to_dict()
                 )
 
-            # Update agent's compressed summary for downstream persistence
+            # Update agent's compressed summary for downstream persistence.
+            # ``compress_mid_execution`` already wrote the compression point
+            # and the row above is the visible summary, so this IS saved —
+            # leaving the flag False made the route persist both again
+            # (prod: every mid-execution compression stored twice).
             agent.compressed_summary = result.compressed_summary
             agent.compression_metadata = result.metadata.to_dict() if result.metadata else None
-            agent.compression_saved = False
+            agent.compression_saved = True
+            self._reset_responses_chain(agent, result.metadata)
 
             # Reset the context limit flag so tools can continue
             agent.context_limit_reached = False
@@ -827,7 +855,9 @@ class LLMHandler(ABC):
 
             agent.compressed_summary = compressed_summary
             agent.compression_metadata = metadata.to_dict()
+            # Nothing reached the DB on this path; the route persists once.
             agent.compression_saved = False
+            self._reset_responses_chain(agent, metadata)
             agent.context_limit_reached = False
             agent.current_token_count = 0
 
diff --git a/application/llm/openai.py b/application/llm/openai.py
index 6fcc8bc3..5e61ca16 100644
--- a/application/llm/openai.py
+++ b/application/llm/openai.py
@@ -259,6 +259,13 @@ class OpenAILLM(BaseLLM):
         # fallback restream of the already-delivered answer.
         self._stream_reached_finish = False
         self._imported_response_id = None
+        # Hash of the system head the current Responses chain already holds.
+        # The server appends a re-sent system message after the stored
+        # transcript instead of deduping it, so a chained round only carries
+        # the head when it changed.
+        self._chain_system_hash = None
+        # Prompt-cache routing key (the user id), set by the agent per call.
+        self._prompt_cache_key = None
         # Files-API ids for inline ``file_data`` content parts already
         # uploaded, keyed by content hash. First-line cache for the
         # in-request tool loop; the Redis-backed cross-request cache
@@ -304,6 +311,7 @@ class OpenAILLM(BaseLLM):
             "call_ids": sorted(self._last_response_call_ids or ()),
             "reasoning_items": self._last_reasoning_items,
             "reasoning_for_calls": self._reasoning_for_calls,
+            "system_hash": self._chain_system_hash,
         }
 
     def import_responses_state(self, state: dict | None) -> bool:
@@ -318,6 +326,7 @@ class OpenAILLM(BaseLLM):
         self._last_response_call_ids = set(state.get("call_ids") or ())
         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")
         return True
 
     def start_responses_turn(self) -> None:
@@ -327,6 +336,7 @@ class OpenAILLM(BaseLLM):
         self._last_response_id = None
         self._last_response_call_ids = set()
         self._imported_response_id = None
+        self._chain_system_hash = None
         self._last_finish_reason = None
 
     def _resolve_file_part(self, item):
@@ -1099,6 +1109,19 @@ class OpenAILLM(BaseLLM):
         ]
         return head + carried + messages[last_assistant + 1:]
 
+    @staticmethod
+    def _system_fingerprint(messages):
+        """Hash of the system messages' content; ``None`` when there are none."""
+        parts = [
+            m.get("content")
+            for m in messages
+            if isinstance(m, dict) and m.get("role") == "system"
+        ]
+        if not parts:
+            return None
+        encoded = json.dumps(parts, sort_keys=True, default=str).encode("utf-8")
+        return hashlib.sha256(encoded).hexdigest()
+
     def _build_responses_input(self, messages, previous_response_id):
         """Build the Responses ``input`` list, honouring store-mode chaining.
 
@@ -1119,11 +1142,21 @@ class OpenAILLM(BaseLLM):
         """
         chained = bool(previous_response_id and settings.OPENAI_RESPONSES_STORE)
         if not chained:
+            self._chain_system_hash = self._system_fingerprint(messages)
             return self._to_responses_input(messages, chained=False), None
 
-        input_items = self._to_responses_input(
-            self._trim_for_previous_response(messages), chained=True
-        )
+        trimmed = self._trim_for_previous_response(messages)
+        head_hash = self._system_fingerprint(trimmed)
+        if head_hash and head_hash == self._chain_system_hash:
+            # The server appends a re-sent system message after the stored
+            # transcript instead of replacing it (measured on Azure: 11,239
+            # vs 5,635 input tokens for the same chained turn), so an
+            # 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
+        input_items = self._to_responses_input(trimmed, chained=True)
         expected = set(self._last_response_call_ids or ())
         if expected:
             kept, dropped = [], []
@@ -1156,6 +1189,7 @@ class OpenAILLM(BaseLLM):
                     "request the provider would reject",
                     sorted(missing),
                 )
+                self._chain_system_hash = self._system_fingerprint(messages)
                 return self._to_responses_input(messages, chained=False), None
         return input_items, previous_response_id
 
@@ -1258,6 +1292,18 @@ class OpenAILLM(BaseLLM):
         # carryover working whether or not the response is also retained
         # server-side (store=true).
         params["include"] = ["reasoning.encrypted_content"]
+        # Backstop against a chain that outgrows the model's native window:
+        # the provider drops middle items instead of failing every request.
+        if getattr(settings, "OPENAI_RESPONSES_TRUNCATION_AUTO", False):
+            params["truncation"] = "auto"
+        # Prompt-cache hints. The key pins a conversation to one cache shard;
+        # retention asks for the extended tier where the deployment offers it.
+        cache_key = getattr(self, "_prompt_cache_key", None)
+        if cache_key and getattr(settings, "OPENAI_PROMPT_CACHE_KEY", False):
+            params["prompt_cache_key"] = str(cache_key)
+        retention = getattr(settings, "OPENAI_PROMPT_CACHE_RETENTION", None)
+        if retention:
+            params["prompt_cache_retention"] = retention
         return params
 
     @staticmethod
diff --git a/application/storage/db/repositories/conversations.py b/application/storage/db/repositories/conversations.py
index cd27bbea..7105c3be 100644
--- a/application/storage/db/repositories/conversations.py
+++ b/application/storage/db/repositories/conversations.py
@@ -506,6 +506,21 @@ class ConversationsRepository:
         # Shape-gate: see ``set_compression_flags``.
         if not looks_like_uuid(conversation_id):
             return False
+        # Idempotent on the same point: two persist paths (the mid-execution
+        # handler and the route's end-of-stream save) used to both append it.
+        last_point = self._conn.execute(
+            text(
+                "SELECT compression_metadata -> 'compression_points' -> -1 "
+                "FROM conversations WHERE id = CAST(:id AS uuid)"
+            ),
+            {"id": conversation_id},
+        ).scalar()
+        if (
+            isinstance(last_point, dict)
+            and last_point.get("query_index") == point.get("query_index")
+            and last_point.get("compressed_summary") == point.get("compressed_summary")
+        ):
+            return True
         result = self._conn.execute(
             text(
                 """
diff --git a/docs/content/Deploying/DocsGPT-Settings.mdx b/docs/content/Deploying/DocsGPT-Settings.mdx
index dab7f779..c8368303 100644
--- a/docs/content/Deploying/DocsGPT-Settings.mdx
+++ b/docs/content/Deploying/DocsGPT-Settings.mdx
@@ -497,6 +497,11 @@ See [Access Control, Roles & Teams](/Deploying/Access-Control) for the full mode
 | Setting | Default | Description |
 | --- | --- | --- |
 | `OPENAI_RESPONSES_STORE` | `false` | When `true`, allows OpenAI to persist [Responses API](/Models/cloud-providers#openai-responses-api-and-reasoning) state server-side. |
+| `OPENAI_RESPONSES_CHAIN_ACROSS_TURNS` | `true` | In store mode, chain a new user turn onto the previous turn's response with `previous_response_id`. Set to `false` for stateless turns that always rebuild from the saved history. |
+| `OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS` | unset | Stop chaining across turns once the previous turn's reported prompt reached this many tokens; the turn then starts from the saved (compressible) history. Unset means the model's `context_window`. |
+| `OPENAI_RESPONSES_TRUNCATION_AUTO` | `false` | Send `truncation: "auto"` on Responses API calls so the provider drops middle items instead of rejecting a request that exceeds the model's window. |
+| `OPENAI_PROMPT_CACHE_KEY` | `true` | Send a `prompt_cache_key` (the user id) on Responses API calls so a user's requests route to the same prompt-cache shard. |
+| `OPENAI_PROMPT_CACHE_RETENTION` | unset | Value for `prompt_cache_retention` on Responses API calls, for example `24h`, where the provider offers extended prompt-cache retention. |
 | `V1_SESSION_TTL_SECONDS` | `86400` | Redis correlation lifetime, in seconds, for OpenAI-compatible client sessions. Raw external session IDs are hashed and never stored. |
 | `TITLE_MODEL_ID` | unset | Optional model ID used for asynchronous first-party conversation titles. When unset, the answer model is used. Hidden API conversations never invoke title generation. |
 
diff --git a/docs/content/Models/cloud-providers.mdx b/docs/content/Models/cloud-providers.mdx
index 7adb3710..e230bef3 100644
--- a/docs/content/Models/cloud-providers.mdx
+++ b/docs/content/Models/cloud-providers.mdx
@@ -66,6 +66,9 @@ For OpenAI models that support it, DocsGPT can call the newer **Responses API**
 
 - `reasoning_effort` — per-model reasoning effort hint (for example `medium`) declared in the model catalog.
 - `OPENAI_RESPONSES_STORE` (default `false`) — when `true`, lets OpenAI persist Responses API state server-side and chains calls with `previous_response_id`. When `false`, DocsGPT requests encrypted reasoning items and persists those ciphertext items itself so reasoning survives tool calls and later conversation turns without OpenAI retaining the response.
+- `OPENAI_RESPONSES_CHAIN_ACROSS_TURNS` (default `true`) and `OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS` (default: the model's `context_window`) — in store mode a new user turn chains onto the previous response, which keeps the provider's stored transcript and its prompt cache warm. The chain is bounded: once the previous turn's reported prompt reaches the budget, or after the conversation history has been compressed, the next turn starts from DocsGPT's own saved history instead. Chained tool rounds do not re-send an unchanged system message.
+- `OPENAI_RESPONSES_TRUNCATION_AUTO` (default `false`) — send `truncation: "auto"` so the provider trims middle items instead of failing a request that exceeds the model's window.
+- `OPENAI_PROMPT_CACHE_KEY` (default `true`) and `OPENAI_PROMPT_CACHE_RETENTION` (default unset) — prompt-cache hints on Responses API calls: a per-user `prompt_cache_key`, and a `prompt_cache_retention` value such as `24h` where the provider supports it.
 - `V1_SESSION_TTL_SECONDS` (default `86400`) — lifetime of the hashed Redis mapping used to associate OpenAI-compatible client session headers with hidden DocsGPT conversations. Explicit DocsGPT conversation IDs always take precedence.
 
 See [App Configuration](/Deploying/DocsGPT-Settings) for the full settings reference.
diff --git a/tests/api/answer/services/compression/test_compression_reuse.py b/tests/api/answer/services/compression/test_compression_reuse.py
new file mode 100644
index 00000000..53235811
--- /dev/null
+++ b/tests/api/answer/services/compression/test_compression_reuse.py
@@ -0,0 +1,246 @@
+"""Compression must stick: a saved summary is applied on every later turn,
+re-compression only summarises the tail, an empty summary is a failure, and
+the visible ``[Context Compression Summary]`` rows are never replayed.
+
+Background (prod + OSS reproduction, 2026-09-01/03): the turn-start path
+returned the full raw history whenever the conversation was under the
+threshold, so a summary was used exactly once; over the threshold it
+re-summarised everything from query 0 on every turn (14-24 s each); one
+conversation was "compressed" to a 0-token summary and carried on with
+nothing.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from application.api.answer.services.compression import CompressionService
+from application.api.answer.services.compression.orchestrator import (
+    CompressionOrchestrator,
+)
+from application.api.answer.services.compression.threshold_checker import (
+    CompressionThresholdChecker,
+)
+from application.api.answer.services.compression.token_counter import TokenCounter
+from application.api.answer.services.compression.types import CompressionResult
+from application.api.answer.services.conversation_service import (
+    COMPRESSION_SUMMARY_PROMPT,
+)
+
+EPOCH = "2026-09-03T09:00:00+00:00"
+POINT = {
+    "timestamp": EPOCH,
+    "query_index": 1,
+    "compressed_summary": "S",
+    "original_token_count": 900,
+    "compressed_token_count": 5,
+    "compression_ratio": 180.0,
+    "model_used": "m",
+    "compression_prompt_version": "v1.0",
+}
+
+
+def _compressed_conversation():
+    big = "word " * 600
+    return {
+        "queries": [
+            {"prompt": "q0", "response": big},
+            {"prompt": "q1", "response": big},
+            {"prompt": "q2", "response": "r2"},
+            {"prompt": "q3", "response": "r3"},
+        ],
+        "compression_metadata": {
+            "is_compressed": True,
+            "last_compression_at": EPOCH,
+            "compression_points": [POINT],
+        },
+        "agent_id": "agent-1",
+    }
+
+
+@pytest.fixture
+def conversation_service():
+    return MagicMock()
+
+
+@pytest.fixture
+def threshold_checker():
+    return MagicMock()
+
+
+@pytest.fixture
+def orchestrator(conversation_service, threshold_checker):
+    return CompressionOrchestrator(
+        conversation_service=conversation_service, threshold_checker=threshold_checker
+    )
+
+
+@pytest.mark.unit
+class TestTurnStartReuse:
+    def test_under_threshold_returns_existing_summary_and_recent(
+        self, orchestrator, conversation_service, threshold_checker
+    ):
+        conversation_service.get_conversation.return_value = _compressed_conversation()
+        threshold_checker.should_compress.return_value = False
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.success is True
+        assert result.compression_performed is False
+        assert result.compressed_summary == "S"
+        assert [q["prompt"] for q in result.recent_queries] == ["q2", "q3"]
+        assert result.last_compression_at == EPOCH
+
+    def test_uncompressed_under_threshold_returns_full_history(
+        self, orchestrator, conversation_service, threshold_checker
+    ):
+        conv = _compressed_conversation()
+        conv["compression_metadata"] = {}
+        conversation_service.get_conversation.return_value = conv
+        threshold_checker.should_compress.return_value = False
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.compressed_summary is None
+        assert len(result.recent_queries) == 4
+        assert result.last_compression_at is None
+
+    @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_over_threshold_compresses_only_the_tail(
+        self,
+        mock_settings,
+        MockCompressionService,
+        MockLLMCreator,
+        _key,
+        _provider,
+        orchestrator,
+        conversation_service,
+        threshold_checker,
+    ):
+        mock_settings.COMPRESSION_MODEL_OVERRIDE = None
+        conversation = _compressed_conversation()
+        conversation_service.get_conversation.return_value = conversation
+        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-03T10: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.success and result.compression_performed
+        args, kwargs = svc.compress_and_save.call_args
+        # queries 0-1 are already inside point 1; only 2-3 are new.
+        start_index = kwargs.get("start_index", args[3] if len(args) > 3 else 0)
+        assert start_index == 2
+        assert (kwargs.get("compress_up_to_index") or args[2]) == 3
+
+
+@pytest.mark.unit
+class TestEffectiveTokenCount:
+    def test_counts_summary_plus_recent_only(self):
+        conv = _compressed_conversation()
+        effective = TokenCounter.count_effective_conversation_tokens(conv)
+        raw = TokenCounter.count_conversation_tokens(conv)
+        expected = TokenCounter.count_message_tokens([{"content": "S"}]) + (
+            TokenCounter.count_query_tokens(conv["queries"][2:])
+        )
+        assert effective == expected
+        assert effective < raw
+
+    def test_uncompressed_conversation_counts_everything(self):
+        conv = _compressed_conversation()
+        conv["compression_metadata"] = None
+        assert TokenCounter.count_effective_conversation_tokens(conv) == (
+            TokenCounter.count_conversation_tokens(conv)
+        )
+
+    @patch(
+        "application.api.answer.services.compression.threshold_checker.get_token_limit",
+        return_value=1000,
+    )
+    def test_should_compress_uses_effective_count(self, _limit):
+        checker = CompressionThresholdChecker(threshold_percentage=0.8)
+        conv = _compressed_conversation()
+        # Raw history is ~1.2k tokens (over 800); summary + tail is tiny.
+        assert TokenCounter.count_conversation_tokens(conv) > 800
+        assert checker.should_compress(conv, "m", current_query_tokens=10) is False
+
+
+@pytest.mark.unit
+class TestServiceIncremental:
+    def _service(self, summary_text="<summary>new</summary>"):
+        llm = MagicMock()
+        llm.gen.return_value = summary_text
+        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"}]
+        return svc
+
+    def test_compress_conversation_tail_only(self):
+        svc = self._service()
+        conv = _compressed_conversation()
+
+        metadata = svc.compress_conversation(conv, compress_up_to_index=3, start_index=2)
+
+        queries, existing = svc.prompt_builder.build_prompt.call_args[0]
+        assert [q["prompt"] for q in queries] == ["q2", "q3"]
+        assert existing == [POINT]
+        assert metadata.query_index == 3
+        assert metadata.compressed_summary == "new"
+
+    def test_nothing_new_since_last_point_is_rejected(self):
+        svc = self._service()
+        with pytest.raises(ValueError):
+            svc.compress_conversation(
+                _compressed_conversation(), compress_up_to_index=1, start_index=2
+            )
+
+    def test_empty_summary_raises(self):
+        svc = self._service("<summary>   </summary>")
+        with pytest.raises(ValueError):
+            svc.compress_conversation(_compressed_conversation(), compress_up_to_index=3)
+
+    def test_get_compressed_context_skips_summary_rows(self):
+        svc = CompressionService(llm=None, model_id="m")
+        conv = _compressed_conversation()
+        conv["queries"].insert(2, {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"})
+        summary, recent = svc.get_compressed_context(conv)
+        assert summary == "S"
+        assert [q["prompt"] for q in recent] == ["q2", "q3"]
+
+
+@pytest.mark.unit
+class TestHistoryHelpers:
+    def test_as_history_skips_summary_rows(self):
+        result = CompressionResult.success_no_compression(
+            [
+                {"prompt": "q", "response": "r"},
+                {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"},
+            ]
+        )
+        assert [h["prompt"] for h in result.as_history()] == ["q"]
+
+    def test_success_from_existing(self):
+        result = CompressionResult.success_from_existing(
+            "S", [{"prompt": "q", "response": "r"}], last_compression_at=EPOCH
+        )
+        assert result.success and not result.compression_performed
+        assert result.compressed_summary == "S"
+        assert result.last_compression_at == EPOCH
diff --git a/tests/api/answer/services/compression/test_service.py b/tests/api/answer/services/compression/test_service.py
index 6145a168..8265b6cc 100644
--- a/tests/api/answer/services/compression/test_service.py
+++ b/tests/api/answer/services/compression/test_service.py
@@ -157,7 +157,7 @@ class TestCompressConversation:
                 {"query_index": 0, "compressed_summary": "Previous summary"}
             ]
 
-    def test_zero_compressed_tokens_ratio(self, mock_llm, sample_conversation):
+    def test_zero_compressed_tokens_is_rejected(self, mock_llm, sample_conversation):
         mock_builder = MagicMock()
         mock_builder.build_prompt.return_value = [
             {"role": "system", "content": "C"},
@@ -175,8 +175,10 @@ class TestCompressConversation:
             MockTC.count_query_tokens.return_value = 1000
             MockTC.count_message_tokens.return_value = 0
 
-            result = svc.compress_conversation(sample_conversation, 2)
-            assert result.compression_ratio == 0
+            # An empty summary must never become a compression point (it
+            # replaced a 494k-token conversation with nothing in prod).
+            with pytest.raises(ValueError, match="empty summary"):
+                svc.compress_conversation(sample_conversation, 2)
 
     def test_llm_error_propagates(self, sample_conversation):
         llm = MagicMock()
diff --git a/tests/llm/handlers/test_mid_execution_chain.py b/tests/llm/handlers/test_mid_execution_chain.py
new file mode 100644
index 00000000..eedfa796
--- /dev/null
+++ b/tests/llm/handlers/test_mid_execution_chain.py
@@ -0,0 +1,109 @@
+"""Mid-execution compression must persist once, break the Responses chain,
+and treat an empty summary as a failure.
+
+Background (prod 2026-09-01): the DB path wrote the compression point and the
+summary row, then left ``compression_saved`` False so the route wrote both
+again; and the rebuilt four-message input was still sent with
+``previous_response_id``, so Azure prepended the uncompressed transcript
+(236k input tokens for four local messages).
+"""
+
+from typing import Any, Dict, Generator
+from unittest.mock import Mock, patch
+
+import pytest
+
+from application.llm.handlers.base import LLMHandler, LLMResponse, ToolCall
+
+
+class _Handler(LLMHandler):
+    def parse_response(self, response: Any) -> LLMResponse:
+        return LLMResponse(
+            content=str(response), tool_calls=[], finish_reason="stop", raw_response=response
+        )
+
+    def create_tool_message(self, tool_call: ToolCall, result: Any) -> Dict:
+        return {"role": "tool", "content": str(result), "tool_call_id": tool_call.id}
+
+    def _iterate_stream(self, response: Any) -> Generator:
+        yield from response
+
+
+def _agent():
+    agent = Mock()
+    agent.conversation_id = "conv1"
+    agent.initial_user_id = "user1"
+    agent.model_id = "gpt-5.6"
+    agent.decoded_token = {}
+    agent.context_limit_reached = True
+    agent.current_token_count = 999
+    agent.llm = Mock()
+    return agent
+
+
+def _result(compressed=100, original=1000, summary="summary"):
+    metadata = Mock()
+    metadata.compressed_token_count = compressed
+    metadata.original_token_count = original
+    metadata.compression_ratio = original / max(compressed, 1)
+    metadata.timestamp = "2026-09-03T10:00:00+00:00"
+    metadata.to_dict.return_value = {"timestamp": metadata.timestamp}
+    result = Mock()
+    result.success = True
+    result.compression_performed = True
+    result.compressed_summary = summary
+    result.recent_queries = []
+    result.metadata = metadata
+    result.error = None
+    return result
+
+
+def _run(handler, agent, result):
+    conv_service = Mock()
+    conv_service.get_conversation.return_value = {"queries": []}
+    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={"queries": []}
+    ), patch.object(
+        handler,
+        "_rebuild_messages_after_compression",
+        return_value=[{"role": "system", "content": "rebuilt"}],
+    ), patch.object(handler, "_prune_messages_minimal", return_value=None):
+        return handler._perform_mid_execution_compression(
+            agent, [{"role": "user", "content": "hi"}]
+        ), conv_service
+
+
+@pytest.mark.unit
+def test_db_path_persists_once_and_breaks_chain():
+    agent = _agent()
+    (success, messages), conv_service = _run(_Handler(), agent, _result())
+
+    assert success is True and messages
+    # The orchestrator's compress_and_save already wrote the point; the
+    # handler wrote the visible row. The route must not write either again.
+    conv_service.append_compression_message.assert_called_once()
+    assert agent.compression_saved is True
+    # The rebuilt messages are the whole context now: forget the chain so
+    # the next call is not prepended with the uncompressed transcript.
+    agent.llm.start_responses_turn.assert_called_once()
+    assert agent.last_compression_at == "2026-09-03T10:00:00+00:00"
+    assert agent.context_limit_reached is False
+
+
+@pytest.mark.unit
+def test_empty_summary_is_a_failure_not_a_success():
+    agent = _agent()
+    (success, messages), conv_service = _run(
+        _Handler(), agent, _result(compressed=0, original=1000, summary="")
+    )
+    assert success is False and messages is None
+    conv_service.append_compression_message.assert_not_called()
+    agent.llm.start_responses_turn.assert_not_called()
diff --git a/tests/llm/test_base.py b/tests/llm/test_base.py
index 67ce9254..ea02c045 100644
--- a/tests/llm/test_base.py
+++ b/tests/llm/test_base.py
@@ -1127,7 +1127,9 @@ class TestPerformMidExecutionCompressionAdditional:
         assert success is True
         assert msgs == rebuilt
         assert agent.compressed_summary == "compressed text"
-        assert agent.compression_saved is False
+        # The DB path already wrote the point and the summary row; the
+        # route must not persist them again.
+        assert agent.compression_saved is True
         assert agent.context_limit_reached is False
 
     def test_compression_not_performed_returns_false(self, monkeypatch):
@@ -1867,7 +1869,9 @@ class TestMidExecutionCompressionMetadata:
             agent, [{"role": "user", "content": "hi"}]
         )
         assert success is True
-        assert agent.compression_saved is False
+        # The DB path already wrote the point and the summary row; the
+        # route must not persist them again.
+        assert agent.compression_saved is True
         assert agent.context_limit_reached is False
         assert agent.current_token_count == 0
         mock_conv_service.append_compression_message.assert_called_once()
@@ -2231,7 +2235,9 @@ class TestPerformMidExecutionCompressionSuccess:
             )
         assert success is True
         assert result_msgs == rebuilt
-        assert agent.compression_saved is False
+        # The DB path already wrote the point and the summary row; the
+        # route must not persist them again.
+        assert agent.compression_saved is True
 
 
 @pytest.mark.unit
diff --git a/tests/llm/test_responses_chain_budget.py b/tests/llm/test_responses_chain_budget.py
new file mode 100644
index 00000000..f64b0bd3
--- /dev/null
+++ b/tests/llm/test_responses_chain_budget.py
@@ -0,0 +1,299 @@
+"""Bounded cross-turn chaining, chained system-head dedupe and prompt-cache
+hints on the OpenAI Responses path.
+
+Background (prod, 2026-09-01/03): ``previous_response_id`` chained every user
+turn onto the previous one, so Azure's stored transcript grew without bound
+(889k prompt tokens for a 37k-token local history) while every local guard
+measured the local history. Each chained round also re-sent the system
+message, which the server appends rather than dedupes, and after a
+compression the rebuilt local messages were still chained onto the
+uncompressed transcript.
+"""
+
+import types
+from unittest.mock import MagicMock
+
+import pytest
+
+from application.core.model_settings import ModelCapabilities
+
+
+def _make_llm(monkeypatch, store_responses=True, **extra_settings):
+    monkeypatch.setattr("application.llm.openai.OpenAI", MagicMock())
+    monkeypatch.setattr(
+        "application.llm.openai.StorageCreator",
+        types.SimpleNamespace(get_storage=lambda: None),
+    )
+    monkeypatch.setattr(
+        "application.llm.openai.settings",
+        types.SimpleNamespace(
+            OPENAI_API_KEY="k",
+            API_KEY="k",
+            OPENAI_BASE_URL="",
+            AZURE_DEPLOYMENT_NAME="dep",
+            OPENAI_RESPONSES_STORE=store_responses,
+            OPENAI_REASONING_SUMMARY="auto",
+            **extra_settings,
+        ),
+    )
+    from application.llm.openai import OpenAILLM
+
+    llm = OpenAILLM(api_key="k")
+    llm.capabilities = ModelCapabilities(
+        supports_tools=True, supports_structured_output=True, api_flavor="responses"
+    )
+    return llm
+
+
+def _messages(system="sys"):
+    return [
+        {"role": "system", "content": system},
+        {"role": "user", "content": "q1"},
+        {"role": "assistant", "content": "a1"},
+        {"role": "user", "content": "q2"},
+    ]
+
+
+def _roles(items):
+    return [i.get("role") for i in items if isinstance(i, dict) and i.get("role")]
+
+
+# ── chained system head ──────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+def test_chained_input_omits_unchanged_system_head(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    # The unchained send is what puts the system message into the stored
+    # transcript; the chained follow-up must not append a second copy.
+    first, prev = llm._build_responses_input(_messages(), None)
+    assert prev is None
+    assert "system" in _roles(first)
+
+    chained, prev = llm._build_responses_input(_messages(), "resp_1")
+    assert prev == "resp_1"
+    assert _roles(chained) == ["user"]
+
+
+@pytest.mark.unit
+def test_chained_input_resends_changed_system_head(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages("sys v1"), None)
+
+    chained, _ = llm._build_responses_input(_messages("sys v2"), "resp_1")
+    assert _roles(chained) == ["system", "user"]
+
+    # ...and the new head becomes the one the chain holds.
+    again, _ = llm._build_responses_input(_messages("sys v2"), "resp_2")
+    assert _roles(again) == ["user"]
+
+
+@pytest.mark.unit
+def test_chained_system_head_hash_roundtrips_through_state(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages(), None)
+    state = llm.export_responses_state()
+    assert state.get("system_hash")
+
+    resumed = _make_llm(monkeypatch)
+    assert resumed.import_responses_state(state) is True
+    chained, _ = resumed._build_responses_input(_messages(), "resp_1")
+    assert _roles(chained) == ["user"]
+
+
+@pytest.mark.unit
+def test_start_responses_turn_forgets_system_head(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages(), None)
+    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.
+    assert _roles(chained) == ["system", "user"]
+
+
+@pytest.mark.unit
+def test_unchained_input_always_carries_system(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages(), None)
+    again, prev = llm._build_responses_input(_messages(), None)
+    assert prev is None
+    assert "system" in _roles(again)
+
+
+# ── request params: truncation + cache hints ────────────────────────────────
+
+
+def _params(llm, **kwargs):
+    return llm._build_responses_params(
+        "gpt-5.6", [{"role": "user", "content": []}], tools=None,
+        response_format=None, previous_response_id=None, stream=True,
+        kwargs=kwargs,
+    )
+
+
+@pytest.mark.unit
+def test_build_responses_params_defaults_omit_truncation_and_cache_hints(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._prompt_cache_key = "conv-123"
+    params = _params(llm)
+    assert "truncation" not in params
+    assert "prompt_cache_key" not in params
+    assert "prompt_cache_retention" not in params
+
+
+@pytest.mark.unit
+def test_build_responses_params_truncation_and_cache_hints(monkeypatch):
+    llm = _make_llm(
+        monkeypatch,
+        OPENAI_RESPONSES_TRUNCATION_AUTO=True,
+        OPENAI_PROMPT_CACHE_KEY=True,
+        OPENAI_PROMPT_CACHE_RETENTION="24h",
+    )
+    llm._prompt_cache_key = "conv-123"
+    params = _params(llm)
+    assert params["truncation"] == "auto"
+    assert params["prompt_cache_key"] == "conv-123"
+    assert params["prompt_cache_retention"] == "24h"
+
+
+@pytest.mark.unit
+def test_build_responses_params_cache_key_needs_a_conversation(monkeypatch):
+    llm = _make_llm(monkeypatch, OPENAI_PROMPT_CACHE_KEY=True)
+    llm._prompt_cache_key = None
+    assert "prompt_cache_key" not in _params(llm)
+
+
+# ── agent: when does the next turn chain? ───────────────────────────────────
+
+
+def _agent(monkeypatch, history, last_compression_at=None, **overrides):
+    from application.agents import base as base_mod
+    from application.agents.base import BaseAgent
+
+    class _Agent(BaseAgent):
+        def _gen_inner(self, query, log_context):
+            yield from ()
+
+    agent = _Agent.__new__(_Agent)
+    agent.chat_history = history
+    agent.llm = types.SimpleNamespace(
+        responses_chain_key=lambda: "key",
+        _uses_responses_api=lambda: True,
+    )
+    agent.model_id = "m"
+    agent.model_user_id = None
+    agent.user = "u"
+    agent.last_compression_at = last_compression_at
+    defaults = {
+        "OPENAI_RESPONSES_STORE": True,
+        "OPENAI_RESPONSES_CHAIN_ACROSS_TURNS": True,
+        "OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS": None,
+    }
+    defaults.update(overrides)
+    for key, value in defaults.items():
+        monkeypatch.setattr(base_mod.settings, key, value, raising=False)
+    monkeypatch.setattr(
+        "application.core.model_utils.get_token_limit", lambda *a, **k: 1000
+    )
+    return agent
+
+
+def _turn(prompt_tokens, epoch=None, rid="resp_1"):
+    meta = {
+        "response_id": rid,
+        "response_chain_key": "key",
+        "usage": {"prompt_tokens": prompt_tokens},
+    }
+    if epoch:
+        meta["compression_epoch"] = epoch
+    return {"prompt": "q", "response": "a", "metadata": meta}
+
+
+@pytest.mark.unit
+def test_previous_response_id_chains_within_budget(monkeypatch):
+    agent = _agent(monkeypatch, [_turn(500)])
+    assert agent._previous_response_id() == "resp_1"
+
+
+@pytest.mark.unit
+def test_previous_response_id_stops_at_model_window(monkeypatch):
+    # Last turn already cost the whole window: start this one from the
+    # bounded local history instead of growing Azure's transcript further.
+    agent = _agent(monkeypatch, [_turn(1000)])
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_previous_response_id_honours_explicit_budget(monkeypatch):
+    agent = _agent(
+        monkeypatch, [_turn(500)], OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS=400
+    )
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_previous_response_id_chains_when_usage_unknown(monkeypatch):
+    turn = _turn(0)
+    turn["metadata"].pop("usage")
+    agent = _agent(monkeypatch, [turn])
+    assert agent._previous_response_id() == "resp_1"
+
+
+@pytest.mark.unit
+def test_previous_response_id_kill_switch(monkeypatch):
+    agent = _agent(
+        monkeypatch, [_turn(10)], OPENAI_RESPONSES_CHAIN_ACROSS_TURNS=False
+    )
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_previous_response_id_breaks_after_newer_compression(monkeypatch):
+    agent = _agent(
+        monkeypatch,
+        [_turn(10, epoch="2026-09-03T09:00:00+00:00")],
+        last_compression_at="2026-09-03T10:00:00+00:00",
+    )
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_previous_response_id_chains_when_turn_saw_the_compression(monkeypatch):
+    epoch = "2026-09-03T10:00:00+00:00"
+    agent = _agent(monkeypatch, [_turn(10, epoch=epoch)], last_compression_at=epoch)
+    assert agent._previous_response_id() == "resp_1"
+
+
+@pytest.mark.unit
+def test_previous_response_id_compression_epoch_format_tolerant(monkeypatch):
+    # JSONB round-trips give "2026-09-03 10:00:00.123+00:00"; the point
+    # itself is written as an ISO datetime. Both must compare equal.
+    agent = _agent(
+        monkeypatch,
+        [_turn(10, epoch="2026-09-03T10:00:00.123000+00:00")],
+        last_compression_at="2026-09-03 10:00:00.123+00:00",
+    )
+    assert agent._previous_response_id() == "resp_1"
+
+
+@pytest.mark.unit
+def test_previous_response_id_breaks_when_turn_predates_any_compression(monkeypatch):
+    agent = _agent(
+        monkeypatch, [_turn(10)], last_compression_at="2026-09-03T10:00:00+00:00"
+    )
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_emit_responses_metadata_records_compression_epoch(monkeypatch):
+    agent = _agent(monkeypatch, [], last_compression_at="2026-09-03T10:00:00+00:00")
+    agent.llm = types.SimpleNamespace(
+        responses_chain_key=lambda: "key",
+        _uses_responses_api=lambda: True,
+        _last_response_id="resp_9",
+        _last_usage={"prompt_tokens": 1},
+        export_responses_state=lambda: {"chain_key": "key"},
+    )
+    events = list(agent._emit_responses_metadata())
+    assert events and events[0]["metadata"]["compression_epoch"] == "2026-09-03T10:00:00+00:00"
+    assert events[0]["metadata"]["response_id"] == "resp_9"

From 1f86139b9cffe6633da06578e8cc02e36961d860 Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Sat, 5 Sep 2026 09:30:13 +0100
Subject: [PATCH 2/6] =?UTF-8?q?fix(compression,llm):=20address=20review=20?=
 =?UTF-8?q?=E2=80=94=20exact=20point=20dedupe,=20marked=20summary=20rows,?=
 =?UTF-8?q?=20opaque=20cache=20key?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- append_compression_point only skips a point when both query_index and
  compressed_summary are present and match the last one; points without
  those fields (as in the repository tests) were all being treated as
  duplicates.
- The incremental compression tail and the orchestrator's "anything new
  since the last point" check exclude the visible summary row, which the
  prompt already receives through existing_compressions.
- Summary rows carry a persisted metadata marker; replay filters on the
  marker, and falls back to the label only for rows written before it that
  have no tool calls and no per-turn metadata, so a user who types the
  label text keeps their turn.
- The prompt_cache_key is a hash of the user id, never the id itself.
- Describe truncation="auto" as dropping the oldest items.
---
 application/agents/base.py                    | 17 +++-
 .../services/compression/orchestrator.py      | 12 ++-
 .../answer/services/compression/service.py    | 14 ++-
 .../api/answer/services/compression/types.py  | 27 +++++-
 .../answer/services/conversation_service.py   |  7 +-
 application/core/settings.py                  |  7 +-
 application/llm/openai.py                     |  4 +-
 .../storage/db/repositories/conversations.py  |  2 +
 docs/content/Deploying/DocsGPT-Settings.mdx   |  4 +-
 docs/content/Models/cloud-providers.mdx       |  4 +-
 .../compression/test_compression_reuse.py     | 94 +++++++++++++++++++
 tests/llm/test_responses_chain_budget.py      | 15 +++
 .../db/repositories/test_conversations.py     | 12 +++
 13 files changed, 198 insertions(+), 21 deletions(-)

diff --git a/application/agents/base.py b/application/agents/base.py
index 7af5cfcd..328d22eb 100644
--- a/application/agents/base.py
+++ b/application/agents/base.py
@@ -1,3 +1,4 @@
+import hashlib
 import json
 import logging
 import uuid
@@ -57,6 +58,17 @@ def _parse_epoch(value: Any) -> Optional[datetime]:
     return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
 
 
+def _cache_key_for_user(user_id: Any) -> Optional[str]:
+    """Opaque, stable prompt-cache routing key for a user.
+
+    The key only needs to be the same for every call the user makes; it
+    must not carry the identifier itself to the provider.
+    """
+    if not user_id:
+        return None
+    return hashlib.sha256(str(user_id).encode("utf-8")).hexdigest()[:32]
+
+
 def _epoch_text(value: Any) -> Optional[str]:
     """Serialize a compression timestamp for message metadata."""
     if value is None:
@@ -1337,8 +1349,9 @@ class BaseAgent(ABC):
             # Route a user's calls to the same cache shard. Keyed by user, not
             # conversation: a new conversation has no id until its first turn
             # is saved, and a key that appears on turn two would look up a
-            # different shard from the one turn one populated.
-            self.llm._prompt_cache_key = (
+            # different shard from the one turn one populated. Hashed so the
+            # identifier itself never reaches the provider.
+            self.llm._prompt_cache_key = _cache_key_for_user(
                 getattr(self, "initial_user_id", None) or getattr(self, "user", None)
             )
 
diff --git a/application/api/answer/services/compression/orchestrator.py b/application/api/answer/services/compression/orchestrator.py
index c23e7e43..ad117559 100644
--- a/application/api/answer/services/compression/orchestrator.py
+++ b/application/api/answer/services/compression/orchestrator.py
@@ -7,7 +7,10 @@ from application.api.answer.services.compression.service import CompressionServi
 from application.api.answer.services.compression.threshold_checker import (
     CompressionThresholdChecker,
 )
-from application.api.answer.services.compression.types import CompressionResult
+from application.api.answer.services.compression.types import (
+    CompressionResult,
+    is_compression_summary_row,
+)
 from application.core.model_utils import (
     get_api_key_for_provider,
     get_provider_from_model_id,
@@ -214,7 +217,12 @@ class CompressionOrchestrator:
                     start_index = int(points[-1].get("query_index", -1)) + 1
                 except (TypeError, ValueError):
                     start_index = 0
-            if start_index > compress_up_to:
+            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:
                 logger.info(
                     f"No new queries since the last compression point for "
                     f"conversation {conversation_id}; reusing its summary"
diff --git a/application/api/answer/services/compression/service.py b/application/api/answer/services/compression/service.py
index 0b51e401..ef1f494d 100644
--- a/application/api/answer/services/compression/service.py
+++ b/application/api/answer/services/compression/service.py
@@ -82,8 +82,18 @@ class CompressionService:
                     f"compress_up_to_index {compress_up_to_index}"
                 )
             # Only the queries after the previous compression point are new;
-            # the earlier ones are already inside that point's summary.
-            queries_to_compress = queries[start_index : compress_up_to_index + 1]
+            # the earlier ones are already inside that point's summary, and so
+            # is the visible summary row that follows the point.
+            queries_to_compress = [
+                q
+                for q in queries[start_index : compress_up_to_index + 1]
+                if not is_compression_summary_row(q)
+            ]
+            if not queries_to_compress:
+                raise ValueError(
+                    "Nothing to compress: no new queries since the last "
+                    "compression point"
+                )
 
             # Check if there are existing compressions. ``compression_metadata``
             # is a nullable JSONB column, so a never-compressed conversation
diff --git a/application/api/answer/services/compression/types.py b/application/api/answer/services/compression/types.py
index ff0ff4f3..b7c20ba8 100644
--- a/application/api/answer/services/compression/types.py
+++ b/application/api/answer/services/compression/types.py
@@ -4,15 +4,32 @@ from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
-# ``prompt`` of the visible summary row appended after a compression. These
-# rows are display-only: their content already reaches the model through the
-# system prompt (``compressed_summary``), so history replay skips them.
+# The visible summary row appended after a compression. These rows are
+# display-only: their content already reaches the model through the system
+# prompt (``compressed_summary``), so history replay skips them. New rows carry
+# ``metadata[COMPRESSION_SUMMARY_MARKER]``; the prompt text is the display label.
 COMPRESSION_SUMMARY_PROMPT = "[Context Compression Summary]"
+COMPRESSION_SUMMARY_MARKER = "compression_summary"
 
 
 def is_compression_summary_row(query: Any) -> bool:
-    """True for the visible summary row written by ``append_compression_message``."""
-    return isinstance(query, dict) and query.get("prompt") == COMPRESSION_SUMMARY_PROMPT
+    """True for the visible summary row written by ``append_compression_message``.
+
+    Rows written since the marker exists are recognised by it alone. Rows
+    written before it are recognised by the label only when nothing marks them
+    as a real turn (no tool calls, no per-turn metadata), so a user who types
+    the label text as a question keeps that turn in their history.
+    """
+    if not isinstance(query, dict):
+        return False
+    metadata = query.get("metadata")
+    if isinstance(metadata, dict) and metadata.get(COMPRESSION_SUMMARY_MARKER) is True:
+        return True
+    return (
+        query.get("prompt") == COMPRESSION_SUMMARY_PROMPT
+        and not query.get("tool_calls")
+        and not metadata
+    )
 
 
 @dataclass
diff --git a/application/api/answer/services/conversation_service.py b/application/api/answer/services/conversation_service.py
index f1522240..a80ca036 100644
--- a/application/api/answer/services/conversation_service.py
+++ b/application/api/answer/services/conversation_service.py
@@ -14,7 +14,10 @@ from typing import Any, Dict, List, Optional
 from sqlalchemy import text as sql_text
 
 from application.core.settings import settings
-from application.api.answer.services.compression.types import COMPRESSION_SUMMARY_PROMPT
+from application.api.answer.services.compression.types import (
+    COMPRESSION_SUMMARY_MARKER,
+    COMPRESSION_SUMMARY_PROMPT,
+)
 from application.storage.db.base_repository import looks_like_uuid
 from application.storage.db.repositories.agents import AgentsRepository
 from application.storage.db.repositories.conversations import (
@@ -605,6 +608,8 @@ class ConversationService:
                 )
                 repo.append_message(conv_pg_id, {
                     "prompt": COMPRESSION_SUMMARY_PROMPT,
+                    # Durable marker: replay filters on this, not on the label.
+                    "metadata": {COMPRESSION_SUMMARY_MARKER: True},
                     "response": summary,
                     "thought": "",
                     "sources": [],
diff --git a/application/core/settings.py b/application/core/settings.py
index 56cea673..e2eb1b60 100644
--- a/application/core/settings.py
+++ b/application/core/settings.py
@@ -377,11 +377,12 @@ class Settings(BaseSettings):
     # compressed after that turn was produced.
     OPENAI_RESPONSES_CHAIN_ACROSS_TURNS: bool = True
     OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS: Optional[int] = None
-    # ``truncation: "auto"`` lets the provider drop middle items instead of
-    # failing every request once a chain exceeds the model's native window.
+    # ``truncation: "auto"`` lets the provider drop the oldest input items
+    # instead of failing every request once a chain exceeds the model's window.
     OPENAI_RESPONSES_TRUNCATION_AUTO: bool = False
     # Prompt-cache hints on the Responses API: route a user's calls to the
-    # same cache shard, and request extended retention where offered.
+    # same cache shard (opaque per-user key), and request extended retention
+    # where offered.
     OPENAI_PROMPT_CACHE_KEY: bool = True
     OPENAI_PROMPT_CACHE_RETENTION: Optional[str] = None
     OPENAI_REASONING_SUMMARY: str = "auto"
diff --git a/application/llm/openai.py b/application/llm/openai.py
index 5e61ca16..a983ba6e 100644
--- a/application/llm/openai.py
+++ b/application/llm/openai.py
@@ -264,7 +264,7 @@ class OpenAILLM(BaseLLM):
         # transcript instead of deduping it, so a chained round only carries
         # the head when it changed.
         self._chain_system_hash = None
-        # Prompt-cache routing key (the user id), set by the agent per call.
+        # 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
         # uploaded, keyed by content hash. First-line cache for the
@@ -1293,7 +1293,7 @@ class OpenAILLM(BaseLLM):
         # server-side (store=true).
         params["include"] = ["reasoning.encrypted_content"]
         # Backstop against a chain that outgrows the model's native window:
-        # the provider drops middle items instead of failing every request.
+        # the provider drops the oldest input items instead of failing.
         if getattr(settings, "OPENAI_RESPONSES_TRUNCATION_AUTO", False):
             params["truncation"] = "auto"
         # Prompt-cache hints. The key pins a conversation to one cache shard;
diff --git a/application/storage/db/repositories/conversations.py b/application/storage/db/repositories/conversations.py
index 7105c3be..b432f988 100644
--- a/application/storage/db/repositories/conversations.py
+++ b/application/storage/db/repositories/conversations.py
@@ -517,6 +517,8 @@ class ConversationsRepository:
         ).scalar()
         if (
             isinstance(last_point, dict)
+            and point.get("query_index") is not None
+            and point.get("compressed_summary") is not None
             and last_point.get("query_index") == point.get("query_index")
             and last_point.get("compressed_summary") == point.get("compressed_summary")
         ):
diff --git a/docs/content/Deploying/DocsGPT-Settings.mdx b/docs/content/Deploying/DocsGPT-Settings.mdx
index c8368303..3779e7aa 100644
--- a/docs/content/Deploying/DocsGPT-Settings.mdx
+++ b/docs/content/Deploying/DocsGPT-Settings.mdx
@@ -499,8 +499,8 @@ See [Access Control, Roles & Teams](/Deploying/Access-Control) for the full mode
 | `OPENAI_RESPONSES_STORE` | `false` | When `true`, allows OpenAI to persist [Responses API](/Models/cloud-providers#openai-responses-api-and-reasoning) state server-side. |
 | `OPENAI_RESPONSES_CHAIN_ACROSS_TURNS` | `true` | In store mode, chain a new user turn onto the previous turn's response with `previous_response_id`. Set to `false` for stateless turns that always rebuild from the saved history. |
 | `OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS` | unset | Stop chaining across turns once the previous turn's reported prompt reached this many tokens; the turn then starts from the saved (compressible) history. Unset means the model's `context_window`. |
-| `OPENAI_RESPONSES_TRUNCATION_AUTO` | `false` | Send `truncation: "auto"` on Responses API calls so the provider drops middle items instead of rejecting a request that exceeds the model's window. |
-| `OPENAI_PROMPT_CACHE_KEY` | `true` | Send a `prompt_cache_key` (the user id) on Responses API calls so a user's requests route to the same prompt-cache shard. |
+| `OPENAI_RESPONSES_TRUNCATION_AUTO` | `false` | Send `truncation: "auto"` on Responses API calls so the provider drops the oldest conversation items instead of rejecting a request that exceeds the model's window. |
+| `OPENAI_PROMPT_CACHE_KEY` | `true` | Send an opaque per-user `prompt_cache_key` (a hash, never the user id itself) on Responses API calls so a user's requests route to the same prompt-cache shard. |
 | `OPENAI_PROMPT_CACHE_RETENTION` | unset | Value for `prompt_cache_retention` on Responses API calls, for example `24h`, where the provider offers extended prompt-cache retention. |
 | `V1_SESSION_TTL_SECONDS` | `86400` | Redis correlation lifetime, in seconds, for OpenAI-compatible client sessions. Raw external session IDs are hashed and never stored. |
 | `TITLE_MODEL_ID` | unset | Optional model ID used for asynchronous first-party conversation titles. When unset, the answer model is used. Hidden API conversations never invoke title generation. |
diff --git a/docs/content/Models/cloud-providers.mdx b/docs/content/Models/cloud-providers.mdx
index e230bef3..b9f07ce2 100644
--- a/docs/content/Models/cloud-providers.mdx
+++ b/docs/content/Models/cloud-providers.mdx
@@ -67,8 +67,8 @@ For OpenAI models that support it, DocsGPT can call the newer **Responses API**
 - `reasoning_effort` — per-model reasoning effort hint (for example `medium`) declared in the model catalog.
 - `OPENAI_RESPONSES_STORE` (default `false`) — when `true`, lets OpenAI persist Responses API state server-side and chains calls with `previous_response_id`. When `false`, DocsGPT requests encrypted reasoning items and persists those ciphertext items itself so reasoning survives tool calls and later conversation turns without OpenAI retaining the response.
 - `OPENAI_RESPONSES_CHAIN_ACROSS_TURNS` (default `true`) and `OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS` (default: the model's `context_window`) — in store mode a new user turn chains onto the previous response, which keeps the provider's stored transcript and its prompt cache warm. The chain is bounded: once the previous turn's reported prompt reaches the budget, or after the conversation history has been compressed, the next turn starts from DocsGPT's own saved history instead. Chained tool rounds do not re-send an unchanged system message.
-- `OPENAI_RESPONSES_TRUNCATION_AUTO` (default `false`) — send `truncation: "auto"` so the provider trims middle items instead of failing a request that exceeds the model's window.
-- `OPENAI_PROMPT_CACHE_KEY` (default `true`) and `OPENAI_PROMPT_CACHE_RETENTION` (default unset) — prompt-cache hints on Responses API calls: a per-user `prompt_cache_key`, and a `prompt_cache_retention` value such as `24h` where the provider supports it.
+- `OPENAI_RESPONSES_TRUNCATION_AUTO` (default `false`) — send `truncation: "auto"` so the provider drops the oldest conversation items instead of failing a request that exceeds the model's window.
+- `OPENAI_PROMPT_CACHE_KEY` (default `true`) and `OPENAI_PROMPT_CACHE_RETENTION` (default unset) — prompt-cache hints on Responses API calls: an opaque per-user `prompt_cache_key` (a hash, never the user id itself), and a `prompt_cache_retention` value such as `24h` where the provider supports it.
 - `V1_SESSION_TTL_SECONDS` (default `86400`) — lifetime of the hashed Redis mapping used to associate OpenAI-compatible client session headers with hidden DocsGPT conversations. Explicit DocsGPT conversation IDs always take precedence.
 
 See [App Configuration](/Deploying/DocsGPT-Settings) for the full settings reference.
diff --git a/tests/api/answer/services/compression/test_compression_reuse.py b/tests/api/answer/services/compression/test_compression_reuse.py
index 53235811..b3124a81 100644
--- a/tests/api/answer/services/compression/test_compression_reuse.py
+++ b/tests/api/answer/services/compression/test_compression_reuse.py
@@ -244,3 +244,97 @@ class TestHistoryHelpers:
         assert result.success and not result.compression_performed
         assert result.compressed_summary == "S"
         assert result.last_compression_at == EPOCH
+
+
+@pytest.mark.unit
+class TestSummaryRowMarker:
+    """The visible summary row is recognised by its persisted marker; a user
+    who types the label text as a question keeps that turn."""
+
+    def test_marked_row_is_a_summary_row(self):
+        from application.api.answer.services.compression.types import (
+            COMPRESSION_SUMMARY_MARKER,
+            is_compression_summary_row,
+        )
+
+        row = {"prompt": "anything", "response": "S", "metadata": {COMPRESSION_SUMMARY_MARKER: True}}
+        assert is_compression_summary_row(row) is True
+
+    def test_legacy_row_without_metadata_is_a_summary_row(self):
+        from application.api.answer.services.compression.types import is_compression_summary_row
+
+        assert is_compression_summary_row({"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"}) is True
+        assert is_compression_summary_row(
+            {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S", "metadata": {}}
+        ) is True
+
+    def test_user_turn_with_the_label_text_is_kept(self):
+        from application.api.answer.services.compression.types import is_compression_summary_row
+
+        real_turn = {
+            "prompt": COMPRESSION_SUMMARY_PROMPT,
+            "response": "an answer",
+            "metadata": {"usage": {"prompt_tokens": 10}, "response_id": "resp_1"},
+        }
+        assert is_compression_summary_row(real_turn) is False
+        with_tools = {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "r", "tool_calls": [{"tool_name": "x"}]}
+        assert is_compression_summary_row(with_tools) is False
+        result = CompressionResult.success_no_compression([real_turn])
+        assert [h["prompt"] for h in result.as_history()] == [COMPRESSION_SUMMARY_PROMPT]
+
+
+@pytest.mark.unit
+class TestIncrementalTailExcludesSummaryRows:
+    def test_compress_conversation_skips_the_summary_row_in_the_tail(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 = _compressed_conversation()
+        # A mid-execution compression appends its visible row right after the point.
+        conv["queries"].insert(2, {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"})
+
+        svc.compress_conversation(conv, compress_up_to_index=4, start_index=2)
+
+        queries, existing = svc.prompt_builder.build_prompt.call_args[0]
+        assert [q["prompt"] for q in queries] == ["q2", "q3"]
+        assert existing == [POINT]
+
+    def test_only_summary_rows_since_the_point_is_rejected(self):
+        svc = CompressionService(llm=MagicMock(), model_id="m")
+        conv = _compressed_conversation()
+        conv["queries"] = conv["queries"][:2] + [{"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"}]
+        with pytest.raises(ValueError, match="Nothing to compress"):
+            svc.compress_conversation(conv, compress_up_to_index=2, start_index=2)
+
+    @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_orchestrator_reuses_summary_when_only_summary_rows_follow_the_point(
+        self, mock_settings, MockCompressionService, MockLLMCreator, _key, _provider,
+        orchestrator, conversation_service, threshold_checker,
+    ):
+        mock_settings.COMPRESSION_MODEL_OVERRIDE = None
+        conv = _compressed_conversation()
+        conv["queries"] = conv["queries"][:2] + [{"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"}]
+        conversation_service.get_conversation.return_value = conv
+        threshold_checker.should_compress.return_value = True
+        MockLLMCreator.create_llm.return_value = MagicMock()
+        svc = MagicMock()
+        svc.get_compressed_context.return_value = ("S", [])
+        MockCompressionService.return_value = svc
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.success and not result.compression_performed
+        assert result.compressed_summary == "S"
+        svc.compress_and_save.assert_not_called()
diff --git a/tests/llm/test_responses_chain_budget.py b/tests/llm/test_responses_chain_budget.py
index f64b0bd3..4cae62a7 100644
--- a/tests/llm/test_responses_chain_budget.py
+++ b/tests/llm/test_responses_chain_budget.py
@@ -297,3 +297,18 @@ def test_emit_responses_metadata_records_compression_epoch(monkeypatch):
     events = list(agent._emit_responses_metadata())
     assert events and events[0]["metadata"]["compression_epoch"] == "2026-09-03T10:00:00+00:00"
     assert events[0]["metadata"]["response_id"] == "resp_9"
+
+
+# ── cache key is opaque ─────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+def test_cache_key_for_user_is_opaque_and_stable():
+    from application.agents.base import _cache_key_for_user
+
+    key = _cache_key_for_user("user_2Vhzgd63RSgixvvbF8Z2nhtqnE9")
+    assert key and "user_2Vhzgd" not in key
+    assert len(key) == 32 and all(c in "0123456789abcdef" for c in key)
+    assert key == _cache_key_for_user("user_2Vhzgd63RSgixvvbF8Z2nhtqnE9")
+    assert key != _cache_key_for_user("someone-else")
+    assert _cache_key_for_user(None) is None
diff --git a/tests/storage/db/repositories/test_conversations.py b/tests/storage/db/repositories/test_conversations.py
index 21b1c544..0ab638da 100644
--- a/tests/storage/db/repositories/test_conversations.py
+++ b/tests/storage/db/repositories/test_conversations.py
@@ -381,6 +381,18 @@ class TestCompressionMetadata:
         points = fetched["compression_metadata"]["compression_points"]
         assert [p["summary"] for p in points] == ["p2", "p3", "p4"]
 
+    def test_append_compression_point_ignores_an_identical_point(self, pg_conn):
+        repo = _repo(pg_conn)
+        conv = repo.create("user-1", "c")
+        point = {"query_index": 3, "compressed_summary": "S", "original_token_count": 90}
+        assert repo.append_compression_point(conv["id"], point, max_points=3) is True
+        # Two persist paths used to append the same point twice.
+        assert repo.append_compression_point(conv["id"], dict(point), max_points=3) is True
+        later = {"query_index": 5, "compressed_summary": "S2", "original_token_count": 40}
+        assert repo.append_compression_point(conv["id"], later, max_points=3) is True
+        points = repo.get(conv["id"], "user-1")["compression_metadata"]["compression_points"]
+        assert [p["query_index"] for p in points] == [3, 5]
+
 
 class TestResolveAgentRef:
     """The repo must translate Mongo ObjectId-shaped ``agent_id`` values

From fdba261d113cb689671c91fa3c22c5aeb43a6ac9 Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Sat, 5 Sep 2026 09:49:22 +0100
Subject: [PATCH 3/6] fix(db): make compression-point deduplication atomic

The duplicate check for append_compression_point ran as a SELECT before
the UPDATE, so two sessions persisting the same point could both pass it.
The predicate now lives inside the UPDATE's CASE expression and is
evaluated under the row lock; a point without query_index or
compressed_summary never matches and is always appended. Adds a
two-session regression test that races the same point and expects one.
---
 .../storage/db/repositories/conversations.py  | 76 +++++++++----------
 .../db/repositories/test_conversations.py     | 40 ++++++++++
 2 files changed, 78 insertions(+), 38 deletions(-)

diff --git a/application/storage/db/repositories/conversations.py b/application/storage/db/repositories/conversations.py
index b432f988..aa01af17 100644
--- a/application/storage/db/repositories/conversations.py
+++ b/application/storage/db/repositories/conversations.py
@@ -508,49 +508,47 @@ class ConversationsRepository:
             return False
         # Idempotent on the same point: two persist paths (the mid-execution
         # handler and the route's end-of-stream save) used to both append it.
-        last_point = self._conn.execute(
-            text(
-                "SELECT compression_metadata -> 'compression_points' -> -1 "
-                "FROM conversations WHERE id = CAST(:id AS uuid)"
-            ),
-            {"id": conversation_id},
-        ).scalar()
-        if (
-            isinstance(last_point, dict)
-            and point.get("query_index") is not None
-            and point.get("compressed_summary") is not None
-            and last_point.get("query_index") == point.get("query_index")
-            and last_point.get("compressed_summary") == point.get("compressed_summary")
-        ):
-            return True
+        # The duplicate check lives inside the UPDATE so it runs under the
+        # row lock — a separate SELECT would let two sessions both pass it.
+        # A point without ``query_index``/``compressed_summary`` never matches
+        # (NULL comparisons are false), so it is always appended.
+        dup_index = point.get("query_index")
+        dup_summary = point.get("compressed_summary")
         result = self._conn.execute(
             text(
                 """
                 UPDATE conversations SET
-                    compression_metadata = jsonb_set(
-                        COALESCE(compression_metadata, '{}'::jsonb),
-                        '{compression_points}',
-                        COALESCE(
-                            (
-                                SELECT jsonb_agg(elem ORDER BY rn)
-                                FROM (
-                                    SELECT
-                                        elem,
-                                        row_number() OVER () AS rn,
-                                        count(*) OVER () AS cnt
-                                    FROM jsonb_array_elements(
-                                        COALESCE(
-                                            compression_metadata -> 'compression_points',
-                                            '[]'::jsonb
-                                        ) || jsonb_build_array(CAST(:point AS jsonb))
-                                    ) AS elem
-                                ) ranked
-                                WHERE rn > cnt - :max_points
+                    compression_metadata = CASE
+                        WHEN (compression_metadata -> 'compression_points' -> -1
+                              ->> 'query_index') = CAST(:dup_index AS text)
+                         AND (compression_metadata -> 'compression_points' -> -1
+                              ->> 'compressed_summary') = CAST(:dup_summary AS text)
+                        THEN compression_metadata
+                        ELSE jsonb_set(
+                            COALESCE(compression_metadata, '{}'::jsonb),
+                            '{compression_points}',
+                            COALESCE(
+                                (
+                                    SELECT jsonb_agg(elem ORDER BY rn)
+                                    FROM (
+                                        SELECT
+                                            elem,
+                                            row_number() OVER () AS rn,
+                                            count(*) OVER () AS cnt
+                                        FROM jsonb_array_elements(
+                                            COALESCE(
+                                                compression_metadata -> 'compression_points',
+                                                '[]'::jsonb
+                                            ) || jsonb_build_array(CAST(:point AS jsonb))
+                                        ) AS elem
+                                    ) ranked
+                                    WHERE rn > cnt - :max_points
+                                ),
+                                '[]'::jsonb
                             ),
-                            '[]'::jsonb
-                        ),
-                        true
-                    ),
+                            true
+                        )
+                    END,
                     updated_at = now()
                 WHERE id = CAST(:id AS uuid)
                 """
@@ -559,6 +557,8 @@ class ConversationsRepository:
                 "id": conversation_id,
                 "point": json.dumps(point, cls=PGNativeJSONEncoder),
                 "max_points": int(max_points),
+                "dup_index": None if dup_index is None else str(dup_index),
+                "dup_summary": None if dup_summary is None else str(dup_summary),
             },
         )
         return result.rowcount > 0
diff --git a/tests/storage/db/repositories/test_conversations.py b/tests/storage/db/repositories/test_conversations.py
index 0ab638da..5241f464 100644
--- a/tests/storage/db/repositories/test_conversations.py
+++ b/tests/storage/db/repositories/test_conversations.py
@@ -393,6 +393,46 @@ class TestCompressionMetadata:
         points = repo.get(conv["id"], "user-1")["compression_metadata"]["compression_points"]
         assert [p["query_index"] for p in points] == [3, 5]
 
+    def test_append_compression_point_dedupes_across_concurrent_sessions(self, pg_engine):
+        """The duplicate check runs inside the UPDATE, under the row lock, so
+        two sessions racing to persist the same point store it once."""
+        import threading
+        import time
+
+        with pg_engine.connect() as setup:
+            with setup.begin():
+                conv = _repo(setup).create("user-1", "c")
+        point = {"query_index": 3, "compressed_summary": "S", "original_token_count": 90}
+        results, errors = [], []
+
+        def append(hold_before_commit):
+            try:
+                with pg_engine.connect() as conn:
+                    with conn.begin():
+                        results.append(
+                            _repo(conn).append_compression_point(
+                                conv["id"], dict(point), max_points=3
+                            )
+                        )
+                        time.sleep(hold_before_commit)
+            except Exception as exc:  # pragma: no cover - surfaced by the assertion
+                errors.append(exc)
+
+        first = threading.Thread(target=append, args=(0.5,))
+        second = threading.Thread(target=append, args=(0.0,))
+        first.start()
+        time.sleep(0.1)  # second blocks on the row lock until first commits
+        second.start()
+        first.join()
+        second.join()
+
+        assert not errors
+        assert results == [True, True]
+        with pg_engine.connect() as conn:
+            fetched = _repo(conn).get(conv["id"], "user-1")
+        points = fetched["compression_metadata"]["compression_points"]
+        assert [p["query_index"] for p in points] == [3]
+
 
 class TestResolveAgentRef:
     """The repo must translate Mongo ObjectId-shaped ``agent_id`` values

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 4/6] 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"]

From 5837c300df6c3104480b700a95f0c5c3232d2389 Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Sat, 5 Sep 2026 11:07:07 +0100
Subject: [PATCH 5/6] fix(llm): record Responses chain state only for completed
 responses

The non-streaming path recorded the response id, reasoning items and the
staged system-head hash before checking the terminal status, so a
response that came back failed could become the id the next call chains
onto and commit a head the stored transcript never received. Chain state
is now recorded after the status checks; a response cut off by
max_output_tokens still records it, since its input was accepted.
---
 application/llm/openai.py                | 11 +++++-
 tests/llm/test_responses_chain_budget.py | 48 ++++++++++++++++++++++++
 2 files changed, 57 insertions(+), 2 deletions(-)

diff --git a/application/llm/openai.py b/application/llm/openai.py
index d0aa7ab9..fb0cbd70 100644
--- a/application/llm/openai.py
+++ b/application/llm/openai.py
@@ -1551,11 +1551,13 @@ class OpenAILLM(BaseLLM):
             "OpenAI Responses request completed id=%s",
             getattr(response, "id", None),
         )
-        self._record_responses_metadata(response)
         content, tool_calls, reasoning_items = self._parse_responses_output(
             response
         )
-        self._last_reasoning_items = reasoning_items
+        # Chain state (response id, head hash, reasoning items) is recorded
+        # only for a response the provider actually completed — a failed one
+        # must not become the id the next call chains onto, nor commit a head
+        # the stored transcript never received.
         details = getattr(response, "incomplete_details", None)
         incomplete_reason = getattr(details, "reason", None)
         if getattr(response, "status", None) == "incomplete":
@@ -1564,6 +1566,9 @@ class OpenAILLM(BaseLLM):
                     self._responses_status_error(response)
                     or "Responses API incomplete: unknown reason"
                 )
+            # Cut off by the output cap: the input was accepted and stored.
+            self._record_responses_metadata(response)
+            self._last_reasoning_items = reasoning_items
             self._last_finish_reason = "length"
             if tools:
                 return _RespChoice(
@@ -1574,6 +1579,8 @@ class OpenAILLM(BaseLLM):
         status_error = self._responses_status_error(response)
         if status_error:
             raise RuntimeError(status_error)
+        self._record_responses_metadata(response)
+        self._last_reasoning_items = reasoning_items
         if tools:
             self._remember_reasoning(tool_calls, reasoning_items)
             message = _RespMessage(
diff --git a/tests/llm/test_responses_chain_budget.py b/tests/llm/test_responses_chain_budget.py
index 3f1a3e7e..9314b17f 100644
--- a/tests/llm/test_responses_chain_budget.py
+++ b/tests/llm/test_responses_chain_budget.py
@@ -354,3 +354,51 @@ def test_failed_chained_request_resends_changed_head_on_retry(monkeypatch):
     _accepted(llm, "resp_2")
     after, _ = llm._build_responses_input(_messages("sys v2"), "resp_2")
     assert _roles(after) == ["user"]
+
+
+# ── a failed non-streaming response must not become chain state ─────────────
+
+
+@pytest.mark.unit
+def test_failed_response_records_no_chain_state(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    # An accepted turn with head "sys v1".
+    llm._build_responses_input(_messages("sys v1"), None)
+    _accepted(llm, "resp_ok")
+    committed = llm._chain_system_hash
+
+    failed = types.SimpleNamespace(
+        id="resp_bad", status="failed", error=types.SimpleNamespace(message="boom"),
+        incomplete_details=None, output=[], usage=None,
+    )
+    llm.client.responses.create = MagicMock(return_value=failed)
+    with pytest.raises(RuntimeError):
+        llm._responses_gen(
+            "gpt-5.6", _messages("sys v2"), tools=None, previous_response_id="resp_ok"
+        )
+
+    # Neither the failed id nor the head it carried became chain state
+    # (the in-turn id is cleared before every request; the retry chains via
+    # the caller's previous_response_id)...
+    assert llm._last_response_id != "resp_bad"
+    assert llm._chain_system_hash == committed
+    # ...so the retry chains onto resp_ok and re-sends the changed head.
+    retry, prev = llm._build_responses_input(_messages("sys v2"), "resp_ok")
+    assert prev == "resp_ok"
+    assert _roles(retry) == ["system", "user"]
+
+
+@pytest.mark.unit
+def test_output_capped_response_still_records_chain_state(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    capped = types.SimpleNamespace(
+        id="resp_len", status="incomplete",
+        incomplete_details=types.SimpleNamespace(reason="max_output_tokens"),
+        output=[types.SimpleNamespace(type="message", content=[
+            types.SimpleNamespace(type="output_text", text="partial")])],
+        usage=None, error=None,
+    )
+    llm.client.responses.create = MagicMock(return_value=capped)
+    assert llm._responses_gen("gpt-5.6", _messages(), tools=None) == "partial"
+    assert llm._last_response_id == "resp_len"
+    assert llm._chain_system_hash is not None

From 427d85d7373312adb2496161a291015aefa1ee32 Mon Sep 17 00:00:00 2001
From: Alex <a@tushynski.me>
Date: Sat, 5 Sep 2026 11:26:25 +0100
Subject: [PATCH 6/6] fix(llm): commit the staged head hash unconditionally on
 a recorded response

An unchained request with no system message staged None, and the record
step skipped the commit, so the previous head hash survived a transcript
that never received a head; a later chained request restoring that head
would have omitted it. The staged value is now committed as-is, None
included. Also pins each rejection predicate of is_usable_compression_point
with its own test.
---
 application/llm/openai.py                     | 10 +++---
 .../compression/test_compression_reuse.py     | 31 +++++++++++++++++++
 tests/llm/test_responses_chain_budget.py      | 14 +++++++++
 3 files changed, 51 insertions(+), 4 deletions(-)

diff --git a/application/llm/openai.py b/application/llm/openai.py
index fb0cbd70..4581d5da 100644
--- a/application/llm/openai.py
+++ b/application/llm/openai.py
@@ -1423,10 +1423,12 @@ class OpenAILLM(BaseLLM):
         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
+        # kept) is now the chain's head — including "no head at all", when
+        # an unchained request went out without a system message: a stale
+        # hash would let a later chained request omit a head this
+        # transcript never received.
+        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 fa11dab3..edb763c0 100644
--- a/tests/api/answer/services/compression/test_compression_reuse.py
+++ b/tests/api/answer/services/compression/test_compression_reuse.py
@@ -499,3 +499,34 @@ class TestUnusableSavedPoints:
 
         assert result.compression_performed
         assert svc.compress_and_save.call_args.kwargs["start_index"] == 2
+
+
+@pytest.mark.unit
+class TestUsablePointPredicates:
+    """Each rejection predicate of ``is_usable_compression_point`` on its own."""
+
+    def _with_later_point(self, **point):
+        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, **point}
+        )
+        return conv
+
+    def test_a_positive_count_does_not_rescue_a_blank_summary(self):
+        conv = self._with_later_point(compressed_summary="   ", compressed_token_count=12)
+        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"]
+
+    def test_a_zero_count_does_not_rescue_a_non_blank_summary(self):
+        conv = self._with_later_point(compressed_summary="newer", 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"]
+
+    def test_a_missing_count_with_a_summary_is_usable(self):
+        conv = self._with_later_point(compressed_summary="newer")
+        summary, recent = CompressionService(llm=None, model_id="m").get_compressed_context(conv)
+        assert summary == "newer"
+        assert [q["prompt"] for q in recent] == ["q4"]
diff --git a/tests/llm/test_responses_chain_budget.py b/tests/llm/test_responses_chain_budget.py
index 9314b17f..d0f3a4ac 100644
--- a/tests/llm/test_responses_chain_budget.py
+++ b/tests/llm/test_responses_chain_budget.py
@@ -402,3 +402,17 @@ def test_output_capped_response_still_records_chain_state(monkeypatch):
     assert llm._responses_gen("gpt-5.6", _messages(), tools=None) == "partial"
     assert llm._last_response_id == "resp_len"
     assert llm._chain_system_hash is not None
+
+
+@pytest.mark.unit
+def test_recorded_request_without_a_head_clears_the_committed_hash(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages("sys v1"), None)
+    _accepted(llm, "resp_1")
+    # An unchained fallback with no system message at all is recorded: the
+    # new transcript holds no head, so the old hash must not survive it.
+    llm._build_responses_input([{"role": "user", "content": "q"}], None)
+    _accepted(llm, "resp_2")
+    assert llm._chain_system_hash is None
+    chained, _ = llm._build_responses_input(_messages("sys v1"), "resp_2")
+    assert _roles(chained) == ["system", "user"]