fix(llm,compression): bound cross-turn Responses chaining and make compression stick

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

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

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

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

Measured on Azure with the same client shape as production (stateless
OpenAI client, server-side tools, PDF part): tokens billed on the sixth turn
fell from 58k to 35k, tool rounds add tens of tokens instead of ~2.8k, the
turn after a compression reused the saved summary in under two seconds
instead of re-summarising, and the round after a mid-execution compression
started from the compressed context instead of the full stored transcript.

application/agents/base.py

@@ -2,6 +2,7 @@ import json
 import logging
 import uuid
 from abc import ABC, abstractmethod
+from datetime import datetime, timezone
 from typing import Any, Dict, Generator, List, Optional
 
 from application.agents.tool_executor import (
@@ -32,6 +33,39 @@ from application.logging import build_stack_data, log_activity, LogContext
 logger = logging.getLogger(__name__)
 
 
+def _parse_epoch(value: Any) -> Optional[datetime]:
+    """Parse a compression timestamp as stored anywhere it lands.
+
+    ``compression_points[].timestamp`` is written as an ISO datetime,
+    ``compression_metadata.last_compression_at`` as ``str(datetime)`` (space
+    separator), and the per-turn ``compression_epoch`` as whichever of those
+    the agent carried. All must compare equal.
+    """
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
+    text = str(value).strip()
+    if not text:
+        return None
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def _epoch_text(value: Any) -> Optional[str]:
+    """Serialize a compression timestamp for message metadata."""
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return str(value)
+
+
 class BaseAgent(ABC):
     # Inert defaults: an instance built without __init__ still resolves these.
     _guardrail_engine = None
@@ -64,6 +98,7 @@ class BaseAgent(ABC):
         limited_request_mode: Optional[bool] = False,
         request_limit: Optional[int] = settings.DEFAULT_AGENT_LIMITS["request_limit"],
         compressed_summary: Optional[str] = None,
+        last_compression_at: Optional[Any] = None,
         llm=None,
         llm_handler=None,
         tool_executor: Optional[ToolExecutor] = None,
@@ -160,6 +195,10 @@ class BaseAgent(ABC):
         self.limited_request_mode = limited_request_mode
         self.request_limit = request_limit
         self.compressed_summary = compressed_summary
+        # When the conversation was last compressed (any path). Stamped onto
+        # each turn's metadata so a later turn never chains onto a Responses
+        # id produced before a compression.
+        self.last_compression_at = last_compression_at
         self.current_token_count = 0
         self.context_limit_reached = False
         self.conversation_id: Optional[str] = None
@@ -314,13 +353,25 @@ class BaseAgent(ABC):
             **stored_metadata,
             "responses_state": state,
             "usage": getattr(self.llm, "_last_usage", None),
+            "compression_epoch": _epoch_text(getattr(self, "last_compression_at", None)),
         }
         metadata = {key: value for key, value in metadata.items() if value is not None}
         if metadata:
             yield {"metadata": metadata}
 
     def _previous_response_id(self) -> Optional[str]:
-        """Return the immediately preceding compatible Responses API id."""
+        """Return the preceding turn's Responses id when chaining onto it is safe.
+
+        Chaining keeps the provider's stored transcript (and its prompt cache)
+        warm across turns, but that transcript is invisible to every local
+        guard, so it is bounded. No chaining across turns when the operator
+        turned it off, when the previous turn's reported prompt already
+        reached the chain budget (default: the model's context window), or
+        when the conversation was compressed after that turn was produced —
+        the compressed local history is the context then, not the server's.
+        """
+        if not getattr(settings, "OPENAI_RESPONSES_CHAIN_ACROSS_TURNS", True):
+            return None
         if not self.chat_history:
             return None
         turn = self.chat_history[-1]
@@ -333,13 +384,49 @@ class BaseAgent(ABC):
         current_chain_key = (
             chain_key_factory() if callable(chain_key_factory) else None
         )
