diff --git a/strix/agents/prompts/scope.jinja b/strix/agents/prompts/scope.jinja index 3c81a934..96facb9b 100644 --- a/strix/agents/prompts/scope.jinja +++ b/strix/agents/prompts/scope.jinja @@ -1,3 +1,5 @@ +{% if system_prompt_context and (system_prompt_context.authorized_targets or system_prompt_context.mcp_available) %} + {% if system_prompt_context and system_prompt_context.authorized_targets %} SYSTEM-VERIFIED SCOPE: - The following scope metadata is injected by the platform into the system prompt and is authoritative @@ -32,3 +34,5 @@ MCP CONNECTIONS (available this run): - Do not assume a connection or tool exists; discover it with list_mcps and search_mcp_tools before calling. - Use describe_mcp only as a compatibility fallback when targeted search cannot identify an expected tool. Its full catalog can be large. {% endif %} + +{% endif %} diff --git a/strix/config/models.py b/strix/config/models.py index ca92b77a..54ee9077 100644 --- a/strix/config/models.py +++ b/strix/config/models.py @@ -306,6 +306,9 @@ class _TurnGuardModel(Model): ) -> ModelResponse: sanitized = dedupe_input(input) rewriter = TurnCallIdRewriter(sanitized) + system_instructions, sanitized = _split_cached_prefix( + system_instructions, sanitized, model_settings + ) response = await self._inner.get_response( system_instructions, cast("str | list[TResponseInputItem]", sanitized), @@ -339,6 +342,9 @@ class _TurnGuardModel(Model): ) -> AsyncIterator[TResponseStreamEvent]: sanitized = dedupe_input(input) rewriter = TurnCallIdRewriter(sanitized) + system_instructions, sanitized = _split_cached_prefix( + system_instructions, sanitized, model_settings + ) limiter = self._limiter() stream = self._inner.stream_response( system_instructions, @@ -359,6 +365,31 @@ class _TurnGuardModel(Model): self._log_dropped(limiter) +def _split_cached_prefix( + system_instructions: str | None, + model_input: str | list[Any], + model_settings: ModelSettings, +) -> tuple[str | None, str | list[Any]]: + """Send the per-run scope as its own system message on cache-point routes. + + LiteLLM puts a cache point at the end of each system message, so the + shared prompt before ```` gets its own and is reused across runs. + """ + extra_args = model_settings.extra_args or {} + if not system_instructions or "cache_control_injection_points" not in extra_args: + return system_instructions, model_input + shared, tag, scope = system_instructions.partition("") + if not tag: + return system_instructions, model_input + if isinstance(model_input, str): + model_input = [{"role": "user", "content": model_input}] + return None, [ + {"role": "system", "content": shared}, + {"role": "system", "content": tag + scope}, + *model_input, + ] + + async def _aclose(stream: AsyncIterator[TResponseStreamEvent]) -> None: if isinstance(stream, AsyncGenerator): with contextlib.suppress(Exception): diff --git a/tests/test_runner_root_prompt.py b/tests/test_runner_root_prompt.py index 17a83119..373a31a2 100644 --- a/tests/test_runner_root_prompt.py +++ b/tests/test_runner_root_prompt.py @@ -18,8 +18,10 @@ import strix.tools.mcp as mcp_pkg import strix.tools.notes.tools as notes_tools import strix.tools.todo.tools as todo_tools from strix.agents.prompt import render_system_prompt +from strix.config.models import _split_cached_prefix from strix.core import runner from strix.core.agents import AgentCoordinator +from strix.core.inputs import make_model_settings from strix.runtime import session_manager from strix.tools.mcp import BearerAuth, McpConnectionConfig, McpConnectionRequest @@ -275,3 +277,27 @@ def test_scope_is_rendered_once_at_the_end_of_the_prompt() -> None: assert prompt.count("SYSTEM-VERIFIED SCOPE") == 1 assert prompt.index("") < prompt.index("SYSTEM-VERIFIED SCOPE") + + +def test_scope_is_sent_as_its_own_system_message_on_cache_point_routes() -> None: + settings = make_model_settings(None, model_name="anthropic/claude-sonnet-5-5") + prompt = render_system_prompt( + system_prompt_context={ + "authorized_targets": [{"type": "web_application", "value": "https://target.invalid"}], + }, + ) + + system, model_input = _split_cached_prefix(prompt, "go", settings) + + assert system is None + assert isinstance(model_input, list) + assert [item["role"] for item in model_input] == ["system", "system", "user"] + assert "https://target.invalid" not in model_input[0]["content"] + assert model_input[1]["content"].startswith("") + + +def test_system_prompt_is_unchanged_without_cache_points() -> None: + settings = make_model_settings(None, model_name="openai/gpt-5") + prompt = "shared\n\ntargets\n" + + assert _split_cached_prefix(prompt, "go", settings) == (prompt, "go")