diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index c236eb209d..cbaa3337f0 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/backend/packages/harness/deerflow/models/AGENTS.md @@ -63,6 +63,12 @@ the same policy first so run metadata reports the effective values. Design note: - `ClaudeChatModel.model_post_init` calls `load_claude_code_credential()` for every instance, and `create_chat_model` builds fresh instances per run (lead agent, title, summarization, subagents) - `$CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR` is a one-shot handoff: a pipe returns EOF and a file keeps its advanced offset. `_read_secret_from_file_descriptor` therefore caches a non-empty secret per `(env_var, fd)` under a lock held across the read. Do not drop the cache or the lock — later instances would get no credential, and the Anthropic SDK raises `TypeError: Could not resolve authentication method` before sending. Empty reads and `OSError` are not cached. The key is the descriptor number on purpose — a closed handoff keeps serving its token, and a secret placed on a recycled number in-process is not re-read unless the cache is cleared. The cache is per process, so a new process (e.g. a uvicorn `--reload` worker) cannot recover a drained descriptor. Pinned by `tests/test_credential_loader.py`, including a two-instance `ClaudeChatModel` test +### Claude Prompt Caching (`packages/harness/deerflow/models/claude_provider.py`) + +- The request payload shares objects with the caller: langchain-anthropic forwards Claude-native blocks (an image or document with a `source`, search results) and list-form system blocks by reference, and a reused tool binding passes its own tool dicts (the lead agent re-binds per call, so its tool dicts are fresh). Writing `cache_control` in place checkpointed the markers with the thread's messages, and the stale ones pushed every later request past the 4-breakpoint limit +- Every request goes through `_strip_cache_control`, with caching on or off, so markers stored by older checkpoints never reach the API. It copies a marked block without its marker and replaces the system, message, content and tool lists and every message dict with copies; langchain-anthropic already builds fresh message dicts and content lists, so that part is defensive +- `_apply_prompt_caching` must call `_strip_cache_control` first: it then replaces slots in those payload-owned lists with marked copies and writes `msg["content"]` on copied message dicts, placing at most four breakpoints. Pinned by `tests/test_claude_provider_prompt_caching.py` + ### vLLM Provider (`packages/harness/deerflow/models/vllm_provider.py`) - `VllmChatModel` subclasses `langchain_openai:ChatOpenAI` for vLLM 0.19.0 OpenAI-compatible endpoints diff --git a/backend/packages/harness/deerflow/models/claude_provider.py b/backend/packages/harness/deerflow/models/claude_provider.py index 7116ad13e1..23ea975132 100644 --- a/backend/packages/harness/deerflow/models/claude_provider.py +++ b/backend/packages/harness/deerflow/models/claude_provider.py @@ -146,6 +146,11 @@ class ClaudeChatModel(ChatAnthropic): if self.enable_prompt_caching: self._apply_prompt_caching(payload) + else: + # Checkpoints written before breakpoints were placed on copies can + # carry markers on old blocks; a thread with more than four of them + # must not stay broken just because caching was switched off. + self._strip_cache_control(payload) if self.auto_thinking_budget: self._apply_thinking_budget(payload) @@ -197,28 +202,38 @@ class ClaudeChatModel(ChatAnthropic): placed on the *last* eligible blocks because later breakpoints cover a larger prefix and yield better cache hit rates. + The budget covers the whole request, so markers already present are + dropped first. Checkpoints written before breakpoints were placed on + copies can still carry markers on old blocks, and those would otherwise + push every later request of the thread past the limit. A marked block + is always a copy; ``_strip_cache_control`` explains why. + The system prompt is expected to be fully static (no per-user memory or current date). Dynamic context is injected per-turn via DynamicContextMiddleware as a in the first HumanMessage. """ MAX_CACHE_BREAKPOINTS = 4 - # Collect candidate blocks in document order: + # Must run first: stripping leaves every list and message dict below + # owned by this payload, which is what makes replacing a candidate + # slot (and writing msg["content"]) safe for any payload. + self._strip_cache_control(payload) + + # Collect candidate slots (list, index) in document order: # 1. system text blocks # 2. content blocks of the last prompt_cache_size messages # 3. the last tool definition - candidates: list[dict] = [] + candidates: list[tuple[list, int]] = [] # 1. System blocks system = payload.get("system") if system and isinstance(system, list): - for block in system: + for index, block in enumerate(system): if isinstance(block, dict) and block.get("type") == "text": - candidates.append(block) + candidates.append((system, index)) elif system and isinstance(system, str): - new_block: dict = {"type": "text", "text": system} - payload["system"] = [new_block] - candidates.append(new_block) + payload["system"] = [{"type": "text", "text": system}] + candidates.append((payload["system"], 0)) # 2. Recent message blocks messages = payload.get("messages", []) @@ -229,23 +244,22 @@ class ClaudeChatModel(ChatAnthropic): continue content = msg.get("content") if isinstance(content, list): - for block in content: + for index, block in enumerate(content): if isinstance(block, dict): - candidates.append(block) + candidates.append((content, index)) elif isinstance(content, str) and content: - new_block = {"type": "text", "text": content} - msg["content"] = [new_block] - candidates.append(new_block) + msg["content"] = [{"type": "text", "text": content}] + candidates.append((msg["content"], 0)) # 3. Last tool definition tools = payload.get("tools", []) if tools and isinstance(tools[-1], dict): - candidates.append(tools[-1]) + candidates.append((tools, len(tools) - 1)) # Apply cache_control only to the last MAX_CACHE_BREAKPOINTS candidates # to stay within the API limit. - for block in candidates[-MAX_CACHE_BREAKPOINTS:]: - block["cache_control"] = {"type": "ephemeral"} + for container, index in candidates[-MAX_CACHE_BREAKPOINTS:]: + container[index] = {**container[index], "cache_control": {"type": "ephemeral"}} def _apply_thinking_budget(self, payload: dict) -> None: """Auto-allocate thinking budget (80% of max_tokens).""" @@ -262,26 +276,46 @@ class ClaudeChatModel(ChatAnthropic): @staticmethod def _strip_cache_control(payload: dict) -> None: - """Remove cache_control markers before OAuth requests reach Anthropic.""" - for section in ("system", "messages"): - items = payload.get(section) - if not isinstance(items, list): - continue - for item in items: - if not isinstance(item, dict): - continue - item.pop("cache_control", None) - content = item.get("content") - if isinstance(content, list): - for block in content: - if isinstance(block, dict): - block.pop("cache_control", None) + """Remove cache_control markers without writing into the payload's objects. + + Every request is stripped: prompt caching strips before placing its own + breakpoints, a request without caching is stripped in + ``_get_request_payload``, and OAuth requests again before they reach + Anthropic. + + The payload shares objects with the caller. langchain-anthropic passes + Claude-native blocks (an image or document with a ``source``, a search + result) and list-form system blocks through by reference, and a reused + tool binding passes its own tool dicts. Changing them in place changes + the thread's messages, and a marker written there is checkpointed with + them. So a block that carries a marker is copied without it, and the + system, message, content and tool lists and every message dict are + replaced with copies. langchain-anthropic already builds fresh message + dicts and content lists, so those copies are defensive; they are what + lets ``_apply_prompt_caching`` write into them for any payload. + """ + + def without_markers(items: list) -> list: + return [{key: value for key, value in item.items() if key != "cache_control"} if isinstance(item, dict) and "cache_control" in item else item for item in items] + + system = payload.get("system") + if isinstance(system, list): + payload["system"] = without_markers(system) + + messages = payload.get("messages") + if isinstance(messages, list): + copied: list = [] + for message in messages: + if isinstance(message, dict): + message = {key: value for key, value in message.items() if key != "cache_control"} + if isinstance(message.get("content"), list): + message["content"] = without_markers(message["content"]) + copied.append(message) + payload["messages"] = copied tools = payload.get("tools") if isinstance(tools, list): - for tool in tools: - if isinstance(tool, dict): - tool.pop("cache_control", None) + payload["tools"] = without_markers(tools) def _create(self, payload: dict) -> Any: if self._is_oauth: diff --git a/backend/tests/test_claude_provider_prompt_caching.py b/backend/tests/test_claude_provider_prompt_caching.py index e212b7329a..d439ae718d 100644 --- a/backend/tests/test_claude_provider_prompt_caching.py +++ b/backend/tests/test_claude_provider_prompt_caching.py @@ -2,12 +2,14 @@ Validates that the function never places more than 4 cache_control breakpoints (the hard limit enforced by the Anthropic API and AWS Bedrock) regardless of -how many system blocks, message content blocks, or tool definitions are present. +how many system blocks, message content blocks, or tool definitions are present, +and that it never writes into blocks the caller still owns. """ from unittest import mock import pytest +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage from deerflow.models.claude_provider import ClaudeChatModel @@ -247,3 +249,99 @@ def test_old_messages_outside_cache_window_not_cached(model): # Only the last message should be within the cache window assert "cache_control" not in payload["messages"][0]["content"][0] assert payload["messages"][1]["content"][0].get("cache_control") == {"type": "ephemeral"} + + +# --------------------------------------------------------------------------- +# Caller-owned blocks +# --------------------------------------------------------------------------- + + +def _native_image() -> dict: + """A Claude-native image block, which langchain-anthropic forwards by reference.""" + return {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "iVBORw0KGgo="}} + + +def test_breakpoints_are_placed_on_copies_of_caller_blocks(model): + system_block = {"type": "text", "text": "sys"} + image = _native_image() + tool = {"name": "bash"} + payload: dict = { + "system": [system_block], + "messages": [{"role": "user", "content": [image]}], + "tools": [tool], + } + + model._apply_prompt_caching(payload) + + assert _count_cache_control(payload) == 3 + assert payload["messages"][0]["content"][0] == {**_native_image(), "cache_control": {"type": "ephemeral"}} + assert "cache_control" not in system_block + assert "cache_control" not in image + assert "cache_control" not in tool + + +def test_repeated_requests_leave_thread_messages_unmarked(model): + """Regression: a marker written into a message's own block was checkpointed. + + Every later request then carried the stale markers of earlier turns on top + of its own four, and the API rejected the thread from the third turn on. + """ + history: list = [SystemMessage(content="You are helpful.")] + for turn in range(1, 6): + history.append(HumanMessage(content=[_native_image()], id=f"human-{turn}")) + payload = model._get_request_payload(history) + assert _count_cache_control(payload) <= 4, f"turn {turn}" + history.append(AIMessage(content=f"answer {turn}", id=f"ai-{turn}")) + + assert all(message.content == [_native_image()] for message in history if isinstance(message, HumanMessage)) + + +@pytest.mark.parametrize("enable_prompt_caching", [True, False]) +def test_markers_stored_in_the_thread_stop_reaching_the_api(model, enable_prompt_caching): + """Threads checkpointed before the fix carry a marker on every such block. + + Recovery must not depend on caching being on: switching it off is the + obvious reaction to the 400s, and five stored markers still exceed the limit. + """ + model.enable_prompt_caching = enable_prompt_caching + stored = {**_native_image(), "cache_control": {"type": "ephemeral"}} + history: list = [SystemMessage(content="You are helpful.")] + for turn in range(1, 6): + history.append(HumanMessage(content=[dict(stored)], id=f"human-{turn}")) + payload = model._get_request_payload(history) + assert _count_cache_control(payload) <= (4 if enable_prompt_caching else 0), f"turn {turn}" + history.append(AIMessage(content=f"answer {turn}", id=f"ai-{turn}")) + + +def test_markers_already_in_the_request_are_replaced(model): + """Checkpoints written by older versions can carry markers on old blocks.""" + stale = [{**_native_image(), "cache_control": {"type": "ephemeral"}} for _ in range(5)] + payload: dict = { + "system": [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}], + "messages": [{"role": "user", "content": [block]} for block in stale], + "tools": [{"name": "bash", "cache_control": {"type": "ephemeral"}}], + } + + model._apply_prompt_caching(payload) + + # Candidates: the system block, the last three messages, the tool — the + # system block falls outside the last four. + assert _count_cache_control(payload) == 4 + assert [index for index, message in enumerate(payload["messages"]) if "cache_control" in message["content"][0]] == [2, 3, 4] + assert "cache_control" not in payload["system"][0] + assert payload["tools"][0]["cache_control"] == {"type": "ephemeral"} + assert all("cache_control" in block for block in stale) + + +def test_strip_cache_control_copies_instead_of_popping(): + system_block = {"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}} + image = {**_native_image(), "cache_control": {"type": "ephemeral"}} + tool = {"name": "bash", "cache_control": {"type": "ephemeral"}} + message = {"role": "user", "content": [image], "cache_control": {"type": "ephemeral"}} + payload: dict = {"system": [system_block], "messages": [message], "tools": [tool]} + + ClaudeChatModel._strip_cache_control(payload) + + assert _count_cache_control(payload) == 0 + assert payload["messages"][0] == {"role": "user", "content": [_native_image()]} + assert all("cache_control" in item for item in (system_block, image, tool, message))