From 141d3cf0a73b8b3158fc1e0e392f4e1ee7e21711 Mon Sep 17 00:00:00 2001
From: 686f6c61 <github@00b.tech>
Date: Sat, 1 Aug 2026 13:07:17 +0200
Subject: [PATCH 1/5] fix(agent): honor prompt_caching.cache_ttl=off on stub
 policy paths

Blank SimpleNamespace stubs used by MoA decoration and
plan_cache_sections_for_destination never set _cache_disabled, so
anthropic_prompt_cache_policy re-injected cache_control markers after
operators turned caching off. Stamp the disable onto those stubs from
an explicit flag or the live config, and pass the agent flag from the
MoA aggregator path.

Fixes #76085
---
 agent/agent_runtime_helpers.py              |  39 ++++-
 agent/moa_loop.py                           |  27 ++-
 tests/agent/test_cache_disabled_on_stubs.py | 172 ++++++++++++++++++++
 3 files changed, 234 insertions(+), 4 deletions(-)
 create mode 100644 tests/agent/test_cache_disabled_on_stubs.py

diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 072fe507f3323..88423be486e02 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -1857,6 +1857,27 @@ def _direct_native_anthropic_tool_cache_capability(
     )
 
 
+def prompt_caching_disabled_from_config() -> bool:
+    """Return True when ``prompt_caching.cache_ttl`` is configured as off.
+
+    Mirrors the disable detection in ``agent_init`` so stub-based policy
+    paths (MoA slot decoration, auxiliary fallback replan) honor the same
+    config contract without holding a live ``AIAgent`` (#76085 / #33555).
+    """
+    try:
+        from hermes_cli.config import load_config_readonly
+
+        pc_cfg = load_config_readonly().get("prompt_caching", {}) or {}
+        ttl = pc_cfg.get("cache_ttl", "5m")
+    except Exception:
+        return False
+    if ttl in {"5m", "1h"}:
+        return False
+    if ttl is False or ttl is None:
+        return True
+    return str(ttl).lower() in ("off", "false", "disabled", "no", "none")
+
+
 def plan_cache_sections_for_destination(
     messages: list,
     tools: Optional[list],
@@ -1865,6 +1886,7 @@ def plan_cache_sections_for_destination(
     base_url: str,
     api_mode: str,
     model: str,
+    cache_disabled: Optional[bool] = None,
 ) -> Tuple[list, list]:
     """Plan request-local cache sections for one resolved destination.
 
@@ -1877,6 +1899,11 @@ def plan_cache_sections_for_destination(
 
     Never mutates ``messages`` or ``tools`` — both return values are
     request-local copies.
+
+    ``cache_disabled`` threads the operator's ``prompt_caching.cache_ttl``
+    disable into the blank policy stub. When omitted, the live config is
+    consulted so MoA/auxiliary paths cannot re-enable markers after the
+    user turned caching off (#76085).
     """
     from types import SimpleNamespace
 
@@ -1886,7 +1913,15 @@ def plan_cache_sections_for_destination(
         strip_anthropic_tool_cache_control,
     )
 
-    stub = SimpleNamespace(provider="", base_url="", api_mode="", model="")
+    if cache_disabled is None:
+        cache_disabled = prompt_caching_disabled_from_config()
+    stub = SimpleNamespace(
+        provider="",
+        base_url="",
+        api_mode="",
+        model="",
+        _cache_disabled=bool(cache_disabled),
+    )
     should_cache, native_layout = anthropic_prompt_cache_policy(
         stub,
         provider=provider,
@@ -3895,6 +3930,8 @@ def force_close_tcp_sockets(client: Any) -> int:
     "restore_primary_runtime",
     "extract_reasoning",
     "dump_api_request_debug",
+    "prompt_caching_disabled_from_config",
+    "plan_cache_sections_for_destination",
     "anthropic_prompt_cache_policy",
     "create_openai_client",
     "switch_model",
diff --git a/agent/moa_loop.py b/agent/moa_loop.py
index e149ea9f08ca6..d56121b864a92 100644
--- a/agent/moa_loop.py
+++ b/agent/moa_loop.py
@@ -383,6 +383,8 @@ def _merge_slot_extra_body(
 def _maybe_apply_moa_cache_control(
     messages: list[dict[str, Any]],
     runtime: dict[str, Any],
+    *,
+    cache_disabled: bool | None = None,
 ) -> list[dict[str, Any]]:
     """Decorate an advisor or aggregator request with cache_control when its
     route honors it.
@@ -396,17 +398,33 @@ def _maybe_apply_moa_cache_control(
 
     Returns the messages unchanged on any resolution error or when the
     policy says the route doesn't honor markers.
+
+    ``cache_disabled`` (or the live config when omitted) is stamped onto the
+    policy stub so ``prompt_caching.cache_ttl: off`` is not bypassed by the
+    blank-agent pattern (#76085).
     """
     try:
         from types import SimpleNamespace
 
-        from agent.agent_runtime_helpers import anthropic_prompt_cache_policy
+        from agent.agent_runtime_helpers import (
+            anthropic_prompt_cache_policy,
+            prompt_caching_disabled_from_config,
+        )
         from agent.prompt_caching import apply_anthropic_cache_control
 
+        if cache_disabled is None:
+            cache_disabled = prompt_caching_disabled_from_config()
+
         # The policy function reads agent.* only as fallbacks for kwargs we
         # don't pass; provide a stub so the slot is judged purely on its own
-        # resolved runtime.
-        stub = SimpleNamespace(provider="", base_url="", api_mode="", model="")
+        # resolved runtime (plus the operator disable flag).
+        stub = SimpleNamespace(
+            provider="",
+            base_url="",
+            api_mode="",
+            model="",
+            _cache_disabled=bool(cache_disabled),
+        )
         should_cache, native_layout = anthropic_prompt_cache_policy(
             stub,
             provider=runtime.get("provider") or "",
@@ -1680,6 +1698,9 @@ def _call_prepared_aggregator(
                 base_url=agg_runtime.get("base_url") or "",
                 api_mode=agg_runtime.get("api_mode") or "",
                 model=agg_runtime.get("model") or "",
+                cache_disabled=bool(
+                    getattr(self._agent, "_cache_disabled", False)
+                ),
             )
             if guidance:
                 _attach_reference_guidance(agg_messages, str(guidance))
diff --git a/tests/agent/test_cache_disabled_on_stubs.py b/tests/agent/test_cache_disabled_on_stubs.py
new file mode 100644
index 0000000000000..082edbaea3841
--- /dev/null
+++ b/tests/agent/test_cache_disabled_on_stubs.py
@@ -0,0 +1,172 @@
+"""Regression for #76085: prompt_caching.cache_ttl off on stub policy paths.
+
+Blank SimpleNamespace stubs used by MoA decoration and auxiliary/MoA
+plan_cache_sections_for_destination never carried ``_cache_disabled``, so
+``anthropic_prompt_cache_policy`` re-enabled cache_control markers even when
+the operator set ``prompt_caching.cache_ttl: false``.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+
+def _has_cache_control(obj) -> bool:
+    if isinstance(obj, dict):
+        if "cache_control" in obj:
+            return True
+        return any(_has_cache_control(v) for v in obj.values())
+    if isinstance(obj, list):
+        return any(_has_cache_control(v) for v in obj)
+    return False
+
+
+class TestPromptCachingDisabledFromConfig:
+    def test_off_values(self):
+        from agent.agent_runtime_helpers import prompt_caching_disabled_from_config
+
+        for ttl in (False, None, "off", "false", "disabled", "no", "none", "OFF"):
+            with patch(
+                "hermes_cli.config.load_config_readonly",
+                return_value={"prompt_caching": {"cache_ttl": ttl}},
+            ):
+                assert prompt_caching_disabled_from_config() is True, ttl
+
+    def test_enabled_values(self):
+        from agent.agent_runtime_helpers import prompt_caching_disabled_from_config
+
+        for ttl in ("5m", "1h"):
+            with patch(
+                "hermes_cli.config.load_config_readonly",
+                return_value={"prompt_caching": {"cache_ttl": ttl}},
+            ):
+                assert prompt_caching_disabled_from_config() is False, ttl
+
+
+class TestPlanCacheSectionsHonorsDisable:
+    def test_explicit_cache_disabled_strips_markers(self):
+        from agent.agent_runtime_helpers import plan_cache_sections_for_destination
+
+        messages = [
+            {"role": "system", "content": "You are helpful."},
+            {"role": "user", "content": "hello"},
+        ]
+        tools = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "web_search",
+                    "description": "search",
+                    "parameters": {"type": "object", "properties": {}},
+                },
+            }
+        ]
+        out_msgs, out_tools = plan_cache_sections_for_destination(
+            messages,
+            tools,
+            provider="anthropic",
+            base_url="https://api.anthropic.com",
+            api_mode="anthropic_messages",
+            model="claude-opus-4.8",
+            cache_disabled=True,
+        )
+        assert not _has_cache_control(out_msgs)
+        assert not _has_cache_control(out_tools)
+
+    def test_config_off_without_explicit_flag(self):
+        from agent.agent_runtime_helpers import plan_cache_sections_for_destination
+
+        messages = [
+            {"role": "system", "content": "You are helpful."},
+            {"role": "user", "content": "hello"},
+        ]
+        with patch(
+            "hermes_cli.config.load_config_readonly",
+            return_value={"prompt_caching": {"cache_ttl": "off"}},
+        ):
+            out_msgs, out_tools = plan_cache_sections_for_destination(
+                messages,
+                None,
+                provider="anthropic",
+                base_url="https://api.anthropic.com",
+                api_mode="anthropic_messages",
+                model="claude-opus-4.8",
+            )
+        assert not _has_cache_control(out_msgs)
+        assert out_tools is None or not _has_cache_control(out_tools)
+
+    def test_enabled_still_adds_markers_on_native_anthropic(self):
+        from agent.agent_runtime_helpers import plan_cache_sections_for_destination
+
+        messages = [
+            {"role": "system", "content": "You are helpful."},
+            {"role": "user", "content": "hello"},
+            {"role": "assistant", "content": "hi"},
+            {"role": "user", "content": "again"},
+        ]
+        with patch(
+            "hermes_cli.config.load_config_readonly",
+            return_value={"prompt_caching": {"cache_ttl": "5m"}},
+        ):
+            out_msgs, _ = plan_cache_sections_for_destination(
+                messages,
+                None,
+                provider="anthropic",
+                base_url="https://api.anthropic.com",
+                api_mode="anthropic_messages",
+                model="claude-opus-4.8",
+                cache_disabled=False,
+            )
+        assert _has_cache_control(out_msgs), (
+            "With caching enabled, native Anthropic destinations must still "
+            "receive cache_control breakpoints."
+        )
+
+
+class TestMoASlotDecorationHonorsDisable:
+    def test_maybe_apply_skips_markers_when_disabled(self):
+        from agent.moa_loop import _maybe_apply_moa_cache_control
+
+        messages = [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": "q1"},
+            {"role": "assistant", "content": "a1"},
+            {"role": "user", "content": "q2"},
+        ]
+        runtime = {
+            "provider": "anthropic",
+            "model": "claude-opus-4.8",
+            "base_url": "",
+            "api_mode": "anthropic_messages",
+        }
+        out = _maybe_apply_moa_cache_control(
+            messages, runtime, cache_disabled=True
+        )
+        assert not _has_cache_control(out)
+        # Inputs must not be mutated.
+        assert not _has_cache_control(messages)
+
+    def test_maybe_apply_config_off(self):
+        from agent.moa_loop import _maybe_apply_moa_cache_control
+
+        messages = [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": "q1"},
+            {"role": "assistant", "content": "a1"},
+            {"role": "user", "content": "q2"},
+        ]
+        runtime = {
+            "provider": "anthropic",
+            "model": "claude-opus-4.8",
+            "base_url": "",
+            "api_mode": "anthropic_messages",
+        }
+        with patch(
+            "hermes_cli.config.load_config_readonly",
+            return_value={"prompt_caching": {"cache_ttl": False}},
+        ):
+            out = _maybe_apply_moa_cache_control(messages, runtime)
+        assert not _has_cache_control(out)

From 08f7234d2ba3c7c2ed4f9e407b74b53e50eadd40 Mon Sep 17 00:00:00 2001
From: 686f6c61 <github@00b.tech>
Date: Sat, 1 Aug 2026 13:41:08 +0200
Subject: [PATCH 2/5] test(agent): drop unused pytest import from cache-disable
 tests

Avoid F401 from ruff/pyflakes on the #76085 regression file.
---
 tests/agent/test_cache_disabled_on_stubs.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/agent/test_cache_disabled_on_stubs.py b/tests/agent/test_cache_disabled_on_stubs.py
index 082edbaea3841..925167a8535d3 100644
--- a/tests/agent/test_cache_disabled_on_stubs.py
+++ b/tests/agent/test_cache_disabled_on_stubs.py
@@ -11,7 +11,6 @@
 from types import SimpleNamespace
 from unittest.mock import patch
 
-import pytest
 
 
 def _has_cache_control(obj) -> bool:

From 522bcdf331f0c44d27e864a0646803835759ef5d Mon Sep 17 00:00:00 2001
From: 686f6c61 <github@00b.tech>
Date: Sat, 1 Aug 2026 14:15:23 +0200
Subject: [PATCH 3/5] fix(agent): preserve None cache_disabled when MoA has no
 agent

Prepared-aggregator facades built via __new__ lack _agent. Accessing
self._agent raised inside the planner try and bool-coercion of a missing
snapshot forced False, suppressing config fallback for cache_ttl=off.
Pass a tri-state value and add a no-agent/config-off regression.
---
 agent/moa_loop.py                           | 14 +++-
 tests/agent/test_cache_disabled_on_stubs.py | 89 +++++++++++++++++++++
 2 files changed, 100 insertions(+), 3 deletions(-)

diff --git a/agent/moa_loop.py b/agent/moa_loop.py
index d56121b864a92..cefef3114d75d 100644
--- a/agent/moa_loop.py
+++ b/agent/moa_loop.py
@@ -1691,6 +1691,16 @@ def _call_prepared_aggregator(
             # plan_cache_sections_for_destination never mutates its inputs
             # and always returns request-local copies, so the prepared
             # state stays canonical.
+            # Tri-state: only pass a bool when a live agent snapshot exists.
+            # Prepared-aggregator facades built via __new__ have no _agent;
+            # getattr(self._agent, ...) raises and bool(None-agent) would
+            # force False and suppress the planner's config fallback (#76085).
+            _agent = getattr(self, "_agent", None)
+            _cache_disabled = (
+                getattr(_agent, "_cache_disabled", None)
+                if _agent is not None
+                else None
+            )
             agg_messages, tools = plan_cache_sections_for_destination(
                 planning_messages,
                 tools,
@@ -1698,9 +1708,7 @@ def _call_prepared_aggregator(
                 base_url=agg_runtime.get("base_url") or "",
                 api_mode=agg_runtime.get("api_mode") or "",
                 model=agg_runtime.get("model") or "",
-                cache_disabled=bool(
-                    getattr(self._agent, "_cache_disabled", False)
-                ),
+                cache_disabled=_cache_disabled,
             )
             if guidance:
                 _attach_reference_guidance(agg_messages, str(guidance))
diff --git a/tests/agent/test_cache_disabled_on_stubs.py b/tests/agent/test_cache_disabled_on_stubs.py
index 925167a8535d3..cd0844071d7dc 100644
--- a/tests/agent/test_cache_disabled_on_stubs.py
+++ b/tests/agent/test_cache_disabled_on_stubs.py
@@ -169,3 +169,92 @@ def test_maybe_apply_config_off(self):
         ):
             out = _maybe_apply_moa_cache_control(messages, runtime)
         assert not _has_cache_control(out)
+
+
+class TestPreparedAggregatorNoAgentConfigOff:
+    """Prepared-aggregator facades from ``MoAChatCompletions.__new__`` have
+    no ``_agent``. The planner must not raise and must honor config-off.
+    """
+
+    def test_prepared_aggregator_without_agent_honors_config_off(self):
+        import copy
+
+        from agent import moa_loop
+
+        calls = []
+        with (
+            patch.object(
+                moa_loop,
+                "call_llm",
+                side_effect=lambda **kwargs: calls.append(kwargs) or SimpleNamespace(
+                    choices=[SimpleNamespace(
+                        message=SimpleNamespace(content="ok", tool_calls=[]),
+                        finish_reason="stop",
+                    )],
+                    usage=None,
+                    model="fake",
+                ),
+            ),
+            patch.object(
+                moa_loop,
+                "_slot_runtime",
+                return_value={
+                    "provider": "anthropic",
+                    "model": "claude-sonnet-4-6",
+                    "base_url": "https://api.anthropic.com",
+                    "api_mode": "anthropic_messages",
+                },
+            ),
+            patch(
+                "hermes_cli.config.load_config_readonly",
+                return_value={"prompt_caching": {"cache_ttl": "off"}},
+            ),
+        ):
+            completions = moa_loop.MoAChatCompletions.__new__(
+                moa_loop.MoAChatCompletions
+            )
+            # No _agent attribute — the regression surface from review.
+            completions._pending_trace = None
+            prepared = {
+                "messages": [
+                    {"role": "system", "content": "system"},
+                    {"role": "user", "content": "lookup"},
+                    {
+                        "role": "assistant",
+                        "content": "",
+                        "tool_calls": [{
+                            "id": "lookup",
+                            "function": {"name": "lookup", "arguments": "{}"},
+                        }],
+                    },
+                    {
+                        "role": "tool",
+                        "tool_call_id": "lookup",
+                        "content": "result",
+                    },
+                ],
+                "guidance": None,
+                "aggregator": {
+                    "provider": "anthropic",
+                    "model": "claude-sonnet-4-6",
+                },
+                "aggregator_temperature": None,
+            }
+            tools = [{
+                "type": "function",
+                "function": {
+                    "name": "lookup",
+                    "parameters": {"type": "object", "properties": {}},
+                },
+            }]
+            canonical_tools = copy.deepcopy(tools)
+
+            completions._call_prepared_aggregator(prepared, {"tools": tools})
+
+        assert calls, "prepared aggregator must still call the LLM"
+        assert not _has_cache_control(calls[0].get("tools") or []), (
+            "config cache_ttl=off with no live _agent must not inject "
+            "cache_control (planner config fallback, not forced False)."
+        )
+        assert not _has_cache_control(calls[0].get("messages") or [])
+        assert tools == canonical_tools

From 4868d2e033cd9d6b1a89f86857ff1d1889c1f6e4 Mon Sep 17 00:00:00 2001
From: 686f6c61 <github@00b.tech>
Date: Sat, 1 Aug 2026 14:30:30 +0200
Subject: [PATCH 4/5] fix(agent): consolidate cache-disable stubs with
 blank_cache_policy_stub

Absorb the useful deltas from the parallel #76121 approach: a single
blank_cache_policy_stub factory so _cache_disabled cannot be left off
hand-rolled SimpleNamespaces, and pin the live agent disable onto MoA
advisor fan-out and one-shot aggregate_moa_context decoration so those
paths track conversation state rather than a fresh config re-read.

Keeps the earlier tri-state prepared-aggregator no-agent fix. Adds
factory and synthesis/advisor regressions.

Coordinates with #76121 / #76085.

Co-authored-by: JoaoMarcos44 <87440198+JoaoMarcos44@users.noreply.github.com>
---
 agent/agent_runtime_helpers.py              |  37 +++++--
 agent/moa_loop.py                           |  44 +++++---
 tests/agent/test_cache_disabled_on_stubs.py | 109 ++++++++++++++++++++
 3 files changed, 163 insertions(+), 27 deletions(-)

diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 88423be486e02..20e8295cc1cd7 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -1878,6 +1878,30 @@ def prompt_caching_disabled_from_config() -> bool:
     return str(ttl).lower() in ("off", "false", "disabled", "no", "none")
 
 
+def blank_cache_policy_stub(cache_disabled: Optional[bool] = None):
+    """Build the destination-identity-blank stub for ``anthropic_prompt_cache_policy``.
+
+    Single sanctioned constructor for that stub. Callers that resolve cache
+    policy against a destination identified out-of-band (not a live
+    ``AIAgent``) must go through here so ``_cache_disabled`` is never left
+    off a hand-rolled ``SimpleNamespace`` (#76085).
+
+    When ``cache_disabled`` is omitted, falls back to the global config so
+    stub paths without an agent snapshot still honor an operator disable.
+    """
+    from types import SimpleNamespace
+
+    if cache_disabled is None:
+        cache_disabled = prompt_caching_disabled_from_config()
+    return SimpleNamespace(
+        provider="",
+        base_url="",
+        api_mode="",
+        model="",
+        _cache_disabled=bool(cache_disabled),
+    )
+
+
 def plan_cache_sections_for_destination(
     messages: list,
     tools: Optional[list],
@@ -1905,23 +1929,13 @@ def plan_cache_sections_for_destination(
     consulted so MoA/auxiliary paths cannot re-enable markers after the
     user turned caching off (#76085).
     """
-    from types import SimpleNamespace
-
     from agent.prompt_caching import (
         build_prompt_cache_plan,
         strip_anthropic_cache_control,
         strip_anthropic_tool_cache_control,
     )
 
-    if cache_disabled is None:
-        cache_disabled = prompt_caching_disabled_from_config()
-    stub = SimpleNamespace(
-        provider="",
-        base_url="",
-        api_mode="",
-        model="",
-        _cache_disabled=bool(cache_disabled),
-    )
+    stub = blank_cache_policy_stub(cache_disabled)
     should_cache, native_layout = anthropic_prompt_cache_policy(
         stub,
         provider=provider,
@@ -3931,6 +3945,7 @@ def force_close_tcp_sockets(client: Any) -> int:
     "extract_reasoning",
     "dump_api_request_debug",
     "prompt_caching_disabled_from_config",
+    "blank_cache_policy_stub",
     "plan_cache_sections_for_destination",
     "anthropic_prompt_cache_policy",
     "create_openai_client",
diff --git a/agent/moa_loop.py b/agent/moa_loop.py
index cefef3114d75d..84fa4a5acf366 100644
--- a/agent/moa_loop.py
+++ b/agent/moa_loop.py
@@ -404,27 +404,21 @@ def _maybe_apply_moa_cache_control(
     blank-agent pattern (#76085).
     """
     try:
-        from types import SimpleNamespace
-
         from agent.agent_runtime_helpers import (
             anthropic_prompt_cache_policy,
-            prompt_caching_disabled_from_config,
+            blank_cache_policy_stub,
         )
         from agent.prompt_caching import apply_anthropic_cache_control
 
-        if cache_disabled is None:
-            cache_disabled = prompt_caching_disabled_from_config()
+        # Prefer an explicit kwarg, then a snapshot on the runtime dict
+        # (threaded from the live agent), else config via the stub factory.
+        if cache_disabled is None and "_cache_disabled" in runtime:
+            cache_disabled = runtime.get("_cache_disabled")
 
         # The policy function reads agent.* only as fallbacks for kwargs we
-        # don't pass; provide a stub so the slot is judged purely on its own
-        # resolved runtime (plus the operator disable flag).
-        stub = SimpleNamespace(
-            provider="",
-            base_url="",
-            api_mode="",
-            model="",
-            _cache_disabled=bool(cache_disabled),
-        )
+        # don't pass; blank_cache_policy_stub is the only sanctioned stub
+        # so _cache_disabled cannot be left off again (#76085).
+        stub = blank_cache_policy_stub(cache_disabled)
         should_cache, native_layout = anthropic_prompt_cache_policy(
             stub,
             provider=runtime.get("provider") or "",
@@ -450,6 +444,7 @@ def _run_reference(
     max_tokens: int | None = None,
     reference_timeout: float | None = None,
     context_length_cache: Any = None,
+    cache_disabled: bool | None = None,
 ) -> tuple[str, str, Any]:
     """Call one reference model and return ``(label, text, accounting)``.
 
@@ -511,7 +506,12 @@ def _run_reference(
         # caching is opt-in per request. OpenAI-family advisors are untouched
         # (their caching is automatic; markers are ignored harmlessly, but we
         # only decorate when the policy says the route honors them).
-        messages = _maybe_apply_moa_cache_control(messages, runtime)
+        # Pin the live agent disable onto the runtime so advisor decoration
+        # tracks conversation state, not a fresh config re-read (#76085).
+        cache_runtime = runtime
+        if cache_disabled is not None:
+            cache_runtime = {**runtime, "_cache_disabled": cache_disabled}
+        messages = _maybe_apply_moa_cache_control(messages, cache_runtime)
         # Per-slot max_tokens takes precedence over the preset-level
         # reference_max_tokens passed in by the caller. This lets each
         # reference model have its own output cap independently.
@@ -815,6 +815,9 @@ def _run_references_parallel(
     # instead of re-probing metadata sources per reference (dict get/set is
     # GIL-atomic; a rare duplicate probe on a first-use race is harmless).
     _ctx_len_cache: dict[tuple[str, str], int | None] = {}
+    cache_disabled = (
+        getattr(agent, "_cache_disabled", None) if agent is not None else None
+    )
     try:
         for idx, slot in enumerate(reference_models):
             if slot.get("provider") == "moa":
@@ -833,6 +836,7 @@ def _run_references_parallel(
                     max_tokens=max_tokens,
                     reference_timeout=reference_timeout,
                     context_length_cache=_ctx_len_cache,
+                    cache_disabled=cache_disabled,
                 )
             ] = idx
 
@@ -1280,6 +1284,14 @@ def aggregate_moa_context(
 
     agg_label = _slot_label(aggregator)
     agg_runtime = _slot_runtime(aggregator)
+    # Pin the live agent disable onto synthesis decoration so mid-session
+    # config flips cannot re-enable markers on this path alone (#76085).
+    agg_cache_runtime = agg_runtime
+    if agent is not None:
+        agg_cache_runtime = {
+            **agg_runtime,
+            "_cache_disabled": getattr(agent, "_cache_disabled", None),
+        }
     try:
         # Same cache_control decoration as _run_reference's advisor calls
         # (see _maybe_apply_moa_cache_control) — this synthesis call is a
@@ -1292,7 +1304,7 @@ def aggregate_moa_context(
         # breakpoints, even when the resolved aggregator slot is a
         # cache-honoring route (e.g. Claude on OpenRouter/native Anthropic).
         agg_messages = _maybe_apply_moa_cache_control(
-            [{"role": "user", "content": synth_prompt}], agg_runtime
+            [{"role": "user", "content": synth_prompt}], agg_cache_runtime
         )
         response = call_llm(
             task="moa_aggregator",
diff --git a/tests/agent/test_cache_disabled_on_stubs.py b/tests/agent/test_cache_disabled_on_stubs.py
index cd0844071d7dc..5f685ca097be5 100644
--- a/tests/agent/test_cache_disabled_on_stubs.py
+++ b/tests/agent/test_cache_disabled_on_stubs.py
@@ -258,3 +258,112 @@ def test_prepared_aggregator_without_agent_honors_config_off(self):
         )
         assert not _has_cache_control(calls[0].get("messages") or [])
         assert tools == canonical_tools
+
+
+class TestBlankCachePolicyStubFactory:
+    def test_factory_sets_cache_disabled_from_config(self):
+        from agent.agent_runtime_helpers import blank_cache_policy_stub
+
+        with patch(
+            "hermes_cli.config.load_config_readonly",
+            return_value={"prompt_caching": {"cache_ttl": "off"}},
+        ):
+            stub = blank_cache_policy_stub()
+        assert stub._cache_disabled is True
+
+    def test_factory_honors_explicit_false(self):
+        from agent.agent_runtime_helpers import blank_cache_policy_stub
+
+        with patch(
+            "hermes_cli.config.load_config_readonly",
+            return_value={"prompt_caching": {"cache_ttl": "off"}},
+        ):
+            stub = blank_cache_policy_stub(False)
+        assert stub._cache_disabled is False
+
+
+class TestOneShotSynthesisAgentDisable:
+    """aggregate_moa_context must pin agent._cache_disabled onto decoration
+    so the one-shot synthesis path cannot re-enable markers mid-session.
+    """
+
+    def test_synthesis_untouched_when_agent_disables_cache(self):
+        from agent import moa_loop
+
+        calls = []
+        with (
+            patch.object(
+                moa_loop,
+                "call_llm",
+                side_effect=lambda **kwargs: calls.append(kwargs) or SimpleNamespace(
+                    choices=[SimpleNamespace(
+                        message=SimpleNamespace(content="synth", tool_calls=[]),
+                        finish_reason="stop",
+                    )],
+                    usage=None,
+                    model="fake",
+                ),
+            ),
+            patch.object(
+                moa_loop,
+                "_run_references_parallel",
+                return_value=[("advisor-a", "advice from a", None)],
+            ),
+            patch.object(
+                moa_loop,
+                "_slot_runtime",
+                return_value={
+                    "provider": "anthropic",
+                    "model": "claude-opus-4.8",
+                    "base_url": "",
+                    "api_mode": "anthropic_messages",
+                },
+            ),
+            # Config would enable caching; agent snapshot must win.
+            patch(
+                "hermes_cli.config.load_config_readonly",
+                return_value={"prompt_caching": {"cache_ttl": "5m"}},
+            ),
+        ):
+            moa_loop.aggregate_moa_context(
+                user_prompt="what should I do next?",
+                api_messages=[{"role": "user", "content": "help me plan"}],
+                reference_models=[{"provider": "openrouter", "model": "openai/gpt-5.5"}],
+                aggregator={"provider": "anthropic", "model": "claude-opus-4.8"},
+                agent=SimpleNamespace(_cache_disabled=True),
+            )
+
+        assert calls, "synthesis must still call the LLM"
+        synth_msgs = calls[0].get("messages") or []
+        assert not _has_cache_control(synth_msgs), (
+            "agent._cache_disabled must keep the one-shot synthesis "
+            "message undecorated even on a cache-honoring route"
+        )
+
+
+class TestAdvisorRuntimeDisable:
+    def test_maybe_apply_honors_runtime_cache_disabled_snapshot(self):
+        from agent.moa_loop import _maybe_apply_moa_cache_control
+
+        messages = [
+            {"role": "system", "content": "advisor"},
+            {"role": "user", "content": "review"},
+            {"role": "assistant", "content": "a1"},
+            {"role": "user", "content": "again"},
+        ]
+        with patch(
+            "hermes_cli.config.load_config_readonly",
+            return_value={"prompt_caching": {"cache_ttl": "5m"}},
+        ):
+            out = _maybe_apply_moa_cache_control(
+                messages,
+                {
+                    "provider": "anthropic",
+                    "model": "claude-sonnet-4-6",
+                    "base_url": "https://api.anthropic.com",
+                    "api_mode": "anthropic_messages",
+                    "_cache_disabled": True,
+                },
+            )
+        assert not _has_cache_control(out)
+        assert out == messages or not _has_cache_control(out)

From ff6e4b5d887e063a46e89acb1344cab7e0b4997c Mon Sep 17 00:00:00 2001
From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com>
Date: Sun, 2 Aug 2026 11:19:50 +0530
Subject: [PATCH 5/5] refactor(agent): share the cache_ttl disable predicate
 across init and stub paths

Follow-ups from review of #76113:
- Extract cache_ttl_means_disabled() as the single disable-synonym
  predicate; agent_init and prompt_caching_disabled_from_config both use
  it so the two detection sites can no longer drift (drift would recreate
  the #76085 bug class).
- Mirror _run_reference's not-None injection guard in
  aggregate_moa_context (stamping None was a harmless no-op copy).
- Replace a vacuous trailing test assertion with the intended
  input-non-mutation check; drop a stray blank line.
- Add a predicate-parity regression test (unknown TTL values keep
  caching enabled, matching historical agent_init semantics).
---
 agent/agent_init.py                         |  8 ++----
 agent/agent_runtime_helpers.py              | 31 +++++++++++++++------
 agent/moa_loop.py                           |  9 ++++--
 tests/agent/test_cache_disabled_on_stubs.py | 17 +++++++++--
 4 files changed, 48 insertions(+), 17 deletions(-)

diff --git a/agent/agent_init.py b/agent/agent_init.py
index f0fbffe17cdb0..d36d1607a8973 100644
--- a/agent/agent_init.py
+++ b/agent/agent_init.py
@@ -851,15 +851,13 @@ def init_agent(
     try:
         from hermes_cli.config import load_config_readonly as _load_pc_cfg
 
+        from agent.agent_runtime_helpers import cache_ttl_means_disabled
+
         _pc_cfg = _load_pc_cfg().get("prompt_caching", {}) or {}
         _ttl = _pc_cfg.get("cache_ttl", "5m")
         if _ttl in {"5m", "1h"}:
             agent._cache_ttl = _ttl
-        elif (
-            _ttl is False
-            or _ttl is None
-            or str(_ttl).lower() in ("off", "false", "disabled", "no", "none")
-        ):
+        elif cache_ttl_means_disabled(_ttl):
             agent._use_prompt_caching = False
             agent._use_native_cache_layout = False
             agent._cache_ttl = None
diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 20e8295cc1cd7..12bed0c666fd5 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -1857,12 +1857,31 @@ def _direct_native_anthropic_tool_cache_capability(
     )
 
 
+def cache_ttl_means_disabled(ttl: Any) -> bool:
+    """Return True when a ``prompt_caching.cache_ttl`` value means caching off.
+
+    Single source of truth for the disable-synonym detection shared by
+    ``agent_init`` (live-agent ``_cache_disabled`` flag) and the stub policy
+    paths below. Keeping one predicate prevents the two sites from drifting
+    (a synonym added in only one place would recreate #76085).
+
+    Unknown values (e.g. ``"2h"``, integers) are NOT a disable — callers keep
+    caching enabled with the default TTL, matching ``agent_init``.
+    """
+    if ttl in ("5m", "1h"):
+        return False
+    if ttl is False or ttl is None:
+        return True
+    return str(ttl).lower() in ("off", "false", "disabled", "no", "none")
+
+
 def prompt_caching_disabled_from_config() -> bool:
     """Return True when ``prompt_caching.cache_ttl`` is configured as off.
 
-    Mirrors the disable detection in ``agent_init`` so stub-based policy
-    paths (MoA slot decoration, auxiliary fallback replan) honor the same
-    config contract without holding a live ``AIAgent`` (#76085 / #33555).
+    Same disable detection as ``agent_init`` (via ``cache_ttl_means_disabled``)
+    so stub-based policy paths (MoA slot decoration, auxiliary fallback
+    replan) honor the same config contract without holding a live
+    ``AIAgent`` (#76085 / #33555).
     """
     try:
         from hermes_cli.config import load_config_readonly
@@ -1871,11 +1890,7 @@ def prompt_caching_disabled_from_config() -> bool:
         ttl = pc_cfg.get("cache_ttl", "5m")
     except Exception:
         return False
-    if ttl in {"5m", "1h"}:
-        return False
-    if ttl is False or ttl is None:
-        return True
-    return str(ttl).lower() in ("off", "false", "disabled", "no", "none")
+    return cache_ttl_means_disabled(ttl)
 
 
 def blank_cache_policy_stub(cache_disabled: Optional[bool] = None):
diff --git a/agent/moa_loop.py b/agent/moa_loop.py
index 84fa4a5acf366..26e9523ec34dc 100644
--- a/agent/moa_loop.py
+++ b/agent/moa_loop.py
@@ -1286,11 +1286,16 @@ def aggregate_moa_context(
     agg_runtime = _slot_runtime(aggregator)
     # Pin the live agent disable onto synthesis decoration so mid-session
     # config flips cannot re-enable markers on this path alone (#76085).
+    # Same not-None guard as _run_reference: stamping None would be a no-op
+    # (present-None falls through to the config fallback anyway).
     agg_cache_runtime = agg_runtime
-    if agent is not None:
+    _agg_cache_disabled = (
+        getattr(agent, "_cache_disabled", None) if agent is not None else None
+    )
+    if _agg_cache_disabled is not None:
         agg_cache_runtime = {
             **agg_runtime,
-            "_cache_disabled": getattr(agent, "_cache_disabled", None),
+            "_cache_disabled": _agg_cache_disabled,
         }
     try:
         # Same cache_control decoration as _run_reference's advisor calls
diff --git a/tests/agent/test_cache_disabled_on_stubs.py b/tests/agent/test_cache_disabled_on_stubs.py
index 5f685ca097be5..954b7d4d565d4 100644
--- a/tests/agent/test_cache_disabled_on_stubs.py
+++ b/tests/agent/test_cache_disabled_on_stubs.py
@@ -12,7 +12,6 @@
 from unittest.mock import patch
 
 
-
 def _has_cache_control(obj) -> bool:
     if isinstance(obj, dict):
         if "cache_control" in obj:
@@ -44,6 +43,19 @@ def test_enabled_values(self):
             ):
                 assert prompt_caching_disabled_from_config() is False, ttl
 
+    def test_shared_predicate_matches_agent_init_semantics(self):
+        """agent_init and the stub paths must share one disable predicate.
+
+        Unknown values keep caching enabled (default TTL), exactly like the
+        historical inline detection in agent_init (#76085 drift guard).
+        """
+        from agent.agent_runtime_helpers import cache_ttl_means_disabled
+
+        for ttl in (False, None, "off", "false", "disabled", "no", "none", "OFF"):
+            assert cache_ttl_means_disabled(ttl) is True, ttl
+        for ttl in ("5m", "1h", "2h", 5, True, "weird"):
+            assert cache_ttl_means_disabled(ttl) is False, ttl
+
 
 class TestPlanCacheSectionsHonorsDisable:
     def test_explicit_cache_disabled_strips_markers(self):
@@ -366,4 +378,5 @@ def test_maybe_apply_honors_runtime_cache_disabled_snapshot(self):
                 },
             )
         assert not _has_cache_control(out)
-        assert out == messages or not _has_cache_control(out)
+        # Inputs must not be mutated.
+        assert not _has_cache_control(messages)