fix(compression,llm): address review — exact point dedupe, marked summary rows, opaque cache key

- append_compression_point only skips a point when both query_index and
  compressed_summary are present and match the last one; points without
  those fields (as in the repository tests) were all being treated as
  duplicates.
- The incremental compression tail and the orchestrator's "anything new
  since the last point" check exclude the visible summary row, which the
  prompt already receives through existing_compressions.
- Summary rows carry a persisted metadata marker; replay filters on the
  marker, and falls back to the label only for rows written before it that
  have no tool calls and no per-turn metadata, so a user who types the
  label text keeps their turn.
- The prompt_cache_key is a hash of the user id, never the id itself.
- Describe truncation="auto" as dropping the oldest items.

application/agents/base.py

@@ -1,3 +1,4 @@
+import hashlib
 import json
 import logging
 import uuid
@@ -57,6 +58,17 @@ def _parse_epoch(value: Any) -> Optional[datetime]:
     return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
 
 
+def _cache_key_for_user(user_id: Any) -> Optional[str]:
+    """Opaque, stable prompt-cache routing key for a user.
+
+    The key only needs to be the same for every call the user makes; it
+    must not carry the identifier itself to the provider.
+    """
+    if not user_id:
+        return None
+    return hashlib.sha256(str(user_id).encode("utf-8")).hexdigest()[:32]
+
+
 def _epoch_text(value: Any) -> Optional[str]:
     """Serialize a compression timestamp for message metadata."""
     if value is None:
@@ -1337,8 +1349,9 @@ class BaseAgent(ABC):
             # Route a user's calls to the same cache shard. Keyed by user, not
             # conversation: a new conversation has no id until its first turn
             # is saved, and a key that appears on turn two would look up a
