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