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

@@ -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

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):

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

tests/agent/test_cache_disabled_on_stubs.py

@@ -12,7 +12,6 @@ from types import SimpleNamespace
 from unittest.mock import patch
 
 
-
 def _has_cache_control(obj) -> bool:
     if isinstance(obj, dict):
         if "cache_control" in obj:
@@ -44,6 +43,19 @@ class TestPromptCachingDisabledFromConfig:
             ):
                 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 @@ class TestAdvisorRuntimeDisable:
                 },
             )
         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)