-            # different shard from the one turn one populated.
-            self.llm._prompt_cache_key = (
+            # different shard from the one turn one populated. Hashed so the
+            # identifier itself never reaches the provider.
+            self.llm._prompt_cache_key = _cache_key_for_user(
                 getattr(self, "initial_user_id", None) or getattr(self, "user", None)
             )
 

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

@@ -7,7 +7,10 @@ from application.api.answer.services.compression.service import CompressionServi
 from application.api.answer.services.compression.threshold_checker import (
     CompressionThresholdChecker,
 )
-from application.api.answer.services.compression.types import CompressionResult
+from application.api.answer.services.compression.types import (
+    CompressionResult,
+    is_compression_summary_row,
+)
 from application.core.model_utils import (
     get_api_key_for_provider,
     get_provider_from_model_id,
@@ -214,7 +217,12 @@ class CompressionOrchestrator:
                     start_index = int(points[-1].get("query_index", -1)) + 1
                 except (TypeError, ValueError):
                     start_index = 0
-            if start_index > compress_up_to:
+            new_queries = [
+                q
+                for q in conversation.get("queries", [])[start_index:]
+                if not is_compression_summary_row(q)
+            ]
+            if start_index > compress_up_to or not new_queries:
                 logger.info(
                     f"No new queries since the last compression point for "
                     f"conversation {conversation_id}; reusing its summary"

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

@@ -82,8 +82,18 @@ class CompressionService:
                     f"compress_up_to_index {compress_up_to_index}"
                 )
             # Only the queries after the previous compression point are new;
-            # the earlier ones are already inside that point's summary.
-            queries_to_compress = queries[start_index : compress_up_to_index + 1]
+            # the earlier ones are already inside that point's summary, and so
+            # is the visible summary row that follows the point.
+            queries_to_compress = [
+                q
+                for q in queries[start_index : compress_up_to_index + 1]
+                if not is_compression_summary_row(q)
+            ]
+            if not queries_to_compress:
+                raise ValueError(
+                    "Nothing to compress: no new queries since the last "
+                    "compression point"
+                )
 
             # Check if there are existing compressions. ``compression_metadata``
             # is a nullable JSONB column, so a never-compressed conversation

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

@@ -4,15 +4,32 @@ from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
-# ``prompt`` of the visible summary row appended after a compression. These
-# rows are display-only: their content already reaches the model through the
-# system prompt (``compressed_summary``), so history replay skips them.
+# The visible summary row appended after a compression. These rows are
+# display-only: their content already reaches the model through the system
+# prompt (``compressed_summary``), so history replay skips them. New rows carry
+# ``metadata[COMPRESSION_SUMMARY_MARKER]``; the prompt text is the display label.
 COMPRESSION_SUMMARY_PROMPT = "[Context Compression Summary]"
+COMPRESSION_SUMMARY_MARKER = "compression_summary"
 
 
 def is_compression_summary_row(query: Any) -> bool:
-    """True for the visible summary row written by ``append_compression_message``."""
-    return isinstance(query, dict) and query.get("prompt") == COMPRESSION_SUMMARY_PROMPT
+    """True for the visible summary row written by ``append_compression_message``.
+
+    Rows written since the marker exists are recognised by it alone. Rows
+    written before it are recognised by the label only when nothing marks them
+    as a real turn (no tool calls, no per-turn metadata), so a user who types
+    the label text as a question keeps that turn in their history.
+    """
+    if not isinstance(query, dict):
+        return False
+    metadata = query.get("metadata")
+    if isinstance(metadata, dict) and metadata.get(COMPRESSION_SUMMARY_MARKER) is True:
+        return True
+    return (
+        query.get("prompt") == COMPRESSION_SUMMARY_PROMPT
+        and not query.get("tool_calls")
+        and not metadata
+    )
 
 
 @dataclass

application/api/answer/services/conversation_service.py

@@ -14,7 +14,10 @@ from typing import Any, Dict, List, Optional
 from sqlalchemy import text as sql_text
 
 from application.core.settings import settings
-from application.api.answer.services.compression.types import COMPRESSION_SUMMARY_PROMPT
+from application.api.answer.services.compression.types import (
+    COMPRESSION_SUMMARY_MARKER,
+    COMPRESSION_SUMMARY_PROMPT,
+)
 from application.storage.db.base_repository import looks_like_uuid
 from application.storage.db.repositories.agents import AgentsRepository
 from application.storage.db.repositories.conversations import (
@@ -605,6 +608,8 @@ class ConversationService:
                 )
                 repo.append_message(conv_pg_id, {
                     "prompt": COMPRESSION_SUMMARY_PROMPT,
+                    # Durable marker: replay filters on this, not on the label.
+                    "metadata": {COMPRESSION_SUMMARY_MARKER: True},
                     "response": summary,
                     "thought": "",
                     "sources": [],

application/core/settings.py

@@ -377,11 +377,12 @@ class Settings(BaseSettings):
     # compressed after that turn was produced.
     OPENAI_RESPONSES_CHAIN_ACROSS_TURNS: bool = True
     OPENAI_RESPONSES_CHAIN_BUDGET_TOKENS: Optional[int] = None
-    # ``truncation: "auto"`` lets the provider drop middle items instead of
-    # failing every request once a chain exceeds the model's native window.
+    # ``truncation: "auto"`` lets the provider drop the oldest input items
+    # instead of failing every request once a chain exceeds the model's window.
     OPENAI_RESPONSES_TRUNCATION_AUTO: bool = False
     # Prompt-cache hints on the Responses API: route a user's calls to the
-    # same cache shard, and request extended retention where offered.
+    # same cache shard (opaque per-user key), and request extended retention
+    # where offered.
     OPENAI_PROMPT_CACHE_KEY: bool = True
     OPENAI_PROMPT_CACHE_RETENTION: Optional[str] = None
     OPENAI_REASONING_SUMMARY: str = "auto"

application/llm/openai.py

@@ -264,7 +264,7 @@ class OpenAILLM(BaseLLM):
         # transcript instead of deduping it, so a chained round only carries
         # the head when it changed.
         self._chain_system_hash = None
-        # Prompt-cache routing key (the user id), set by the agent per call.
+        # Opaque per-user prompt-cache routing key, set by the agent per call.
         self._prompt_cache_key = None
         # Files-API ids for inline ``file_data`` content parts already
         # uploaded, keyed by content hash. First-line cache for the
@@ -1293,7 +1293,7 @@ class OpenAILLM(BaseLLM):
         # server-side (store=true).
         params["include"] = ["reasoning.encrypted_content"]
         # Backstop against a chain that outgrows the model's native window:
-        # the provider drops middle items instead of failing every request.
+        # the provider drops the oldest input items instead of failing.
         if getattr(settings, "OPENAI_RESPONSES_TRUNCATION_AUTO", False):
             params["truncation"] = "auto"
         # Prompt-cache hints. The key pins a conversation to one cache shard;

application/storage/db/repositories/conversations.py

@@ -517,6 +517,8 @@ class ConversationsRepository:
         ).scalar()
         if (
             isinstance(last_point, dict)
+            and point.get("query_index") is not None
+            and point.get("compressed_summary") is not None
             and last_point.get("query_index") == point.get("query_index")
             and last_point.get("compressed_summary") == point.get("compressed_summary")
         ):

docs/content/Deploying/DocsGPT-Settings.mdx

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

docs/content/Models/cloud-providers.mdx

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

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

@@ -244,3 +244,97 @@ class TestHistoryHelpers:
         assert result.success and not result.compression_performed
         assert result.compressed_summary == "S"
         assert result.last_compression_at == EPOCH
+
+
+@pytest.mark.unit
+class TestSummaryRowMarker:
+    """The visible summary row is recognised by its persisted marker; a user
+    who types the label text as a question keeps that turn."""
+
+    def test_marked_row_is_a_summary_row(self):
+        from application.api.answer.services.compression.types import (
+            COMPRESSION_SUMMARY_MARKER,
+            is_compression_summary_row,
+        )
+
+        row = {"prompt": "anything", "response": "S", "metadata": {COMPRESSION_SUMMARY_MARKER: True}}
+        assert is_compression_summary_row(row) is True
+
+    def test_legacy_row_without_metadata_is_a_summary_row(self):
+        from application.api.answer.services.compression.types import is_compression_summary_row
+
+        assert is_compression_summary_row({"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"}) is True
+        assert is_compression_summary_row(
+            {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S", "metadata": {}}
+        ) is True
+
+    def test_user_turn_with_the_label_text_is_kept(self):
+        from application.api.answer.services.compression.types import is_compression_summary_row
+
+        real_turn = {
+            "prompt": COMPRESSION_SUMMARY_PROMPT,
+            "response": "an answer",
+            "metadata": {"usage": {"prompt_tokens": 10}, "response_id": "resp_1"},
+        }
+        assert is_compression_summary_row(real_turn) is False
+        with_tools = {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "r", "tool_calls": [{"tool_name": "x"}]}
+        assert is_compression_summary_row(with_tools) is False
+        result = CompressionResult.success_no_compression([real_turn])
+        assert [h["prompt"] for h in result.as_history()] == [COMPRESSION_SUMMARY_PROMPT]
+
+
+@pytest.mark.unit
+class TestIncrementalTailExcludesSummaryRows:
+    def test_compress_conversation_skips_the_summary_row_in_the_tail(self):
+        llm = MagicMock()
+        llm.gen.return_value = "<summary>new</summary>"
+        svc = CompressionService(llm=llm, model_id="m")
+        svc.prompt_builder = MagicMock(version="v1.0")
+        svc.prompt_builder.build_prompt.return_value = [{"role": "user", "content": "p"}]
+        conv = _compressed_conversation()
+        # A mid-execution compression appends its visible row right after the point.
+        conv["queries"].insert(2, {"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"})
+
+        svc.compress_conversation(conv, compress_up_to_index=4, start_index=2)
+
+        queries, existing = svc.prompt_builder.build_prompt.call_args[0]
+        assert [q["prompt"] for q in queries] == ["q2", "q3"]
+        assert existing == [POINT]
+
+    def test_only_summary_rows_since_the_point_is_rejected(self):
+        svc = CompressionService(llm=MagicMock(), model_id="m")
+        conv = _compressed_conversation()
+        conv["queries"] = conv["queries"][:2] + [{"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"}]
+        with pytest.raises(ValueError, match="Nothing to compress"):
+            svc.compress_conversation(conv, compress_up_to_index=2, start_index=2)
+
+    @patch(
+        "application.api.answer.services.compression.orchestrator.get_provider_from_model_id",
+        return_value="openai",
+    )
+    @patch(
+        "application.api.answer.services.compression.orchestrator.get_api_key_for_provider",
+        return_value="sk",
+    )
+    @patch("application.api.answer.services.compression.orchestrator.LLMCreator")
+    @patch("application.api.answer.services.compression.orchestrator.CompressionService")
+    @patch("application.api.answer.services.compression.orchestrator.settings")
+    def test_orchestrator_reuses_summary_when_only_summary_rows_follow_the_point(
+        self, mock_settings, MockCompressionService, MockLLMCreator, _key, _provider,
+        orchestrator, conversation_service, threshold_checker,
+    ):
+        mock_settings.COMPRESSION_MODEL_OVERRIDE = None
+        conv = _compressed_conversation()
+        conv["queries"] = conv["queries"][:2] + [{"prompt": COMPRESSION_SUMMARY_PROMPT, "response": "S"}]
+        conversation_service.get_conversation.return_value = conv
+        threshold_checker.should_compress.return_value = True
+        MockLLMCreator.create_llm.return_value = MagicMock()
+        svc = MagicMock()
+        svc.get_compressed_context.return_value = ("S", [])
+        MockCompressionService.return_value = svc
+
+        result = orchestrator.compress_if_needed("conv1", "user1", "m", {"sub": "user1"})
+
+        assert result.success and not result.compression_performed
+        assert result.compressed_summary == "S"
+        svc.compress_and_save.assert_not_called()

tests/llm/test_responses_chain_budget.py

@@ -297,3 +297,18 @@ def test_emit_responses_metadata_records_compression_epoch(monkeypatch):
     events = list(agent._emit_responses_metadata())
     assert events and events[0]["metadata"]["compression_epoch"] == "2026-09-03T10:00:00+00:00"
     assert events[0]["metadata"]["response_id"] == "resp_9"
+
+
+# ── cache key is opaque ─────────────────────────────────────────────────────
+
+
+@pytest.mark.unit
+def test_cache_key_for_user_is_opaque_and_stable():
+    from application.agents.base import _cache_key_for_user
+
+    key = _cache_key_for_user("user_2Vhzgd63RSgixvvbF8Z2nhtqnE9")
+    assert key and "user_2Vhzgd" not in key
+    assert len(key) == 32 and all(c in "0123456789abcdef" for c in key)
+    assert key == _cache_key_for_user("user_2Vhzgd63RSgixvvbF8Z2nhtqnE9")
+    assert key != _cache_key_for_user("someone-else")
+    assert _cache_key_for_user(None) is None

tests/storage/db/repositories/test_conversations.py

@@ -381,6 +381,18 @@ class TestCompressionMetadata:
         points = fetched["compression_metadata"]["compression_points"]
         assert [p["summary"] for p in points] == ["p2", "p3", "p4"]
 
+    def test_append_compression_point_ignores_an_identical_point(self, pg_conn):
+        repo = _repo(pg_conn)
+        conv = repo.create("user-1", "c")
+        point = {"query_index": 3, "compressed_summary": "S", "original_token_count": 90}
+        assert repo.append_compression_point(conv["id"], point, max_points=3) is True
+        # Two persist paths used to append the same point twice.
+        assert repo.append_compression_point(conv["id"], dict(point), max_points=3) is True
+        later = {"query_index": 5, "compressed_summary": "S2", "original_token_count": 40}
+        assert repo.append_compression_point(conv["id"], later, max_points=3) is True
+        points = repo.get(conv["id"], "user-1")["compression_metadata"]["compression_points"]
+        assert [p["query_index"] for p in points] == [3, 5]
+
 
 class TestResolveAgentRef:
     """The repo must translate Mongo ObjectId-shaped ``agent_id`` values