-        if (
+        if not (
             current_chain_key
             and meta.get("response_chain_key") == current_chain_key
             and meta.get("response_id")
         ):
+            return None
+
+        current_epoch = _parse_epoch(getattr(self, "last_compression_at", None))
+        if current_epoch is not None:
+            turn_epoch = _parse_epoch(meta.get("compression_epoch"))
+            if turn_epoch is None or turn_epoch < current_epoch:
+                logger.info(
+                    "Responses chain reset: the conversation was compressed after "
+                    "the previous turn; starting from the compressed local history"
+                )
+                return None
+
+        usage = meta.get("usage") if isinstance(meta.get("usage"), dict) else {}
+        try:
+            prompt_tokens = int(usage.get("prompt_tokens") or 0)
+        except (TypeError, ValueError):
+            prompt_tokens = 0
+        if not prompt_tokens:
+            # No provider-reported usage on the previous turn (older rows,
+            # estimate-only providers): nothing to bound against.
             return meta["response_id"]
-        return None
+        budget = getattr(settings, "OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS", None)
+        if not budget:
+            from application.core.model_utils import get_token_limit
+
+            budget = get_token_limit(
+                getattr(self, "model_id", None),
+                user_id=getattr(self, "model_user_id", None) or getattr(self, "user", None),
+            )
+        if budget and prompt_tokens >= int(budget):
+            logger.info(
+                "Responses chain budget reached (%s >= %s prompt tokens on the "
+                "previous turn); starting this turn from the local history",
+                prompt_tokens,
+                budget,
+            )
+            return None
+        return meta["response_id"]
 
     def _previous_responses_state(self) -> Optional[Dict[str, Any]]:
         """Return continuity state from the immediately preceding turn."""
@@ -1238,6 +1325,22 @@ class BaseAgent(ABC):
             previous_response_id = self._previous_response_id()
             if previous_response_id:
                 gen_kwargs["previous_response_id"] = previous_response_id
+                if not preserve_responses_state and hasattr(
+                    self.llm, "_chain_system_hash"
+                ):
+                    # The system head the chain last saw, so the first
+                    # chained call of this turn skips an unchanged system
+                    # message instead of appending a copy server-side.
+                    state = self._previous_responses_state() or {}
+                    self.llm._chain_system_hash = state.get("system_hash")
+        if hasattr(self.llm, "_prompt_cache_key"):
+            # Route a user's calls to the same cache shard. Keyed by user, not
+            # conversation: a new conversation has no id until its first turn
+            # is saved, and a key that appears on turn two would look up a
+            # different shard from the one turn one populated.
+            self.llm._prompt_cache_key = (
+                getattr(self, "initial_user_id", None) or getattr(self, "user", None)
+            )
 
         # Forward OpenAI sampling params (temperature, max_tokens, top_p, ...).
         if self.llm_params:

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

@@ -1,14 +1,13 @@
 """High-level compression orchestration."""
 
 import logging
-from typing import Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, Optional
 
 from application.api.answer.services.compression.service import CompressionService
 from application.api.answer.services.compression.threshold_checker import (
     CompressionThresholdChecker,
 )
 from application.api.answer.services.compression.types import CompressionResult
-from application.api.answer.services.conversation_service import ConversationService
 from application.core.model_utils import (
     get_api_key_for_provider,
     get_provider_from_model_id,
@@ -16,6 +15,9 @@ from application.core.model_utils import (
 from application.core.settings import settings
 from application.llm.llm_creator import LLMCreator
 
+if TYPE_CHECKING:  # pragma: no cover - annotation only
+    from application.api.answer.services.conversation_service import ConversationService
+
 logger = logging.getLogger(__name__)
 
 
@@ -29,7 +31,7 @@ class CompressionOrchestrator:
 
     def __init__(
         self,
-        conversation_service: ConversationService,
+        conversation_service: "ConversationService",
         threshold_checker: Optional[CompressionThresholdChecker] = None,
     ):
         """
@@ -89,6 +91,23 @@ class CompressionOrchestrator:
                 current_query_tokens,
                 user_id=registry_user_id,
             ):
+                compression_meta = conversation.get("compression_metadata") or {}
+                points = compression_meta.get("compression_points") or []
+                if compression_meta.get("is_compressed") and points:
+                    # A saved summary applies to every later turn, not only
+                    # the turn that made it (measured in prod: the full raw
+                    # history was replayed on the very next turn).
+                    summary, recent = self._read_only_service(
+                        model_id
+                    ).get_compressed_context(conversation)
+                    return CompressionResult.success_from_existing(
+                        summary,
+                        recent,
+                        last_compression_at=(
+                            compression_meta.get("last_compression_at")
+                            or points[-1].get("timestamp")
+                        ),
+                    )
                 # No compression needed, return full history
                 queries = conversation.get("queries", [])
                 return CompressionResult.success_no_compression(queries)
@@ -109,6 +128,12 @@ class CompressionOrchestrator:
             )
             return CompressionResult.failure(str(e))
 
+    def _read_only_service(self, model_id: str) -> CompressionService:
+        """A service for applying saved compression points; no LLM is involved."""
+        return CompressionService(
+            llm=None, model_id=model_id, conversation_service=self.conversation_service
+        )
+
     def _perform_compression(
         self,
         conversation_id: str,
@@ -171,7 +196,8 @@ class CompressionOrchestrator:
                 conversation_service=self.conversation_service,
             )
 
-            # Compress all queries up to the latest
+            # Compress everything since the last compression point up to the
+            # latest query; the earlier queries are already in that point.
             queries_count = len(conversation.get("queries", []))
             compress_up_to = queries_count - 1
 
@@ -179,14 +205,41 @@ class CompressionOrchestrator:
                 logger.warning("No queries to compress")
                 return CompressionResult.success_no_compression([])
 
+            points = (conversation.get("compression_metadata") or {}).get(
+                "compression_points"
+            ) or []
+            start_index = 0
+            if points:
+                try:
+                    start_index = int(points[-1].get("query_index", -1)) + 1
+                except (TypeError, ValueError):
+                    start_index = 0
+            if start_index > compress_up_to:
+                logger.info(
+                    f"No new queries since the last compression point for "
+                    f"conversation {conversation_id}; reusing its summary"
+                )
+                summary, recent = compression_service.get_compressed_context(
+                    conversation
+                )
+                return CompressionResult.success_from_existing(
+                    summary, recent, last_compression_at=points[-1].get("timestamp")
+                )
+
             logger.info(
                 f"Initiating compression for conversation {conversation_id}: "
-                f"compressing all {queries_count} queries (0-{compress_up_to})"
+                f"compressing queries {start_index}-{compress_up_to} of "
+                f"{queries_count}"
+                + (
+                    f" on top of the summary at query {points[-1].get('query_index')}"
+                    if points
+                    else ""
+                )
             )
 
             # Perform compression and save to DB
             metadata = compression_service.compress_and_save(
-                conversation_id, conversation, compress_up_to
+                conversation_id, conversation, compress_up_to, start_index=start_index
             )
 
             logger.info(

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

@@ -11,6 +11,7 @@ from application.api.answer.services.compression.prompt_builder import (
 from application.api.answer.services.compression.token_counter import TokenCounter
 from application.api.answer.services.compression.types import (
     CompressionMetadata,
+    is_compression_summary_row,
 )
 from application.core.settings import settings
 
@@ -51,6 +52,7 @@ class CompressionService:
         self,
         conversation: Dict[str, Any],
         compress_up_to_index: int,
+        start_index: int = 0,
     ) -> CompressionMetadata:
         """
         Compress conversation history up to specified index.
@@ -74,8 +76,14 @@ class CompressionService:
                     f"(conversation has {len(queries)} queries)"
                 )
 
-            # Get queries to compress
-            queries_to_compress = queries[: compress_up_to_index + 1]
+            if start_index < 0 or start_index > compress_up_to_index:
+                raise ValueError(
+                    f"Nothing to compress: start_index {start_index} is past "
+                    f"compress_up_to_index {compress_up_to_index}"
+                )
+            # Only the queries after the previous compression point are new;
+            # the earlier ones are already inside that point's summary.
+            queries_to_compress = queries[start_index : compress_up_to_index + 1]
 
             # Check if there are existing compressions. ``compression_metadata``
             # is a nullable JSONB column, so a never-compressed conversation
@@ -85,14 +93,26 @@ class CompressionService:
                 "compression_points", []
             )
 
+            previous_summary_tokens = 0
             if existing_compressions:
+                # Each point already folds in the ones before it, so only
+                # the latest matters — and it is part of what the new
+                # summary replaces.
+                existing_compressions = [existing_compressions[-1]]
+                previous_summary_tokens = TokenCounter.count_message_tokens(
+                    [{"content": existing_compressions[0].get("compressed_summary", "")}]
+                )
                 logger.info(
-                    f"Found {len(existing_compressions)} previous compression(s) - "
-                    f"will incorporate into new summary"
+                    "Found a previous compression point (query %s) - "
+                    "the new summary builds on it",
+                    existing_compressions[0].get("query_index"),
                 )
 
-            # Calculate original token count
-            original_tokens = TokenCounter.count_query_tokens(queries_to_compress)
+            # Calculate original token count: everything the new summary replaces
+            original_tokens = (
+                TokenCounter.count_query_tokens(queries_to_compress)
+                + previous_summary_tokens
+            )
 
             # Log tool call stats
             self._log_tool_call_stats(queries_to_compress)
@@ -105,7 +125,7 @@ class CompressionService:
             # Call LLM to generate compression
             logger.info(
                 f"Starting compression: {len(queries_to_compress)} queries "
-                f"(messages 0-{compress_up_to_index}, {original_tokens} tokens) "
+                f"(messages {start_index}-{compress_up_to_index}, {original_tokens} tokens) "
                 f"using model {self.model_id}"
             )
 
@@ -126,6 +146,13 @@ class CompressionService:
                 [{"content": compressed_summary}]
             )
 
