mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-25 17:18:11 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(agent): cache static system prompt prefixes

Browse source
This commit is contained in:
konsisumer 2026-07-22 22:55:43 +02:00 • committed by Teknium
parent 6c14b12d1f
commit 9fdadf0cd7
10 changed files with 351 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

10
agent/agent_init.py
View file

@ -823,9 +823,10 @@ def init_agent(
# Anthropic prompt caching: auto-enabled for Claude models on native
# Anthropic, OpenRouter, and third-party gateways that speak the
# Anthropic protocol (``api_mode == 'anthropic_messages'``). Reduces
# input costs by ~75% on multi-turn conversations. Uses system_and_3
# strategy (4 breakpoints). See ``_anthropic_prompt_cache_policy``
# for the layout-vs-transport decision.
# input costs by ~75% on multi-turn conversations. Uses four breakpoints:
# the static system prefix, full system prompt, and last two messages
# (falling back to system-and-3 when no static prefix is available). See
# ``_anthropic_prompt_cache_policy`` for the layout-vs-transport decision.
agent._use_prompt_caching, agent._use_native_cache_layout = (
agent._anthropic_prompt_cache_policy()
)
@ -1494,6 +1495,9 @@ def init_agent(
# Cached system prompt -- built once per session, only rebuilt on compression
agent._cached_system_prompt: Optional[str] = None
# Cross-session-stable prefix of the cached prompt. It remains separate
# from the persisted string and is used only to place an early cache marker.
agent._cached_system_prompt_static: Optional[str] = None
# Filesystem checkpoint manager (transparent — not a tool)
from tools.checkpoint_manager import CheckpointManager

45
agent/coding_context.py
View file

@ -520,30 +520,46 @@ class RuntimeMode:
return None
return [self.profile.toolset, *_enabled_mcp_servers(config)]
def system_blocks(self) -> list[str]:
"""Stable system-prompt blocks for this posture (brief + workspace).
def system_prompt_parts(self) -> tuple[list[str], list[str], list[str]]:
"""Return prefix, workspace, and trailing posture blocks separately.
The operating brief carries a model-family edit-format nudge appended
to it (one cached string, not a separate block) so the model is steered
toward the `patch` mode it handles best — see ``_edit_format_line``.
The three lists preserve the historical flat prompt order: the brief,
the live workspace snapshot, then configured operator instructions.
Prompt assembly can therefore put a cache boundary before the snapshot
without changing the persisted system-prompt bytes.
"""
if not self.is_coding:
return []
blocks: list[str] = []
return [], [], []
prefix: list[str] = []
workspace_parts: list[str] = []
trailing: list[str] = []
if self.profile.guidance:
brief = self.profile.guidance
edit_line = _edit_format_line(self.model)
if edit_line:
brief = f"{brief}\n{edit_line}"
blocks.append(brief)
prefix.append(brief)
workspace = build_coding_workspace_block(self.cwd)
if workspace:
blocks.append(workspace)
workspace_parts.append(workspace)
# Operator instructions ride their own block so the brief (block 0) stays
# byte-stable and cache-keyed independently of user config.
if self.instructions:
blocks.append(f"Operator instructions (from config):\n{self.instructions}")
return blocks
trailing.append(f"Operator instructions (from config):\n{self.instructions}")
return prefix, workspace_parts, trailing
def system_blocks(self) -> list[str]:
"""Return posture blocks in their historical display order.
``system_prompt_parts`` is the cache-aware API. This compatibility
helper retains the public flat list for callers outside prompt assembly.
"""
prefix, workspace, trailing = self.system_prompt_parts()
return [*prefix, *workspace, *trailing]
def compact_skill_categories(self) -> frozenset[str]:
"""Skill categories to demote to names-only in the prompt's skill index.
@ -644,6 +660,19 @@ def coding_system_blocks(
).system_blocks()
def coding_system_prompt_parts(
*,
platform: Optional[str] = None,
cwd: Optional[str | Path] = None,
config: Optional[dict[str, Any]] = None,
model: Optional[str] = None,
) -> tuple[list[str], list[str], list[str]]:
"""Return coding prefix, workspace snapshot, and trailing guidance."""
return resolve_runtime_mode(
platform=platform, cwd=cwd, config=config, model=model
).system_prompt_parts()
def coding_compact_skill_categories(
*,
platform: Optional[str] = None,

20
agent/conversation_loop.py
View file

@ -1278,9 +1278,9 @@ def run_conversation(
#
# Hermes invariant: the system prompt is built ONCE per session
# (cached on ``_cached_system_prompt``) and replayed verbatim on
# every turn. We send it as a single content string so the
# bytes are byte-stable across turns and upstream prompt caches
# stay warm.
# every turn. ``apply_anthropic_cache_control`` may split its stable
# prefix into content blocks on the wire, but the stored string and
# its byte-stability remain unchanged.
effective_system = active_system_prompt or ""
if agent.ephemeral_system_prompt:
effective_system = (effective_system + "\n\n" + agent.ephemeral_system_prompt).strip()
@ -1367,15 +1367,21 @@ def run_conversation(
# Apply Anthropic prompt caching for Claude models on native
# Anthropic, OpenRouter, and third-party Anthropic-compatible
# gateways. Auto-detected: if ``_use_prompt_caching`` is set,
# inject cache_control breakpoints (system + last 3 messages)
# to reduce input token costs by ~75% on multi-turn
# conversations.
# gateways. Auto-detected: if ``_use_prompt_caching`` is set, inject
# cache_control breakpoints for the static system prefix, full system
# prompt, and last two messages (or the legacy system-and-3 layout
# when no static prefix is available).
if agent._use_prompt_caching:
_static_system_prefix = getattr(agent, "_cached_system_prompt_static", None)
api_messages = apply_anthropic_cache_control(
api_messages,
cache_ttl=agent._cache_ttl,
native_anthropic=agent._use_native_cache_layout,
static_system_prefix=(
_static_system_prefix
if isinstance(_static_system_prefix, str)
else None
),
)
# Safety net: strip orphaned tool results / add stubs for missing

17
agent/moa_loop.py
View file

@ -388,10 +388,10 @@ def _maybe_apply_moa_cache_control(
Reuses the SAME policy function as the main agent loop
(``anthropic_prompt_cache_policy``) resolved against the slot's own
provider/base_url/api_mode/model, and the SAME breakpoint layout
(``apply_anthropic_cache_control``, system_and_3). This keeps advisor and
aggregator calls decorated exactly like an acting agent on that provider
would be — no MoA-specific caching logic to drift.
provider/base_url/api_mode/model and shared marker helper
(``apply_anthropic_cache_control``). MoA has no per-session static prefix,
so it uses the helper's legacy system-and-3 fallback without carrying a
separate caching strategy.
Returns the messages unchanged on any resolution error or when the
policy says the route doesn't honor markers.
@ -480,10 +480,11 @@ def _run_reference(
reserve_output_tokens=max_tokens,
context_length_cache=context_length_cache,
)
# Apply the same Anthropic-style prompt-caching decoration the main
# agent loop applies (system_and_3 breakpoints). The advisory view is
# append-only across iterations (new turns append before the trailing
# synthetic marker), so on cache-honoring routes (Claude via
# Apply the Anthropic-style prompt-caching decoration used by the main
# agent loop. This fixed reference prompt has no session-specific
# prefix split, so the helper uses its legacy system-and-3 fallback.
# The advisory view is append-only across iterations (new turns append
# before the trailing synthetic marker), so on cache-honoring routes (Claude via
# OpenRouter/native, MiniMax, Qwen/DashScope) iteration N+1's prefix
# replays iteration N's cached prefix. Without this, Claude advisors
# served ZERO cache reads across an entire benchmark run (measured:

64
agent/prompt_caching.py
View file

@ -1,9 +1,11 @@
"""Anthropic prompt caching strategy.
Single layout: ``system_and_3``. 4 cache_control breakpoints — system
prompt + last 3 non-system messages, all at the same TTL (5m or 1h).
Reduces input token costs by ~75% on multi-turn conversations within a
single session.
The default layout uses 4 cache_control breakpoints: the static system
prefix, the end of the system prompt, and the last 2 non-system messages.
When a static system prefix is unavailable, it falls back to one system
breakpoint plus the last 3 messages. All markers use the same TTL (5m or 1h).
This preserves intra-session caching while allowing new sessions to reuse the
stable system-prompt prefix.
Pure functions -- no class state, no AIAgent dependency.
"""
@ -81,15 +83,55 @@ def _build_marker(ttl: str) -> Dict[str, str]:
return marker
def _apply_system_cache_markers(
message: dict,
cache_marker: dict,
static_system_prefix: str | None,
*,
native_anthropic: bool,
) -> int:
"""Mark the static system prefix and full prompt when they can be split.
The system prompt remains one stored string. Splitting it only in the
outgoing request keeps session persistence and non-Anthropic transports
unchanged while making the stable prefix independently cacheable.
"""
content = message.get("content")
if (
isinstance(static_system_prefix, str)
and static_system_prefix
and isinstance(content, str)
and content.startswith(static_system_prefix)
):
suffix = content[len(static_system_prefix):]
if suffix:
message["content"] = [
{
"type": "text",
"text": static_system_prefix,
"cache_control": cache_marker,
},
{"type": "text", "text": suffix, "cache_control": cache_marker},
]
return 2
_apply_cache_marker(message, cache_marker, native_anthropic=native_anthropic)
return 1
def apply_anthropic_cache_control(
api_messages: List[Dict[str, Any]],
cache_ttl: str = "5m",
native_anthropic: bool = False,
static_system_prefix: str | None = None,
) -> List[Dict[str, Any]]:
"""Apply system_and_3 caching strategy to messages for Anthropic models.
"""Apply Anthropic cache-control markers to API messages.
Places up to 4 cache_control breakpoints: system prompt + last 3 non-system
messages, all at the same TTL.
When ``static_system_prefix`` exactly matches the beginning of a string
system prompt, it receives an early marker and the full system prompt gets
a trailing marker. The remaining two markers target the latest cacheable
non-system messages. Without that prefix, the legacy system-and-3 layout
is retained.
Returns:
Deep copy of messages with cache_control breakpoints injected.
@ -103,8 +145,12 @@ def apply_anthropic_cache_control(
breakpoints_used = 0
if messages[0].get("role") == "system":
_apply_cache_marker(messages[0], marker, native_anthropic=native_anthropic)
breakpoints_used += 1
breakpoints_used = _apply_system_cache_markers(
messages[0],
marker,
static_system_prefix,
native_anthropic=native_anthropic,
)
remaining = 4 - breakpoints_used
non_sys = [

65
agent/system_prompt.py
View file

@ -12,9 +12,11 @@ Three tiers are joined with ``\\n\\n``:
* ``stable`` — identity (SOUL.md or DEFAULT_AGENT_IDENTITY), tool
guidance, computer-use guidance, nous subscription block, tool-use
enforcement guidance + per-model operational guidance, skills prompt,
alibaba model-name workaround, environment hints, platform hints.
alibaba model-name workaround, environment hints, coding guidance,
platform hints.
* ``context`` — caller-supplied ``system_message`` plus context files
(AGENTS.md / .cursorrules / etc.) discovered under ``TERMINAL_CWD``.
(AGENTS.md / .cursorrules / etc.) discovered under ``TERMINAL_CWD``,
plus the session's coding-workspace snapshot.
* ``volatile`` — memory snapshot, USER.md profile, external memory
provider block, timestamp/session/model/provider line.
@ -145,14 +147,14 @@ def _tui_embedded_pane_clarifier(hint: str) -> str:
def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None) -> Dict[str, str]:
"""Assemble the system prompt as three ordered parts.
"""Assemble the system prompt as three ordered cache tiers.
Returns a dict with three keys:
* ``stable`` — identity, tool guidance, skills prompt,
environment hints, platform hints, model-family operational
guidance.
* ``context`` — context files (AGENTS.md, .cursorrules, etc.)
and caller-supplied system_message.
* ``stable`` — the cross-session-stable prefix, through the coding
operating brief when a workspace snapshot follows.
* ``context`` — the workspace snapshot followed by the remaining
session-stable guidance, context files, and caller-supplied
system_message.
* ``volatile`` — memory snapshot, user profile, external
memory provider block, timestamp line.
@ -345,25 +347,35 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
stable_parts.append(_env_hints)
# Coding posture (base Hermes, any interactive coding surface in a code
# workspace — see agent/coding_context.py). The operating brief + the live
# git/workspace snapshot are built once here and cached for the session;
# the snapshot is never re-probed per turn (that would break the prompt
# cache), so the brief tells the model to re-check git before relying on it.
# workspace — see agent/coding_context.py). Keep the operating brief in
# the cross-session-stable prefix, while placing the live git/workspace
# snapshot behind its own cache boundary. The post-snapshot blocks must
# stay in their historical position after the workspace snapshot.
coding_workspace_parts: List[str] = []
coding_trailing_parts: List[str] = []
if agent.valid_tool_names:
try:
from agent.coding_context import coding_system_blocks
from agent.coding_context import coding_system_prompt_parts
stable_parts.extend(
coding_system_blocks(
platform=agent.platform,
cwd=resolve_context_cwd(),
model=agent.model,
)
coding_prefix_parts, coding_workspace_parts, coding_trailing_parts = coding_system_prompt_parts(
platform=agent.platform,
cwd=resolve_context_cwd(),
model=agent.model,
)
stable_parts.extend(coding_prefix_parts)
except Exception:
# Coding-context probing must never block prompt build.
pass
# Guidance assembled after the coding posture historically followed the
# workspace snapshot. With no snapshot, the coding tail instead remains
# directly after the coding prefix in the cacheable prefix.
if coding_workspace_parts:
post_workspace_parts: List[str] = []
else:
stable_parts.extend(coding_trailing_parts)
post_workspace_parts = stable_parts
# Local Python toolchain probe — names python/pip/uv/PEP-668 state when
# something is non-default so the model can pick the right install
# strategy without discovering by failure. Emits a single line; emits
@ -376,7 +388,7 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
from tools.env_probe import get_environment_probe_line
_probe_line = get_environment_probe_line()
if _probe_line:
stable_parts.append(_probe_line)
post_workspace_parts.append(_probe_line)
except Exception:
# Probe failure must never block prompt build.
pass
@ -394,7 +406,7 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
except Exception:
active_profile = "default"
if active_profile == "default":
stable_parts.append(
post_workspace_parts.append(
"Active Hermes profile: default. Other profiles (if any) live "
"under " + str(get_hermes_home()) + "/profiles/<name>/. Each profile has its own "
"skills/, plugins/, cron/, and memories/ that affect a different "
@ -403,7 +415,7 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
"you to."
)
else:
stable_parts.append(
post_workspace_parts.append(
f"Active Hermes profile: {active_profile}. This session reads "
f"and writes {get_hermes_home()}/profiles/{active_profile}/. The default "
f"profile's data lives at {get_hermes_home()}/skills/, {get_hermes_home()}/plugins/, "
@ -449,11 +461,16 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
if platform_key == "tui" and _effective_hint:
_effective_hint = _tui_embedded_pane_clarifier(_effective_hint)
if _effective_hint:
stable_parts.append(_effective_hint)
post_workspace_parts.append(_effective_hint)
# ── Context tier (cwd-dependent, may change between sessions) ─
context_parts: List[str] = []
if coding_workspace_parts:
context_parts.extend(coding_workspace_parts)
context_parts.extend(coding_trailing_parts)
context_parts.extend(post_workspace_parts)
# Note: ephemeral_system_prompt is NOT included here. It's injected at
# API-call time only so it stays out of the cached/stored system prompt.
if system_message is not None:
@ -541,6 +558,7 @@ def build_system_prompt(agent: Any, system_message: Optional[str] = None) -> str
"""
parts = build_system_prompt_parts(agent, system_message=system_message)
joined = "\n\n".join(p for p in (parts["stable"], parts["context"], parts["volatile"]) if p)
agent._cached_system_prompt_static = parts["stable"]
# Surface context-file truncation warnings through the normal agent status
# channel so gateway/CLI users see them in chat instead of only in logs.
@ -557,6 +575,7 @@ def invalidate_system_prompt(agent: Any) -> None:
so the rebuilt prompt captures any writes from this session.
"""
agent._cached_system_prompt = None
agent._cached_system_prompt_static = None
if agent._memory_store:
agent._memory_store.load_from_disk()

20
tests/agent/test_anthropic_adapter.py
View file

@ -1134,6 +1134,26 @@ class TestConvertMessages:
assert isinstance(system, list)
assert system[0]["cache_control"] == {"type": "ephemeral"}
def test_static_system_prefix_markers_are_preserved(self):
messages = apply_anthropic_cache_control(
[
{"role": "system", "content": "stable\n\nsession context"},
{"role": "user", "content": "Hi"},
],
static_system_prefix="stable",
)
system, _ = convert_messages_to_anthropic(messages)
assert system == [
{"type": "text", "text": "stable", "cache_control": {"type": "ephemeral"}},
{
"type": "text",
"text": "\n\nsession context",
"cache_control": {"type": "ephemeral"},
},
]
def test_assistant_cache_control_blocks_are_preserved(self):
messages = apply_anthropic_cache_control([
{"role": "system", "content": "System prompt"},

48
tests/agent/test_prompt_caching.py
View file

@ -142,6 +142,54 @@ class TestApplyAnthropicCacheControl:
assert isinstance(sys_content, list)
assert sys_content[0]["cache_control"]["type"] == "ephemeral"
def test_static_system_prefix_gets_its_own_marker(self):
messages = [
{"role": "system", "content": "stable prefix\n\nper-session context"},
{"role": "user", "content": "old request"},
{"role": "assistant", "content": "old response"},
{"role": "user", "content": "new request"},
]
result = apply_anthropic_cache_control(
messages,
static_system_prefix="stable prefix",
)
system_blocks = result[0]["content"]
assert system_blocks == [
{
"type": "text",
"text": "stable prefix",
"cache_control": {"type": "ephemeral"},
},
{
"type": "text",
"text": "\n\nper-session context",
"cache_control": {"type": "ephemeral"},
},
]
assert result[1]["content"] == "old request"
assert result[2]["content"][0]["cache_control"] == {"type": "ephemeral"}
assert result[3]["content"][0]["cache_control"] == {"type": "ephemeral"}
def test_mismatched_static_prefix_uses_legacy_system_breakpoint(self):
messages = [
{"role": "system", "content": "current system prompt"},
{"role": "user", "content": "old request"},
{"role": "assistant", "content": "old response"},
{"role": "user", "content": "new request"},
]
result = apply_anthropic_cache_control(
messages,
static_system_prefix="stale system prompt",
)
assert len(result[0]["content"]) == 1
assert result[1]["content"][0]["cache_control"] == {"type": "ephemeral"}
assert result[2]["content"][0]["cache_control"] == {"type": "ephemeral"}
assert result[3]["content"][0]["cache_control"] == {"type": "ephemeral"}
def test_last_3_non_system_get_markers(self):
msgs = [
{"role": "system", "content": "System"},

89
tests/agent/test_system_prompt.py
View file

@ -1,9 +1,11 @@
"""Tests for agent/system_prompt.py — context-file cwd wiring."""
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
from agent.system_prompt import build_system_prompt_parts
from agent.system_prompt import build_system_prompt, build_system_prompt_parts
def _make_agent(**overrides):
@ -70,6 +72,16 @@ def _stable_prompt(agent):
return build_system_prompt_parts(agent)["stable"]
def _prompt_parts(agent):
with (
patch("run_agent.load_soul_md", return_value=""),
patch("run_agent.build_nous_subscription_prompt", return_value=""),
patch("run_agent.build_environment_hints", return_value=""),
patch("run_agent.build_context_files_prompt", return_value=""),
):
return build_system_prompt_parts(agent)
def _init_code_repo(path):
"""A git repo that actually holds code — the coding posture requires a source
file (or manifest), not a bare ``.git`` (a prose/notes repo stays general)."""
@ -84,9 +96,9 @@ class TestCodingContextBlock:
_init_code_repo(tmp_path)
monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
agent = _make_agent(valid_tool_names=["read_file"], platform="cli")
stable = _stable_prompt(agent)
assert "coding agent" in stable
assert "Workspace" in stable
parts = _prompt_parts(agent)
assert "coding agent" in parts["stable"]
assert "Workspace" in parts["context"]
def test_absent_when_off(self, monkeypatch, tmp_path):
_init_code_repo(tmp_path)
@ -104,6 +116,75 @@ class TestCodingContextBlock:
assert "coding agent" not in _stable_prompt(agent)
def test_build_system_prompt_records_stable_prefix():
agent = _make_agent()
with (
patch("run_agent.load_soul_md", return_value=""),
patch("run_agent.build_nous_subscription_prompt", return_value=""),
patch("run_agent.build_environment_hints", return_value=""),
patch("run_agent.build_context_files_prompt", return_value="context"),
):
prompt = build_system_prompt(agent)
assert prompt.startswith(agent._cached_system_prompt_static)
assert prompt[len(agent._cached_system_prompt_static):].startswith("\n\ncontext")
def test_coding_prompt_preserves_legacy_workspace_order(monkeypatch):
"""The cache split must not reorder the stored coding prompt."""
import agent.system_prompt as system_prompt
agent = _make_agent(
valid_tool_names=["read_file"],
_parallel_tool_call_guidance=False,
)
monkeypatch.setattr(system_prompt, "DEFAULT_AGENT_IDENTITY", "IDENTITY")
monkeypatch.setattr(system_prompt, "HERMES_AGENT_HELP_GUIDANCE", "HELP")
monkeypatch.setattr(system_prompt, "STEER_CHANNEL_NOTE", "STEER")
monkeypatch.setattr(system_prompt, "get_hermes_home", lambda: Path("/hermes"))
expected_profile = (
"Active Hermes profile: default. Other profiles (if any) live "
"under /hermes/profiles/<name>/. Each profile has its own skills/, "
"plugins/, cron/, and memories/ that affect a different session than "
"this one. Do not modify another profile's skills/plugins/cron/memories "
"unless the user explicitly directs you to."
)
expected = "\n\n".join((
"IDENTITY",
"HELP",
"STEER",
"CODING_STABLE",
"WORKSPACE",
"Operator instructions (from config):\nOPERATOR",
expected_profile,
"SYSTEM_MESSAGE",
"CONTEXT_FILES",
"Conversation started: Friday, January 02, 2026",
))
with (
patch("run_agent.load_soul_md", return_value=""),
patch("run_agent.build_nous_subscription_prompt", return_value=""),
patch("run_agent.build_environment_hints", return_value=""),
patch("run_agent.build_context_files_prompt", return_value="CONTEXT_FILES"),
patch(
"agent.coding_context.coding_system_prompt_parts",
return_value=(
["CODING_STABLE"],
["WORKSPACE"],
["Operator instructions (from config):\nOPERATOR"],
),
),
patch("agent.file_safety._resolve_active_profile_name", return_value="default"),
patch("hermes_time.now", return_value=datetime(2026, 1, 2)),
):
prompt = build_system_prompt(agent, system_message="SYSTEM_MESSAGE")
assert prompt == expected
assert agent._cached_system_prompt_static == "\n\n".join(expected.split("\n\n")[:4])
class TestTelegramRichMessagesHint:
"""Verify that TELEGRAM_RICH_MESSAGES_HINT is conditionally included."""

35
tests/run_agent/test_run_agent.py
View file

@ -4205,6 +4205,41 @@ class TestRunConversation:
assert result["final_response"] == "Final answer"
assert result["completed"] is True
def test_prompt_cache_marks_static_system_prefix_on_wire(self, agent):
self._setup_agent(agent)
agent._cached_system_prompt = "stable instructions\n\nsession context"
agent._cached_system_prompt_static = "stable instructions"
agent._use_prompt_caching = True
agent._use_native_cache_layout = False
agent._cache_ttl = "5m"
agent.client.chat.completions.create.return_value = _mock_response(
content="Final answer",
finish_reason="stop",
)
with (
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello")
assert result["completed"] is True
system = agent.client.chat.completions.create.call_args.kwargs["messages"][0]
assert system["role"] == "system"
assert system["content"] == [
{
"type": "text",
"text": "stable instructions",
"cache_control": {"type": "ephemeral"},
},
{
"type": "text",
"text": "\n\nsession context",
"cache_control": {"type": "ephemeral"},
},
]
def test_codex_content_filter_incomplete_routes_to_policy_fallback(self, agent):
self._setup_agent(agent)
agent.api_mode = "codex_responses"