Merge branch 'main' of https://github.com/arc53/DocsGPT

application/agents/base.py

@@ -1,7 +1,9 @@
+import hashlib
 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 +34,50 @@ 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 _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:
+        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 +110,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 +207,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 +365,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 +396,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 +1337,23 @@ 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. 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)
+            )
 
         # Forward OpenAI sampling params (temperature, max_tokens, top_p, ...).
         if self.llm_params:

application/api/answer/services/compression/orchestrator.py

@@ -1,14 +1,17 @@
 """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.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,
     get_provider_from_model_id,
@@ -16,6 +19,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 +35,7 @@ class CompressionOrchestrator:
 
     def __init__(
         self,
-        conversation_service: ConversationService,
+        conversation_service: "ConversationService",
         threshold_checker: Optional[CompressionThresholdChecker] = None,
     ):
         """
@@ -89,6 +95,26 @@ class CompressionOrchestrator:
                 current_query_tokens,
                 user_id=registry_user_id,
             ):
+                compression_meta = conversation.get("compression_metadata") or {}
+                points = compression_meta.get("compression_points") or []
+                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). 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)
+                    return CompressionResult.success_from_existing(
+                        summary,
+                        recent,
+                        last_compression_at=(
+                            compression_meta.get("last_compression_at")
+                            or usable_point.get("timestamp")
+                        ),
+                    )
                 # No compression needed, return full history
                 queries = conversation.get("queries", [])
                 return CompressionResult.success_no_compression(queries)
@@ -109,6 +135,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,
@@ -117,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.
@@ -171,7 +204,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 +213,58 @@ class CompressionOrchestrator:
                 logger.warning("No queries to compress")
                 return CompressionResult.success_no_compression([])
 
+            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 usable_point:
+                try:
+                    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 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"
+                )
+                summary, recent = compression_service.get_compressed_context(
+                    conversation
+                )
+                return CompressionResult.success_from_existing(
+                    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 all {queries_count} queries (0-{compress_up_to})"
+                f"compressing queries {start_index}-{compress_up_to} of "
+                f"{queries_count}{on_top}"
             )
 
             # 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,
+                persist_query_index=persist_query_index,
             )
 
             logger.info(
@@ -223,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.
@@ -264,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:

application/api/answer/services/compression/service.py

@@ -11,6 +11,8 @@ 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,
+    latest_usable_compression_point,
 )
 from application.core.settings import settings
 
@@ -51,6 +53,8 @@ class CompressionService:
         self,
         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.
@@ -58,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
@@ -74,8 +84,24 @@ 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, 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
@@ -85,14 +111,27 @@ class CompressionService:
                 "compression_points", []
             )
 
+            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 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", "")}]
+                )
                 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 +144,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 +165,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
@@ -151,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,
@@ -171,6 +221,8 @@ class CompressionService:
         conversation_id: str,
         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.
@@ -192,7 +244,12 @@ 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,
+            persist_query_index=persist_query_index,
+        )
 
         # Save to database
         self.conversation_service.update_compression_metadata(
@@ -238,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)
@@ -248,7 +313,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, "

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

application/api/answer/services/compression/token_counter.py

@@ -5,6 +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,
+    latest_usable_compression_point,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -131,3 +135,40 @@ 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 = 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):
+                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

application/api/answer/services/compression/types.py

@@ -4,6 +4,63 @@ from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
+# 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_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``.
+
+    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
 class CompressionMetadata:
@@ -42,10 +99,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 +118,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 +170,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"],

application/api/answer/services/conversation_service.py

@@ -14,6 +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_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 (
@@ -603,7 +607,9 @@ 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,
+                    # Durable marker: replay filters on this, not on the label.
+                    "metadata": {COMPRESSION_SUMMARY_MARKER: True},
                     "response": summary,
                     "thought": "",
                     "sources": [],

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,

application/core/settings.py

@@ -369,6 +369,22 @@ 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 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 (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"
 
     # Lets OpenAI-compatible clients identify a logical chat by session header, which

application/llm/handlers/base.py

@@ -592,6 +592,58 @@ 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.
+
+        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]]]:
@@ -623,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"
@@ -645,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:
@@ -661,11 +725,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 +756,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
@@ -739,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
@@ -815,19 +893,25 @@ 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
             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
 

application/llm/openai.py

@@ -259,6 +259,18 @@ 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
+        # 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
         # uploaded, keyed by content hash. First-line cache for the
         # in-request tool loop; the Redis-backed cross-request cache
@@ -304,6 +316,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 +331,8 @@ 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")
+        self._pending_system_hash = None
         return True
 
     def start_responses_turn(self) -> None:
@@ -327,6 +342,8 @@ 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._pending_system_hash = None
         self._last_finish_reason = None
 
     def _resolve_file_part(self, item):
@@ -1099,6 +1116,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 +1149,27 @@ class OpenAILLM(BaseLLM):
         """
         chained = bool(previous_response_id and settings.OPENAI_RESPONSES_STORE)
         if not chained:
+            # 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
 
-        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"]
+            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:
             kept, dropped = [], []
@@ -1156,6 +1202,7 @@ class OpenAILLM(BaseLLM):
                     "request the provider would reject",
                     sorted(missing),
                 )
+                self._pending_system_hash = self._system_fingerprint(messages)
                 return self._to_responses_input(messages, chained=False), None
         return input_items, previous_response_id
 
@@ -1258,6 +1305,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 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;
+        # 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
@@ -1363,6 +1422,13 @@ 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 — 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:
@@ -1487,11 +1553,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":
@@ -1500,6 +1568,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(
@@ -1510,6 +1581,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(

application/storage/db/repositories/conversations.py

@@ -506,34 +506,49 @@ 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.
+        # 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)
                 """
@@ -542,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

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 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. |
 

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 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.

tests/api/answer/services/compression/test_compression_reuse.py

@@ -0,0 +1,532 @@
+"""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
+
+
+@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()
+
+
+@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
+
+
+@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"]

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(

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()

tests/llm/handlers/test_mid_execution_chain.py

@@ -0,0 +1,218 @@
+"""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()
+
+
+# ── 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

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

tests/llm/test_responses_chain_budget.py

@@ -0,0 +1,418 @@
+"""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")]
+
+
+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 ──────────────────────────────────────────────────────
+
+
+@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)
+    _accepted(llm)
+
+    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)
+    _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")
+    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)
+    _accepted(llm)
+    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)
+    _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.
+    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"
+
+
+# ── 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
+
+
+# ── 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"]
+
+
+# ── 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
+
+
+@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"]

tests/storage/db/repositories/test_conversations.py

@@ -381,6 +381,58 @@ 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]
+
+    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