+            # An empty summary is not a compression: it replaced a 494k-token
+            # conversation with nothing in prod (2026-08-27) while reporting
+            # success.
+            if not compressed_summary.strip() or compressed_tokens <= 0:
+                raise ValueError(
+                    "Compression produced an empty summary; keeping original history"
+                )
             # Calculate compression ratio
             compression_ratio = (
                 original_tokens / compressed_tokens if compressed_tokens > 0 else 0
@@ -171,6 +198,7 @@ class CompressionService:
         conversation_id: str,
         conversation: Dict[str, Any],
         compress_up_to_index: int,
+        start_index: int = 0,
     ) -> CompressionMetadata:
         """
         Compress conversation and save to database.
@@ -192,7 +220,9 @@ class CompressionService:
             )
 
         # Perform compression
-        metadata = self.compress_conversation(conversation, compress_up_to_index)
+        metadata = self.compress_conversation(
+            conversation, compress_up_to_index, start_index=start_index
+        )
 
         # Save to database
         self.conversation_service.update_compression_metadata(
@@ -248,7 +278,13 @@ class CompressionService:
             # Get only messages after compression point
             queries = conversation.get("queries", [])
             total_queries = len(queries)
-            recent_queries = queries[last_compressed_index + 1 :]
+            # The visible summary rows appended after a compression are not
+            # history: their content already rides in the system prompt.
+            recent_queries = [
+                q
+                for q in queries[last_compressed_index + 1 :]
+                if not is_compression_summary_row(q)
+            ]
 
             logger.info(
                 f"Using compressed context: summary ({compressed_tokens} tokens, "

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,7 @@ from typing import Any, Dict, List
 
 from application.utils import num_tokens_from_string
 from application.core.settings import settings
+from application.api.answer.services.compression.types import is_compression_summary_row
 
 logger = logging.getLogger(__name__)
 
@@ -131,3 +132,36 @@ class TokenCounter:
         except Exception as e:
             logger.error(f"Error calculating conversation tokens: {str(e)}")
             return 0
+
+    @staticmethod
+    def count_effective_conversation_tokens(conversation: Dict[str, Any]) -> int:
+        """Tokens the next turn will actually replay.
+
+        The latest summary plus the queries after its compression point, or
+        everything when the conversation was never compressed.
+        ``count_conversation_tokens`` counts the raw history regardless, which
+        is what made every turn after a compression trigger it again.
+        """
+        try:
+            queries = conversation.get("queries", []) or []
+            metadata = conversation.get("compression_metadata") or {}
+            points = metadata.get("compression_points") or []
+            if not (metadata.get("is_compressed") and points):
+                return TokenCounter.count_query_tokens(queries)
+            latest = points[-1] or {}
+            try:
+                last_index = int(latest.get("query_index", -1))
+            except (TypeError, ValueError):
+                last_index = -1
+            recent = [
+                q
+                for q in queries[last_index + 1 :]
+                if not is_compression_summary_row(q)
+            ]
+            summary_tokens = TokenCounter.count_message_tokens(
+                [{"content": latest.get("compressed_summary") or ""}]
+            )
+            return summary_tokens + TokenCounter.count_query_tokens(recent)
+        except Exception as e:
+            logger.error(f"Error calculating effective conversation tokens: {str(e)}")
+            return 0

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

@@ -4,6 +4,16 @@ from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
+# ``prompt`` of the visible summary row appended after a compression. These
+# rows are display-only: their content already reaches the model through the
+# system prompt (``compressed_summary``), so history replay skips them.
+COMPRESSION_SUMMARY_PROMPT = "[Context Compression Summary]"
+
+
+def is_compression_summary_row(query: Any) -> bool:
+    """True for the visible summary row written by ``append_compression_message``."""
+    return isinstance(query, dict) and query.get("prompt") == COMPRESSION_SUMMARY_PROMPT
+
 
 @dataclass
 class CompressionMetadata:
@@ -42,10 +52,17 @@ class CompressionResult:
     metadata: Optional[CompressionMetadata] = None
     error: Optional[str] = None
     compression_performed: bool = False
+    # When the conversation was last compressed — the point this turn's
+    # history reflects, whether it was made now or reused from the DB.
+    last_compression_at: Optional[Any] = None
 
     @classmethod
     def success_with_compression(
-        cls, summary: str, queries: List[Dict], metadata: CompressionMetadata
+        cls,
+        summary: str,
+        queries: List[Dict],
+        metadata: CompressionMetadata,
+        last_compression_at: Optional[Any] = None,
     ) -> "CompressionResult":
         """Create a successful result with compression."""
         return cls(
@@ -54,6 +71,31 @@ class CompressionResult:
             recent_queries=queries,
             metadata=metadata,
             compression_performed=True,
+            last_compression_at=(
+                last_compression_at
+                if last_compression_at is not None
+                else getattr(metadata, "timestamp", None)
+            ),
+        )
+
+    @classmethod
+    def success_from_existing(
+        cls,
+        summary: Optional[str],
+        queries: List[Dict],
+        last_compression_at: Optional[Any] = None,
+    ) -> "CompressionResult":
+        """A previously saved compression point applied to this turn.
+
+        No LLM call was made; the summary and the queries after its point
+        are what the turn replays instead of the raw history.
+        """
+        return cls(
+            success=True,
+            compressed_summary=summary,
+            recent_queries=queries,
+            compression_performed=False,
+            last_compression_at=last_compression_at,
         )
 
     @classmethod
@@ -81,6 +123,8 @@ class CompressionResult:
         """
         out: List[Dict[str, str]] = []
         for q in self.recent_queries:
