Tune dialectic cache layout for Gemini

src/dialectic/core.py

@@ -37,6 +37,8 @@ from src.utils.clients import (
     honcho_llm_call,
 )
 from src.utils.formatting import format_new_turn_with_timestamp
+from src.utils.prompt_cache_layouts import build_system_messages
+from src.utils.types import SupportedProviders
 
 logger = logging.getLogger(__name__)
 
@@ -85,22 +87,41 @@ class DialecticAgent:
         self.observed_peer_card: list[str] | None = observed_peer_card
         self.metric_key: str | None = metric_key
         self.reasoning_level: ReasoningLevel = reasoning_level
+        self._provider: SupportedProviders = settings.DIALECTIC.LEVELS[
+            self.reasoning_level
+        ].PROVIDER
+        self._base_system_prompt: str = prompts.agent_system_prompt(
+            observer, observed, observer_peer_card, observed_peer_card
+        )
 
         # Initialize conversation history with system prompt
-        self.messages: list[dict[str, str]] = [
-            {
-                "role": "system",
-                "content": prompts.agent_system_prompt(
-                    observer, observed, observer_peer_card, observed_peer_card
-                ),
-            }
-        ]
+        self.messages: list[dict[str, str]] = self._build_system_messages()
         self._session_history_initialized: bool = False
         self._prefetched_conclusion_count: int = 0
         self._run_id: str = str(uuid.uuid4())[
             :8
         ]  # Always generate for event correlation
 
+    def _build_system_messages(
+        self,
+        session_history_section: str | None = None,
+    ) -> list[dict[str, str]]:
+        """Build provider-aware system messages for the dialectic prompt prefix."""
+        return build_system_messages(
+            self._provider,
+            self._base_system_prompt,
+            session_history_section,
+            wrapper_tag="rolling_history",
+        )
+
+    def _set_system_messages(self, session_history_section: str | None = None) -> None:
+        """Update system messages without disturbing conversation state."""
+        conversation_messages = [
+            message for message in self.messages if message.get("role") != "system"
+        ]
+        self.messages = self._build_system_messages(session_history_section)
+        self.messages.extend(conversation_messages)
+
     async def _initialize_session_history(self) -> None:
         """Fetch and inject session history into the system prompt if configured."""
         if self._session_history_initialized:
@@ -141,9 +162,7 @@ class DialecticAgent:
             "</session_history>"
         )
 
-        # Keep session history in its own system message so the stable base
-        # instructions can be cached independently of rolling session context.
-        self.messages.append({"role": "system", "content": session_history_section})
+        self._set_system_messages(session_history_section)
 
     async def _prefetch_relevant_observations(self, query: str) -> str | None:
         """

src/utils/prompt_cache_layouts.py

@@ -0,0 +1,85 @@
+from typing import Literal
+
+from src.utils.types import SupportedProviders
+
+PromptCacheLayoutMode = Literal[
+    "auto",
+    "split",
+    "split_reverse",
+    "merged_system",
+    "history_in_user",
+    "base_in_user",
+    "all_user",
+]
+
+_MERGED_SYSTEM_PROVIDERS: frozenset[SupportedProviders] = frozenset({"google"})
+
+
+def provider_default_prompt_cache_layout(
+    provider: SupportedProviders,
+) -> PromptCacheLayoutMode:
+    """Return the default prompt-cache layout for a provider."""
+    if provider in _MERGED_SYSTEM_PROVIDERS:
+        return "merged_system"
+    return "split"
+
+
+def resolve_prompt_cache_layout_mode(
+    provider: SupportedProviders,
+    layout_mode: PromptCacheLayoutMode,
+) -> PromptCacheLayoutMode:
+    """Resolve ``auto`` to the provider's default layout."""
+    if layout_mode == "auto":
+        return provider_default_prompt_cache_layout(provider)
+    return layout_mode
+
+
+def merge_system_prompt_with_rolling_context(
+    base_prompt: str,
+    rolling_context: str,
+    *,
+    wrapper_tag: str = "rolling_history",
+) -> str:
+    """Combine stable and rolling system context into a single system prompt."""
+    stable = base_prompt.strip()
+    rolling = rolling_context.strip()
+
+    if not rolling:
+        return stable
+
+    wrapped_context = f"<{wrapper_tag}>\n{rolling}\n</{wrapper_tag}>"
+    if not stable:
+        return wrapped_context
+    return f"{stable}\n\n{wrapped_context}"
+
+
+def build_system_messages(
+    provider: SupportedProviders,
+    base_prompt: str,
+    rolling_context: str | None = None,
+    *,
+    wrapper_tag: str = "rolling_history",
+) -> list[dict[str, str]]:
+    """Build provider-aware system messages for cacheable prompt prefixes."""
+    base_content = base_prompt.strip()
+    rolling_content = rolling_context.strip() if rolling_context else ""
+
+    if not rolling_content:
+        return [{"role": "system", "content": base_content}]
+
+    if provider_default_prompt_cache_layout(provider) == "merged_system":
+        return [
+            {
+                "role": "system",
+                "content": merge_system_prompt_with_rolling_context(
+                    base_content,
+                    rolling_content,
+                    wrapper_tag=wrapper_tag,
+                ),
+            }
+        ]
+
+    return [
+        {"role": "system", "content": base_content},
+        {"role": "system", "content": rolling_content},
+    ]

