fix(openai): cache stable prompts explicitly

docs/v6/reference/agents.mdx

@@ -68,6 +68,10 @@ workspaces; computer use over an `rfb` capability currently requires a POSIX wor
 (`model`, `max_steps`, `timeout_seconds`, `tool_timeout_seconds`, `system_prompt`, `citations_enabled`, `stop_on`) lives on the
 config; `__call__(run)` takes only the run.
 
+For GPT-5.6 and later, `OpenAIAgent` places an explicit cache breakpoint after `system_prompt` and
+leaves changing rollout input outside the cached prefix. Set `OpenAIConfig.prompt_cache_key` to group
+high-volume requests that share the same prefix.
+
 `timeout_seconds` bounds the complete agent phase. For provider tool agents, `tool_timeout_seconds` bounds each complete SSH-backed tool call, including multi-operation editor calls. It is unset by default except on `ClaudeConfig`, where it defaults to 120 seconds. A timeout is returned to the model as a tool error so the agent can continue. `ClaudeSDKAgent` does not apply this setting because its SSH process is the complete Claude Code agent, not one tool call.
 
 ```python

hud/agents/openai/agent.py

@@ -4,6 +4,7 @@
 
 import json
 import logging
+import re
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Literal, cast
 
@@ -16,7 +17,10 @@
     ToolParam,
 )
 from openai.types.responses.easy_input_message_param import EasyInputMessageParam
-from openai.types.responses.response_create_params import ToolChoice  # noqa: TC002
+from openai.types.responses.response_create_params import (  # noqa: TC002
+    PromptCacheOptions,
+    ToolChoice,
+)
 from openai.types.responses.response_input_param import (
     ComputerCallOutput,
     Message,
@@ -40,6 +44,15 @@
 
 logger = logging.getLogger(__name__)
 
+_GPT_MODEL_VERSION = re.compile(r"(?:^|[./])gpt-(\d+)(?:\.(\d+))?(?:-|$)")
+
+
+def _supports_explicit_prompt_caching(model: str) -> bool:
+    match = _GPT_MODEL_VERSION.search(model.lower())
+    if match is None:
+        return False
+    return (int(match.group(1)), int(match.group(2) or 0)) >= (5, 6)
+
 
 @dataclass
 class OpenAIRunState(RunState[ResponseInputItemParam]):
@@ -78,6 +91,7 @@ def __init__(self, config: OpenAIConfig | None = None) -> None:
         self.reasoning: Reasoning | None = config.reasoning
         self.tool_choice: ToolChoice | None = config.tool_choice
         self.parallel_tool_calls = config.parallel_tool_calls
+        self.prompt_cache_key = config.prompt_cache_key
         self.text = config.text
         self.truncation: Literal["auto", "disabled"] | None = config.truncation
 
@@ -178,6 +192,29 @@ async def get_response(
             else:
                 return AgentStep(content="", done=True)
 
+        explicit_prompt_caching = system_prompt is not None and _supports_explicit_prompt_caching(
+            self._model
+        )
+        if explicit_prompt_caching and oai_state.last_response_id is None:
+            new_items = [
+                Message(
+                    role="developer",
+                    content=[
+                        ResponseInputTextParam(
+                            type="input_text",
+                            text=system_prompt,
+                            prompt_cache_breakpoint={"mode": "explicit"},
+                        )
+                    ],
+                ),
+                *new_items,
+            ]
+
+        instructions_param: str | Omit | None = Omit() if explicit_prompt_caching else system_prompt
+        prompt_cache_options: PromptCacheOptions | Omit = (
+            {"mode": "explicit"} if explicit_prompt_caching else Omit()
+        )
+
         include_param: list[ResponseIncludable] | Omit = Omit()
         if citations_enabled:
             include_param = ["web_search_call.action.sources"]
@@ -212,7 +249,7 @@ async def get_response(
         response = await self.openai_client.responses.create(
             model=self._model,
             input=new_items,
-            instructions=system_prompt,
+            instructions=instructions_param,
             max_output_tokens=self.max_output_tokens,
             temperature=self.temperature,
             text=self.text if self.text is not None else Omit(),
@@ -223,6 +260,8 @@ async def get_response(
             previous_response_id=(
                 oai_state.last_response_id if oai_state.last_response_id is not None else Omit()
             ),
+            prompt_cache_key=self.prompt_cache_key if self.prompt_cache_key is not None else Omit(),
+            prompt_cache_options=prompt_cache_options,
             truncation=self.truncation if self.truncation is not None else Omit(),
             include=include_param,
         )
@@ -319,6 +358,7 @@ async def get_response(
                 prompt_tokens=response.usage.input_tokens,
                 completion_tokens=response.usage.output_tokens,
                 cached_tokens=response.usage.input_tokens_details.cached_tokens,
+                cache_write_tokens=response.usage.input_tokens_details.cache_write_tokens,
             )
         # The Responses API has no finish_reason; truncation surfaces as
         # incomplete_details.reason ("max_output_tokens" / "content_filter").

hud/agents/tests/test_openai_agent.py

@@ -8,6 +8,7 @@
 from typing import Any, cast
 
 import mcp.types as mcp_types
+from openai import Omit
 from openai.types.responses import ResponseOutputText
 
 from hud.agents.openai.agent import OpenAIAgent, OpenAIRunState
@@ -32,8 +33,16 @@ def __init__(self, response: Any) -> None:
         self.responses = FakeResponses(response)
 
 
-def _agent(response: Any) -> OpenAIAgent:
-    return OpenAIAgent(OpenAIConfig(model="gpt-test", model_client=FakeOpenAI(response)))
+def _agent(
+    response: Any, *, model: str = "gpt-test", prompt_cache_key: str | None = None
+) -> OpenAIAgent:
+    return OpenAIAgent(
+        OpenAIConfig(
+            model=model,
+            model_client=FakeOpenAI(response),
+            prompt_cache_key=prompt_cache_key,
+        )
+    )
 
 
 def test_format_message_shapes_user_text() -> None:
@@ -106,7 +115,7 @@ async def test_get_response_parses_text_and_function_call() -> None:
         usage=SimpleNamespace(
             input_tokens=9,
             output_tokens=4,
-            input_tokens_details=SimpleNamespace(cached_tokens=2),
+            input_tokens_details=SimpleNamespace(cached_tokens=2, cache_write_tokens=3),
         ),
     )
     agent = _agent(response)
@@ -125,6 +134,50 @@ async def test_get_response_parses_text_and_function_call() -> None:
     assert result.usage.prompt_tokens == 9
     assert result.usage.completion_tokens == 4
     assert result.usage.cached_tokens == 2
+    assert result.usage.cache_write_tokens == 3
+
+
+async def test_get_response_caches_gpt_5_6_system_prompt_before_user_input() -> None:
+    agent = _agent(
+        _api_response("resp_cache", []),
+        model="gpt-5.6",
+        prompt_cache_key="shared-agent-v1",
+    )
+    user_message = agent._format_message("user", "dynamic input")
+    state = OpenAIRunState(messages=[user_message])
+
+    await agent.get_response(state, system_prompt="stable instructions")
+
+    call = cast("Any", agent.openai_client.responses).calls[0]
+    assert isinstance(call["instructions"], Omit)
+    assert call["prompt_cache_key"] == "shared-agent-v1"
+    assert call["prompt_cache_options"] == {"mode": "explicit"}
+    assert call["input"] == [
+        {
+            "role": "developer",
+            "content": [
+                {
+                    "type": "input_text",
+                    "text": "stable instructions",
+                    "prompt_cache_breakpoint": {"mode": "explicit"},
+                }
+            ],
+        },
+        user_message,
+    ]
+
+
+async def test_get_response_keeps_system_prompt_as_instructions_before_gpt_5_6() -> None:
+    agent = _agent(_api_response("resp_legacy", []), model="gpt-5.5")
+    user_message = agent._format_message("user", "input")
+    state = OpenAIRunState(messages=[user_message])
+
+    await agent.get_response(state, system_prompt="instructions")
+
+    call = cast("Any", agent.openai_client.responses).calls[0]
+    assert call["instructions"] == "instructions"
+    assert isinstance(call["prompt_cache_options"], Omit)
+    assert call["input"] == [user_message]
 
 
 async def test_get_response_done_when_no_tool_calls() -> None:

hud/agents/types.py

@@ -121,6 +121,7 @@ class OpenAIConfig(AgentConfig):
     text: Any = None  # {"verbosity": "low"|"medium"|"high"}
     truncation: Literal["auto", "disabled"] | None = None
     parallel_tool_calls: bool | None = None
+    prompt_cache_key: str | None = Field(default=None, min_length=1, max_length=64)
 
 
 class OpenAIChatConfig(AgentConfig):
@@ -256,6 +257,7 @@ class Usage(BaseModel):
     prompt_tokens: int | None = None
     completion_tokens: int | None = None
     cached_tokens: int | None = None
+    cache_write_tokens: int | None = None
     cost_usd: float | None = None
     llm_call_count: int | None = None