+            if is_compression_summary_row(q):
+                continue
             entry: Dict[str, str] = {
                 "prompt": q["prompt"],
                 "response": q["response"],

application/api/answer/services/conversation_service.py

@@ -14,6 +14,7 @@ from typing import Any, Dict, List, Optional
 from sqlalchemy import text as sql_text
 
 from application.core.settings import settings
+from application.api.answer.services.compression.types import COMPRESSION_SUMMARY_PROMPT
 from application.storage.db.base_repository import looks_like_uuid
 from application.storage.db.repositories.agents import AgentsRepository
 from application.storage.db.repositories.conversations import (
@@ -603,7 +604,7 @@ class ConversationService:
                     str(conv["id"]) if conv is not None else conversation_id
                 )
                 repo.append_message(conv_pg_id, {
-                    "prompt": "[Context Compression Summary]",
+                    "prompt": COMPRESSION_SUMMARY_PROMPT,
                     "response": summary,
                     "thought": "",
                     "sources": [],

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,21 @@ class Settings(BaseSettings):
     # True persists Responses API calls server-side so previous_response_id can chain turns.
     # False keeps them stateless, carrying reasoning across the tool loop as encrypted items.
     OPENAI_RESPONSES_STORE: bool = False
+    # Cross-turn ``previous_response_id`` chaining (store mode only). The
+    # chained transcript lives on the provider and is invisible to every
+    # local guard, so it is bounded: a turn starts from the local history
+    # when the previous turn's reported prompt already reached the budget
+    # (default: the model's context window) or when the conversation was
+    # compressed after that turn was produced.
+    OPENAI_RESPONSES_CHAIN_ACROSS_TURNS: bool = True
+    OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS: Optional[int] = None
+    # ``truncation: "auto"`` lets the provider drop middle items instead of
+    # failing every request once a chain exceeds the model's native window.
+    OPENAI_RESPONSES_TRUNCATION_AUTO: bool = False
+    # Prompt-cache hints on the Responses API: route a user's calls to the
+    # same cache shard, and request extended retention where offered.
+    OPENAI_PROMPT_CACHE_KEY: bool = True
+    OPENAI_PROMPT_CACHE_RETENTION: Optional[str] = None
     OPENAI_REASONING_SUMMARY: str = "auto"
 
     # Lets OpenAI-compatible clients identify a logical chat by session header, which

application/llm/handlers/base.py

@@ -592,6 +592,22 @@ class LLMHandler(ABC):
             include_tool_calls=include_tool_calls,
         )
 
+    @staticmethod
+    def _reset_responses_chain(agent, metadata) -> None:
+        """After a compression the rebuilt local messages ARE the context.
+
+        Forget the provider-side chain so the next call is not prepended with
+        the uncompressed transcript (prod: 236k input tokens billed for a
+        four-message rebuilt input), and stamp the compression epoch so later
+        turns do not chain onto pre-compression responses either.
+        """
+        starter = getattr(getattr(agent, "llm", None), "start_responses_turn", None)
+        if callable(starter):
+            starter()
+        timestamp = getattr(metadata, "timestamp", None)
+        if timestamp is not None:
+            agent.last_compression_at = timestamp
+
     def _perform_mid_execution_compression(
         self, agent, messages: List[Dict]
     ) -> tuple[bool, Optional[List[Dict]]]:
@@ -661,11 +677,18 @@ class LLMHandler(ABC):
                 logger.warning("Compression not performed")
                 return False, None
 
-            # Check if compression actually reduced tokens
+            # Check if compression actually reduced tokens — and produced
+            # anything at all: a 0-token "summary" replaced a 494k-token
+            # conversation with nothing in prod (2026-08-27).
             if result.metadata:
-                if result.metadata.compressed_token_count >= result.metadata.original_token_count:
+                if (
+                    result.metadata.compressed_token_count <= 0
+                    or result.metadata.compressed_token_count
+                    >= result.metadata.original_token_count
+                ):
                     logger.warning(
-                        "Compression did not reduce token count; falling back to minimal pruning"
+                        "Compression did not reduce token count (or produced an "
+                        "empty summary); falling back to minimal pruning"
                     )
                     pruned = self._prune_messages_minimal(messages)
                     if pruned:
@@ -685,10 +708,15 @@ class LLMHandler(ABC):
                     agent.conversation_id, result.metadata.to_dict()
                 )
 
-            # Update agent's compressed summary for downstream persistence
+            # Update agent's compressed summary for downstream persistence.
+            # ``compress_mid_execution`` already wrote the compression point
+            # and the row above is the visible summary, so this IS saved —
+            # leaving the flag False made the route persist both again
+            # (prod: every mid-execution compression stored twice).
             agent.compressed_summary = result.compressed_summary
             agent.compression_metadata = result.metadata.to_dict() if result.metadata else None
-            agent.compression_saved = False
+            agent.compression_saved = True
+            self._reset_responses_chain(agent, result.metadata)
 
             # Reset the context limit flag so tools can continue
             agent.context_limit_reached = False
@@ -827,7 +855,9 @@ class LLMHandler(ABC):
 
             agent.compressed_summary = compressed_summary
             agent.compression_metadata = metadata.to_dict()
+            # Nothing reached the DB on this path; the route persists once.
             agent.compression_saved = False
+            self._reset_responses_chain(agent, metadata)
             agent.context_limit_reached = False
             agent.current_token_count = 0
 

application/llm/openai.py

@@ -259,6 +259,13 @@ class OpenAILLM(BaseLLM):
         # fallback restream of the already-delivered answer.
         self._stream_reached_finish = False
         self._imported_response_id = None
+        # Hash of the system head the current Responses chain already holds.
+        # The server appends a re-sent system message after the stored
+        # transcript instead of deduping it, so a chained round only carries
+        # the head when it changed.
+        self._chain_system_hash = None
+        # Prompt-cache routing key (the user id), set by the agent per call.
+        self._prompt_cache_key = None
         # Files-API ids for inline ``file_data`` content parts already
         # uploaded, keyed by content hash. First-line cache for the
         # in-request tool loop; the Redis-backed cross-request cache
@@ -304,6 +311,7 @@ class OpenAILLM(BaseLLM):
             "call_ids": sorted(self._last_response_call_ids or ()),
             "reasoning_items": self._last_reasoning_items,
             "reasoning_for_calls": self._reasoning_for_calls,
+            "system_hash": self._chain_system_hash,
         }
 
     def import_responses_state(self, state: dict | None) -> bool:
@@ -318,6 +326,7 @@ class OpenAILLM(BaseLLM):
         self._last_response_call_ids = set(state.get("call_ids") or ())
         self._last_reasoning_items = list(state.get("reasoning_items") or [])
         self._reasoning_for_calls = dict(state.get("reasoning_for_calls") or {})
+        self._chain_system_hash = state.get("system_hash")
         return True
 
     def start_responses_turn(self) -> None:
@@ -327,6 +336,7 @@ class OpenAILLM(BaseLLM):
         self._last_response_id = None
         self._last_response_call_ids = set()
         self._imported_response_id = None
+        self._chain_system_hash = None
         self._last_finish_reason = None
 
     def _resolve_file_part(self, item):
@@ -1099,6 +1109,19 @@ class OpenAILLM(BaseLLM):
         ]
         return head + carried + messages[last_assistant + 1:]
 
+    @staticmethod
+    def _system_fingerprint(messages):
+        """Hash of the system messages' content; ``None`` when there are none."""
+        parts = [
+            m.get("content")
+            for m in messages
+            if isinstance(m, dict) and m.get("role") == "system"
+        ]
+        if not parts:
+            return None
+        encoded = json.dumps(parts, sort_keys=True, default=str).encode("utf-8")
+        return hashlib.sha256(encoded).hexdigest()
+
     def _build_responses_input(self, messages, previous_response_id):
         """Build the Responses ``input`` list, honouring store-mode chaining.
 
@@ -1119,11 +1142,21 @@ class OpenAILLM(BaseLLM):
         """
         chained = bool(previous_response_id and settings.OPENAI_RESPONSES_STORE)
         if not chained:
+            self._chain_system_hash = self._system_fingerprint(messages)
             return self._to_responses_input(messages, chained=False), None
 
-        input_items = self._to_responses_input(
-            self._trim_for_previous_response(messages), chained=True
-        )
+        trimmed = self._trim_for_previous_response(messages)
+        head_hash = self._system_fingerprint(trimmed)
+        if head_hash and head_hash == self._chain_system_hash:
+            # The server appends a re-sent system message after the stored
+            # transcript instead of replacing it (measured on Azure: 11,239
+            # vs 5,635 input tokens for the same chained turn), so an
+            # unchanged head stays home. After a compression the head also
+            # carries the summary, which made every tool round re-append it.
+            trimmed = [m for m in trimmed if m.get("role") != "system"]
+        elif head_hash:
+            self._chain_system_hash = head_hash
+        input_items = self._to_responses_input(trimmed, chained=True)
         expected = set(self._last_response_call_ids or ())
         if expected:
             kept, dropped = [], []
@@ -1156,6 +1189,7 @@ class OpenAILLM(BaseLLM):
                     "request the provider would reject",
                     sorted(missing),
                 )
+                self._chain_system_hash = self._system_fingerprint(messages)
                 return self._to_responses_input(messages, chained=False), None
         return input_items, previous_response_id
 
@@ -1258,6 +1292,18 @@ class OpenAILLM(BaseLLM):
         # carryover working whether or not the response is also retained
         # server-side (store=true).
         params["include"] = ["reasoning.encrypted_content"]