tests/utils/test_prompt_cache_layouts.py

@@ -0,0 +1,103 @@
+from unittest.mock import AsyncMock
+
+from src.dialectic.core import DialecticAgent
+from src.utils.prompt_cache_layouts import (
+    build_system_messages,
+    merge_system_prompt_with_rolling_context,
+    provider_default_prompt_cache_layout,
+)
+
+
+def test_provider_default_prompt_cache_layout_is_google_specific() -> None:
+    assert provider_default_prompt_cache_layout("google") == "merged_system"
+    assert provider_default_prompt_cache_layout("anthropic") == "split"
+    assert provider_default_prompt_cache_layout("openai") == "split"
+
+
+def test_build_system_messages_merges_google_history() -> None:
+    messages = build_system_messages(
+        "google",
+        "stable instructions",
+        "rolling history",
+        wrapper_tag="rolling_history",
+    )
+
+    assert messages == [
+        {
+            "role": "system",
+            "content": (
+                "stable instructions\n\n"
+                "<rolling_history>\nrolling history\n</rolling_history>"
+            ),
+        }
+    ]
+
+
+def test_build_system_messages_keeps_anthropic_history_separate() -> None:
+    messages = build_system_messages(
+        "anthropic",
+        "stable instructions",
+        "rolling history",
+        wrapper_tag="rolling_history",
+    )
+
+    assert messages == [
+        {"role": "system", "content": "stable instructions"},
+        {"role": "system", "content": "rolling history"},
+    ]
+
+
+def test_merge_system_prompt_with_rolling_context_strips_noise() -> None:
+    merged = merge_system_prompt_with_rolling_context(
+        "\n stable instructions \n",
+        "\n rolling history \n",
+        wrapper_tag="rolling_history",
+    )
+
+    assert (
+        merged
+        == "stable instructions\n\n<rolling_history>\nrolling history\n</rolling_history>"
+    )
+
+
+def test_dialectic_agent_rebuilds_google_system_messages() -> None:
+    agent = DialecticAgent(
+        db=AsyncMock(),
+        workspace_name="workspace",
+        session_name="session",
+        observer="Mira",
+        observed="Jon",
+        reasoning_level="low",
+    )
+
+    agent._provider = "google"
+    agent._set_system_messages("rolling history")
+
+    assert agent.messages == [
+        {
+            "role": "system",
+            "content": (
+                agent._base_system_prompt.strip()
+                + "\n\n<rolling_history>\nrolling history\n</rolling_history>"
+            ),
+        }
+    ]
+
+
+def test_dialectic_agent_rebuilds_anthropic_system_messages() -> None:
+    agent = DialecticAgent(
+        db=AsyncMock(),
+        workspace_name="workspace",
+        session_name="session",
+        observer="Mira",
+        observed="Jon",
+        reasoning_level="medium",
+    )
+
+    agent._provider = "anthropic"
+    agent._set_system_messages("rolling history")
+
+    assert agent.messages == [
+        {"role": "system", "content": agent._base_system_prompt.strip()},
+        {"role": "system", "content": "rolling history"},
+    ]