+        # Backstop against a chain that outgrows the model's native window:
+        # the provider drops middle items instead of failing every request.
+        if getattr(settings, "OPENAI_RESPONSES_TRUNCATION_AUTO", False):
+            params["truncation"] = "auto"
+        # Prompt-cache hints. The key pins a conversation to one cache shard;
+        # retention asks for the extended tier where the deployment offers it.
+        cache_key = getattr(self, "_prompt_cache_key", None)
+        if cache_key and getattr(settings, "OPENAI_PROMPT_CACHE_KEY", False):
+            params["prompt_cache_key"] = str(cache_key)
+        retention = getattr(settings, "OPENAI_PROMPT_CACHE_RETENTION", None)
+        if retention:
+            params["prompt_cache_retention"] = retention
         return params
 
     @staticmethod

application/storage/db/repositories/conversations.py

@@ -506,6 +506,21 @@ class ConversationsRepository:
         # Shape-gate: see ``set_compression_flags``.
         if not looks_like_uuid(conversation_id):
             return False
+        # Idempotent on the same point: two persist paths (the mid-execution
+        # handler and the route's end-of-stream save) used to both append it.
+        last_point = self._conn.execute(
+            text(
+                "SELECT compression_metadata -> 'compression_points' -> -1 "
+                "FROM conversations WHERE id = CAST(:id AS uuid)"
+            ),
+            {"id": conversation_id},
+        ).scalar()
+        if (
+            isinstance(last_point, dict)
+            and last_point.get("query_index") == point.get("query_index")
+            and last_point.get("compressed_summary") == point.get("compressed_summary")
+        ):
+            return True
         result = self._conn.execute(
             text(
                 """

docs/content/Deploying/DocsGPT-Settings.mdx

@@ -497,6 +497,11 @@ See [Access Control, Roles & Teams](/Deploying/Access-Control) for the full mode
 | Setting | Default | Description |
 | --- | --- | --- |
 | `OPENAI_RESPONSES_STORE` | `false` | When `true`, allows OpenAI to persist [Responses API](/Models/cloud-providers#openai-responses-api-and-reasoning) state server-side. |
+| `OPENAI_RESPONSES_CHAIN_ACROSS_TURNS` | `true` | In store mode, chain a new user turn onto the previous turn's response with `previous_response_id`. Set to `false` for stateless turns that always rebuild from the saved history. |
+| `OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS` | unset | Stop chaining across turns once the previous turn's reported prompt reached this many tokens; the turn then starts from the saved (compressible) history. Unset means the model's `context_window`. |
+| `OPENAI_RESPONSES_TRUNCATION_AUTO` | `false` | Send `truncation: "auto"` on Responses API calls so the provider drops middle items instead of rejecting a request that exceeds the model's window. |
+| `OPENAI_PROMPT_CACHE_KEY` | `true` | Send a `prompt_cache_key` (the user id) on Responses API calls so a user's requests route to the same prompt-cache shard. |
+| `OPENAI_PROMPT_CACHE_RETENTION` | unset | Value for `prompt_cache_retention` on Responses API calls, for example `24h`, where the provider offers extended prompt-cache retention. |
 | `V1_SESSION_TTL_SECONDS` | `86400` | Redis correlation lifetime, in seconds, for OpenAI-compatible client sessions. Raw external session IDs are hashed and never stored. |
 | `TITLE_MODEL_ID` | unset | Optional model ID used for asynchronous first-party conversation titles. When unset, the answer model is used. Hidden API conversations never invoke title generation. |
 

docs/content/Models/cloud-providers.mdx

@@ -66,6 +66,9 @@ For OpenAI models that support it, DocsGPT can call the newer **Responses API**
 
 - `reasoning_effort` — per-model reasoning effort hint (for example `medium`) declared in the model catalog.
 - `OPENAI_RESPONSES_STORE` (default `false`) — when `true`, lets OpenAI persist Responses API state server-side and chains calls with `previous_response_id`. When `false`, DocsGPT requests encrypted reasoning items and persists those ciphertext items itself so reasoning survives tool calls and later conversation turns without OpenAI retaining the response.
+- `OPENAI_RESPONSES_CHAIN_ACROSS_TURNS` (default `true`) and `OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS` (default: the model's `context_window`) — in store mode a new user turn chains onto the previous response, which keeps the provider's stored transcript and its prompt cache warm. The chain is bounded: once the previous turn's reported prompt reaches the budget, or after the conversation history has been compressed, the next turn starts from DocsGPT's own saved history instead. Chained tool rounds do not re-send an unchanged system message.
+- `OPENAI_RESPONSES_TRUNCATION_AUTO` (default `false`) — send `truncation: "auto"` so the provider trims middle items instead of failing a request that exceeds the model's window.
+- `OPENAI_PROMPT_CACHE_KEY` (default `true`) and `OPENAI_PROMPT_CACHE_RETENTION` (default unset) — prompt-cache hints on Responses API calls: a per-user `prompt_cache_key`, and a `prompt_cache_retention` value such as `24h` where the provider supports it.
 - `V1_SESSION_TTL_SECONDS` (default `86400`) — lifetime of the hashed Redis mapping used to associate OpenAI-compatible client session headers with hidden DocsGPT conversations. Explicit DocsGPT conversation IDs always take precedence.
 
 See [App Configuration](/Deploying/DocsGPT-Settings) for the full settings reference.

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

@@ -0,0 +1,246 @@
+"""Compression must stick: a saved summary is applied on every later turn,
+re-compression only summarises the tail, an empty summary is a failure, and
+the visible ``[Context Compression Summary]`` rows are never replayed.
+
+Background (prod + OSS reproduction, 2026-09-01/03): the turn-start path
+returned the full raw history whenever the conversation was under the
+threshold, so a summary was used exactly once; over the threshold it
+re-summarised everything from query 0 on every turn (14-24 s each); one
+conversation was "compressed" to a 0-token summary and carried on with
+nothing.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from application.api.answer.services.compression import CompressionService
+from application.api.answer.services.compression.orchestrator import (
+    CompressionOrchestrator,
+)
+from application.api.answer.services.compression.threshold_checker import (
+    CompressionThresholdChecker,
+)
+from application.api.answer.services.compression.token_counter import TokenCounter
+from application.api.answer.services.compression.types import CompressionResult
+from application.api.answer.services.conversation_service import (
+    COMPRESSION_SUMMARY_PROMPT,
+)
+
+EPOCH = "2026-09-03T09:00:00+00:00"
+POINT = {
+    "timestamp": EPOCH,
+    "query_index": 1,
+    "compressed_summary": "S",
+    "original_token_count": 900,
+    "compressed_token_count": 5,
+    "compression_ratio": 180.0,
+    "model_used": "m",
+    "compression_prompt_version": "v1.0",
+}
+
+
+def _compressed_conversation():
+    big = "word " * 600
+    return {
+        "queries": [
+            {"prompt": "q0", "response": big},
+            {"prompt": "q1", "response": big},
+            {"prompt": "q2", "response": "r2"},
+            {"prompt": "q3", "response": "r3"},
+        ],
+        "compression_metadata": {
+            "is_compressed": True,
+            "last_compression_at": EPOCH,
+            "compression_points": [POINT],
+        },
+        "agent_id": "agent-1",
+    }
+
+
+@pytest.fixture
+def conversation_service():
+    return MagicMock()
+
+
+@pytest.fixture
+def threshold_checker():
+    return MagicMock()
+
+
+@pytest.fixture
+def orchestrator(conversation_service, threshold_checker):
+    return CompressionOrchestrator(
+        conversation_service=conversation_service, threshold_checker=threshold_checker
+    )
+
+
+@pytest.mark.unit
+class TestTurnStartReuse:
+    def test_under_threshold_returns_existing_summary_and_recent(
+        self, orchestrator, conversation_service, threshold_checker
+    ):
+        conversation_service.get_conversation.return_value = _compressed_conversation()
+        threshold_checker.should_compress.return_value = False
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.success is True
+        assert result.compression_performed is False
+        assert result.compressed_summary == "S"
+        assert [q["prompt"] for q in result.recent_queries] == ["q2", "q3"]
+        assert result.last_compression_at == EPOCH
+
+    def test_uncompressed_under_threshold_returns_full_history(
+        self, orchestrator, conversation_service, threshold_checker
+    ):
+        conv = _compressed_conversation()
+        conv["compression_metadata"] = {}
+        conversation_service.get_conversation.return_value = conv
+        threshold_checker.should_compress.return_value = False
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.compressed_summary is None
+        assert len(result.recent_queries) == 4
+        assert result.last_compression_at is None
+
+    @patch(
+        "application.api.answer.services.compression.orchestrator.get_provider_from_model_id",
+        return_value="openai",
+    )
+    @patch(
+        "application.api.answer.services.compression.orchestrator.get_api_key_for_provider",
+        return_value="sk",
+    )
+    @patch("application.api.answer.services.compression.orchestrator.LLMCreator")
+    @patch("application.api.answer.services.compression.orchestrator.CompressionService")
+    @patch("application.api.answer.services.compression.orchestrator.settings")
+    def test_over_threshold_compresses_only_the_tail(
+        self,
+        mock_settings,
+        MockCompressionService,
+        MockLLMCreator,
+        _key,
+        _provider,
+        orchestrator,
+        conversation_service,
+        threshold_checker,
+    ):
+        mock_settings.COMPRESSION_MODEL_OVERRIDE = None
+        conversation = _compressed_conversation()
+        conversation_service.get_conversation.return_value = conversation
+        threshold_checker.should_compress.return_value = True
+        MockLLMCreator.create_llm.return_value = MagicMock()
+        metadata = MagicMock()
+        metadata.compression_ratio = 3.0
+        metadata.original_token_count = 30
+        metadata.compressed_token_count = 10
+        metadata.timestamp = "2026-09-03T10:00:00+00:00"
+        svc = MagicMock()
+        svc.compress_and_save.return_value = metadata
+        svc.get_compressed_context.return_value = ("S2", [])
+        MockCompressionService.return_value = svc
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.success and result.compression_performed
+        args, kwargs = svc.compress_and_save.call_args
+        # queries 0-1 are already inside point 1; only 2-3 are new.
+        start_index = kwargs.get("start_index", args[3] if len(args) > 3 else 0)
+        assert start_index == 2
+        assert (kwargs.get("compress_up_to_index") or args[2]) == 3
+
+
+@pytest.mark.unit
+class TestEffectiveTokenCount:
+    def test_counts_summary_plus_recent_only(self):
+        conv = _compressed_conversation()
+        effective = TokenCounter.count_effective_conversation_tokens(conv)
+        raw = TokenCounter.count_conversation_tokens(conv)
+        expected = TokenCounter.count_message_tokens([{"content": "S"}]) + (
+            TokenCounter.count_query_tokens(conv["queries"][2:])
+        )
+        assert effective == expected
+        assert effective < raw
+
+    def test_uncompressed_conversation_counts_everything(self):
+        conv = _compressed_conversation()
+        conv["compression_metadata"] = None
+        assert TokenCounter.count_effective_conversation_tokens(conv) == (
+            TokenCounter.count_conversation_tokens(conv)
+        )
+
+    @patch(
+        "application.api.answer.services.compression.threshold_checker.get_token_limit",
+        return_value=1000,
+    )
+    def test_should_compress_uses_effective_count(self, _limit):
+        checker = CompressionThresholdChecker(threshold_percentage=0.8)
+        conv = _compressed_conversation()
+        # Raw history is ~1.2k tokens (over 800); summary + tail is tiny.
+        assert TokenCounter.count_conversation_tokens(conv) > 800
+        assert checker.should_compress(conv, "m", current_query_tokens=10) is False
+
+
+@pytest.mark.unit
+class TestServiceIncremental:
+    def _service(self, summary_text="<summary>new</summary>"):
+        llm = MagicMock()
+        llm.gen.return_value = summary_text
+        svc = CompressionService(llm=llm, model_id="m")
+        svc.prompt_builder = MagicMock(version="v1.0")
+        svc.prompt_builder.build_prompt.return_value = [{"role": "user", "content": "p"}]
+        return svc
+
+    def test_compress_conversation_tail_only(self):
+        svc = self._service()
+        conv = _compressed_conversation()
+
+        metadata = svc.compress_conversation(conv, compress_up_to_index=3, start_index=2)
+
+        queries, existing = svc.prompt_builder.build_prompt.call_args[0]
+        assert [q["prompt"] for q in queries] == ["q2", "q3"]
+        assert existing == [POINT]
+        assert metadata.query_index == 3
+        assert metadata.compressed_summary == "new"
+
+    def test_nothing_new_since_last_point_is_rejected(self):
+        svc = self._service()
+        with pytest.raises(ValueError):
+            svc.compress_conversation(
+                _compressed_conversation(), compress_up_to_index=1, start_index=2
+            )
+
+    def test_empty_summary_raises(self):
+        svc = self._service("<summary>   </summary>")
+        with pytest.raises(ValueError):
+            svc.compress_conversation(_compressed_conversation(), compress_up_to_index=3)
+
+    def test_get_compressed_context_skips_summary_rows(self):
+        svc = CompressionService(llm=None, model_id="m")
+        conv = _compressed_conversation()
+        conv["queries"].insert(2, {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"})
+        summary, recent = svc.get_compressed_context(conv)
+        assert summary == "S"
+        assert [q["prompt"] for q in recent] == ["q2", "q3"]
+
+
+@pytest.mark.unit
+class TestHistoryHelpers:
+    def test_as_history_skips_summary_rows(self):
+        result = CompressionResult.success_no_compression(
+            [
+                {"prompt": "q", "response": "r"},
+                {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"},
+            ]
+        )
+        assert [h["prompt"] for h in result.as_history()] == ["q"]
+
+    def test_success_from_existing(self):
+        result = CompressionResult.success_from_existing(
+            "S", [{"prompt": "q", "response": "r"}], last_compression_at=EPOCH
+        )
+        assert result.success and not result.compression_performed
+        assert result.compressed_summary == "S"
+        assert result.last_compression_at == EPOCH

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,109 @@
+"""Mid-execution compression must persist once, break the Responses chain,
+and treat an empty summary as a failure.
+
+Background (prod 2026-09-01): the DB path wrote the compression point and the
+summary row, then left ``compression_saved`` False so the route wrote both
+again; and the rebuilt four-message input was still sent with
+``previous_response_id``, so Azure prepended the uncompressed transcript
+(236k input tokens for four local messages).
+"""
+
+from typing import Any, Dict, Generator
+from unittest.mock import Mock, patch
+
+import pytest
+
+from application.llm.handlers.base import LLMHandler, LLMResponse, ToolCall
+
+
+class _Handler(LLMHandler):
+    def parse_response(self, response: Any) -> LLMResponse:
+        return LLMResponse(
+            content=str(response), tool_calls=[], finish_reason="stop", raw_response=response
+        )
+
+    def create_tool_message(self, tool_call: ToolCall, result: Any) -> Dict:
+        return {"role": "tool", "content": str(result), "tool_call_id": tool_call.id}
+
+    def _iterate_stream(self, response: Any) -> Generator:
+        yield from response
+
+
+def _agent():
+    agent = Mock()
+    agent.conversation_id = "conv1"
+    agent.initial_user_id = "user1"
+    agent.model_id = "gpt-5.6"
+    agent.decoded_token = {}
+    agent.context_limit_reached = True
+    agent.current_token_count = 999
+    agent.llm = Mock()
+    return agent
+
+
+def _result(compressed=100, original=1000, summary="summary"):
+    metadata = Mock()
+    metadata.compressed_token_count = compressed
+    metadata.original_token_count = original
+    metadata.compression_ratio = original / max(compressed, 1)
+    metadata.timestamp = "2026-09-03T10:00:00+00:00"
+    metadata.to_dict.return_value = {"timestamp": metadata.timestamp}
+    result = Mock()
+    result.success = True
+    result.compression_performed = True
+    result.compressed_summary = summary
+    result.recent_queries = []
+    result.metadata = metadata
+    result.error = None
+    return result
+
+
+def _run(handler, agent, result):
+    conv_service = Mock()
+    conv_service.get_conversation.return_value = {"queries": []}
+    orchestrator = Mock()
+    orchestrator.compress_mid_execution.return_value = result
+    with patch(
+        "application.api.answer.services.compression.CompressionOrchestrator",
+        return_value=orchestrator,
+    ), patch(
+        "application.api.answer.services.conversation_service.ConversationService",
+        return_value=conv_service,
+    ), patch.object(
+        handler, "_build_conversation_from_messages", return_value={"queries": []}
+    ), patch.object(
+        handler,
+        "_rebuild_messages_after_compression",
+        return_value=[{"role": "system", "content": "rebuilt"}],
+    ), patch.object(handler, "_prune_messages_minimal", return_value=None):
+        return handler._perform_mid_execution_compression(
+            agent, [{"role": "user", "content": "hi"}]
+        ), conv_service
+
+
+@pytest.mark.unit
+def test_db_path_persists_once_and_breaks_chain():
+    agent = _agent()
+    (success, messages), conv_service = _run(_Handler(), agent, _result())
+
+    assert success is True and messages
+    # The orchestrator's compress_and_save already wrote the point; the
+    # handler wrote the visible row. The route must not write either again.
+    conv_service.append_compression_message.assert_called_once()
+    assert agent.compression_saved is True
+    # The rebuilt messages are the whole context now: forget the chain so
+    # the next call is not prepended with the uncompressed transcript.
+    agent.llm.start_responses_turn.assert_called_once()
+    assert agent.last_compression_at == "2026-09-03T10:00:00+00:00"
+    assert agent.context_limit_reached is False
+
+
+@pytest.mark.unit
+def test_empty_summary_is_a_failure_not_a_success():
+    agent = _agent()
+    (success, messages), conv_service = _run(
+        _Handler(), agent, _result(compressed=0, original=1000, summary="")
+    )
+    assert success is False and messages is None
+    conv_service.append_compression_message.assert_not_called()
+    agent.llm.start_responses_turn.assert_not_called()

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,299 @@
+"""Bounded cross-turn chaining, chained system-head dedupe and prompt-cache
+hints on the OpenAI Responses path.
+
+Background (prod, 2026-09-01/03): ``previous_response_id`` chained every user
+turn onto the previous one, so Azure's stored transcript grew without bound
+(889k prompt tokens for a 37k-token local history) while every local guard
+measured the local history. Each chained round also re-sent the system
+message, which the server appends rather than dedupes, and after a
+compression the rebuilt local messages were still chained onto the
+uncompressed transcript.
+"""
+
+import types
+from unittest.mock import MagicMock
+
+import pytest
+
+from application.core.model_settings import ModelCapabilities
+
+
+def _make_llm(monkeypatch, store_responses=True, **extra_settings):
+    monkeypatch.setattr("application.llm.openai.OpenAI", MagicMock())
+    monkeypatch.setattr(
+        "application.llm.openai.StorageCreator",
+        types.SimpleNamespace(get_storage=lambda: None),
+    )
+    monkeypatch.setattr(
+        "application.llm.openai.settings",
+        types.SimpleNamespace(
+            OPENAI_API_KEY="k",
+            API_KEY="k",
+            OPENAI_BASE_URL="",
+            AZURE_DEPLOYMENT_NAME="dep",
+            OPENAI_RESPONSES_STORE=store_responses,
+            OPENAI_REASONING_SUMMARY="auto",
+            **extra_settings,
+        ),
+    )
+    from application.llm.openai import OpenAILLM
+
+    llm = OpenAILLM(api_key="k")
+    llm.capabilities = ModelCapabilities(
+        supports_tools=True, supports_structured_output=True, api_flavor="responses"
+    )
+    return llm
+
+
+def _messages(system="sys"):
+    return [
+        {"role": "system", "content": system},
+        {"role": "user", "content": "q1"},
+        {"role": "assistant", "content": "a1"},
+        {"role": "user", "content": "q2"},
+    ]
+
+
+def _roles(items):
+    return [i.get("role") for i in items if isinstance(i, dict) and i.get("role")]
+
+
+# ── chained system head ──────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+def test_chained_input_omits_unchanged_system_head(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    # The unchained send is what puts the system message into the stored
+    # transcript; the chained follow-up must not append a second copy.
+    first, prev = llm._build_responses_input(_messages(), None)
+    assert prev is None
+    assert "system" in _roles(first)
+
+    chained, prev = llm._build_responses_input(_messages(), "resp_1")
+    assert prev == "resp_1"
+    assert _roles(chained) == ["user"]
+
+
+@pytest.mark.unit
+def test_chained_input_resends_changed_system_head(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages("sys v1"), None)
+
+    chained, _ = llm._build_responses_input(_messages("sys v2"), "resp_1")
+    assert _roles(chained) == ["system", "user"]
+
+    # ...and the new head becomes the one the chain holds.
+    again, _ = llm._build_responses_input(_messages("sys v2"), "resp_2")
+    assert _roles(again) == ["user"]
+
+
+@pytest.mark.unit
+def test_chained_system_head_hash_roundtrips_through_state(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages(), None)
+    state = llm.export_responses_state()
+    assert state.get("system_hash")
+
+    resumed = _make_llm(monkeypatch)
+    assert resumed.import_responses_state(state) is True
+    chained, _ = resumed._build_responses_input(_messages(), "resp_1")
+    assert _roles(chained) == ["user"]
+
+
+@pytest.mark.unit
+def test_start_responses_turn_forgets_system_head(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages(), None)
+    llm.start_responses_turn()
+    chained, _ = llm._build_responses_input(_messages(), "resp_1")
+    # A fresh chain has no head on the server yet, so the head is sent.
+    assert _roles(chained) == ["system", "user"]
+
+
+@pytest.mark.unit
+def test_unchained_input_always_carries_system(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._build_responses_input(_messages(), None)
+    again, prev = llm._build_responses_input(_messages(), None)
+    assert prev is None
+    assert "system" in _roles(again)
+
+
+# ── request params: truncation + cache hints ────────────────────────────────
+
+
+def _params(llm, **kwargs):
+    return llm._build_responses_params(
+        "gpt-5.6", [{"role": "user", "content": []}], tools=None,
+        response_format=None, previous_response_id=None, stream=True,
+        kwargs=kwargs,
+    )
+
+
+@pytest.mark.unit
+def test_build_responses_params_defaults_omit_truncation_and_cache_hints(monkeypatch):
+    llm = _make_llm(monkeypatch)
+    llm._prompt_cache_key = "conv-123"
+    params = _params(llm)
+    assert "truncation" not in params
+    assert "prompt_cache_key" not in params
+    assert "prompt_cache_retention" not in params
+
+
+@pytest.mark.unit
+def test_build_responses_params_truncation_and_cache_hints(monkeypatch):
+    llm = _make_llm(
+        monkeypatch,
+        OPENAI_RESPONSES_TRUNCATION_AUTO=True,
+        OPENAI_PROMPT_CACHE_KEY=True,
+        OPENAI_PROMPT_CACHE_RETENTION="24h",
+    )
+    llm._prompt_cache_key = "conv-123"
+    params = _params(llm)
+    assert params["truncation"] == "auto"
+    assert params["prompt_cache_key"] == "conv-123"
+    assert params["prompt_cache_retention"] == "24h"
+
+
+@pytest.mark.unit
+def test_build_responses_params_cache_key_needs_a_conversation(monkeypatch):
+    llm = _make_llm(monkeypatch, OPENAI_PROMPT_CACHE_KEY=True)
+    llm._prompt_cache_key = None
+    assert "prompt_cache_key" not in _params(llm)
+
+
+# ── agent: when does the next turn chain? ───────────────────────────────────
+
+
+def _agent(monkeypatch, history, last_compression_at=None, **overrides):
+    from application.agents import base as base_mod
+    from application.agents.base import BaseAgent
+
+    class _Agent(BaseAgent):
+        def _gen_inner(self, query, log_context):
+            yield from ()
+
+    agent = _Agent.__new__(_Agent)
+    agent.chat_history = history
+    agent.llm = types.SimpleNamespace(
+        responses_chain_key=lambda: "key",
+        _uses_responses_api=lambda: True,
+    )
+    agent.model_id = "m"
+    agent.model_user_id = None
+    agent.user = "u"
+    agent.last_compression_at = last_compression_at
+    defaults = {
+        "OPENAI_RESPONSES_STORE": True,
+        "OPENAI_RESPONSES_CHAIN_ACROSS_TURNS": True,
+        "OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS": None,
+    }
+    defaults.update(overrides)
+    for key, value in defaults.items():
+        monkeypatch.setattr(base_mod.settings, key, value, raising=False)
+    monkeypatch.setattr(
+        "application.core.model_utils.get_token_limit", lambda *a, **k: 1000
+    )
+    return agent
+
+
+def _turn(prompt_tokens, epoch=None, rid="resp_1"):
+    meta = {
+        "response_id": rid,
+        "response_chain_key": "key",
+        "usage": {"prompt_tokens": prompt_tokens},
+    }
+    if epoch:
+        meta["compression_epoch"] = epoch
+    return {"prompt": "q", "response": "a", "metadata": meta}
+
+
+@pytest.mark.unit
+def test_previous_response_id_chains_within_budget(monkeypatch):
+    agent = _agent(monkeypatch, [_turn(500)])
+    assert agent._previous_response_id() == "resp_1"
+
+
+@pytest.mark.unit
+def test_previous_response_id_stops_at_model_window(monkeypatch):
+    # Last turn already cost the whole window: start this one from the
+    # bounded local history instead of growing Azure's transcript further.
+    agent = _agent(monkeypatch, [_turn(1000)])
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_previous_response_id_honours_explicit_budget(monkeypatch):
+    agent = _agent(
+        monkeypatch, [_turn(500)], OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS=400
+    )
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_previous_response_id_chains_when_usage_unknown(monkeypatch):
+    turn = _turn(0)
+    turn["metadata"].pop("usage")
+    agent = _agent(monkeypatch, [turn])
+    assert agent._previous_response_id() == "resp_1"
+
+
+@pytest.mark.unit
+def test_previous_response_id_kill_switch(monkeypatch):
+    agent = _agent(
+        monkeypatch, [_turn(10)], OPENAI_RESPONSES_CHAIN_ACROSS_TURNS=False
+    )
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_previous_response_id_breaks_after_newer_compression(monkeypatch):
+    agent = _agent(
+        monkeypatch,
+        [_turn(10, epoch="2026-09-03T09:00:00+00:00")],
+        last_compression_at="2026-09-03T10:00:00+00:00",
+    )
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_previous_response_id_chains_when_turn_saw_the_compression(monkeypatch):
+    epoch = "2026-09-03T10:00:00+00:00"
+    agent = _agent(monkeypatch, [_turn(10, epoch=epoch)], last_compression_at=epoch)
+    assert agent._previous_response_id() == "resp_1"
+
+
+@pytest.mark.unit
+def test_previous_response_id_compression_epoch_format_tolerant(monkeypatch):
+    # JSONB round-trips give "2026-09-03 10:00:00.123+00:00"; the point
+    # itself is written as an ISO datetime. Both must compare equal.
+    agent = _agent(
+        monkeypatch,
+        [_turn(10, epoch="2026-09-03T10:00:00.123000+00:00")],
+        last_compression_at="2026-09-03 10:00:00.123+00:00",
+    )
+    assert agent._previous_response_id() == "resp_1"
+
+
+@pytest.mark.unit
+def test_previous_response_id_breaks_when_turn_predates_any_compression(monkeypatch):
+    agent = _agent(
+        monkeypatch, [_turn(10)], last_compression_at="2026-09-03T10:00:00+00:00"
+    )
+    assert agent._previous_response_id() is None
+
+
+@pytest.mark.unit
+def test_emit_responses_metadata_records_compression_epoch(monkeypatch):
+    agent = _agent(monkeypatch, [], last_compression_at="2026-09-03T10:00:00+00:00")
+    agent.llm = types.SimpleNamespace(
+        responses_chain_key=lambda: "key",
+        _uses_responses_api=lambda: True,
+        _last_response_id="resp_9",
+        _last_usage={"prompt_tokens": 1},
+        export_responses_state=lambda: {"chain_key": "key"},
+    )
+    events = list(agent._emit_responses_metadata())
+    assert events and events[0]["metadata"]["compression_epoch"] == "2026-09-03T10:00:00+00:00"
+    assert events[0]["metadata"]["response_id"] == "resp_9"