Fix #285: make 1h prompt caching reachable via config

src/context_system/prompt_assembly.py

@@ -687,11 +687,11 @@ def build_full_system_prompt_blocks(
     # WI-2.2: TTL selector. ``should_1h_cache_ttl(query_source)`` returns
     # True only when (a) the user is 1h-eligible per the latched
     # evaluation in cache_state.evaluate_prompt_cache_1h_eligibility, AND
-    # (b) the query source is in the GrowthBook-populated allowlist.
+    # (b) the query source is in the configured allowlist (#285:
+    # settings.prompt_cache_1h_sources / CLAWCODEX_PROMPT_CACHE_1H_SOURCES,
+    # installed at session start by initialize_prompt_cache_state).
     # When either condition is False, fall back to "5m" — the safe-default
-    # TTL that Phase 1 already engaged. The allowlist is empty by default
-    # (no GrowthBook port yet), so this defaults to "5m" universally until
-    # a future WI populates it.
+    # TTL that Phase 1 already engaged.
     from src.state.cache_state import should_1h_cache_ttl, should_use_global_cache_scope
     ttl = "1h" if should_1h_cache_ttl(query_source) else "5m"
 

src/init.py

@@ -177,6 +177,18 @@ def run_pre_action(args: object) -> None:
         # workspace through.
         set_session_trust_accepted(False)
 
+    # #285: latch the 1h prompt-cache eligibility decision and install
+    # the configured query-source allowlist — without this the latch
+    # stays None and 1h caching is permanently dormant. Fail-soft.
+    try:
+        from src.state.session_start import initialize_prompt_cache_state
+
+        initialize_prompt_cache_state()
+    except Exception:
+        logging.getLogger(__name__).debug(
+            "prompt-cache state init failed", exc_info=True
+        )
+
     profile_checkpoint("pre_action_end")
 
 

src/plugins/mcp_integration.py

@@ -2,10 +2,13 @@
 
 import logging
 from dataclasses import dataclass, field
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from .types import LoadedPlugin, PluginManifest
 
+if TYPE_CHECKING:
+    from src.services.mcp.types import McpServerConfig
+
 logger = logging.getLogger(__name__)
 
 
@@ -23,6 +26,13 @@ class McpPluginWrapper:
     server_name: str
     tools: list[McpPluginTool] = field(default_factory=list)
     connected: bool = False
+    # #286: the launch config, set at registration time. With it, the
+    # plugin-scope loader (services/mcp/config.get_managed_mcp_configs)
+    # surfaces this server into the config merge — per-name lookup,
+    # /mcp listings, and scope-policy filtering — like every other
+    # scope. None (legacy registrations) keeps the wrapper tools-only
+    # and invisible to the merge.
+    server_config: "McpServerConfig | None" = None
 
 
 _mcp_plugins: dict[str, McpPluginWrapper] = {}
@@ -33,7 +43,9 @@ def wrap_mcp_server_as_plugin(
     tools: list[dict[str, Any]],
     *,
     description: str = "",
+    server_config: "McpServerConfig | None" = None,
 ) -> McpPluginWrapper:
+    server_type = getattr(server_config, "type", None) or "stdio"
     manifest = PluginManifest(
         name=f"mcp-{server_name}",
         description=description or f"MCP server: {server_name}",
@@ -45,7 +57,7 @@ def wrap_mcp_server_as_plugin(
         manifest=manifest,
         source=f"mcp:{server_name}",
         enabled=True,
-        mcp_servers={server_name: {"type": "stdio"}},
+        mcp_servers={server_name: {"type": server_type}},
     )
 
     mcp_tools: list[McpPluginTool] = []
@@ -62,6 +74,7 @@ def wrap_mcp_server_as_plugin(
         server_name=server_name,
         tools=mcp_tools,
         connected=True,
+        server_config=server_config,
     )
 
     _mcp_plugins[server_name] = wrapper

src/services/mcp/config.py

@@ -499,6 +499,16 @@ def get_mcp_configs_by_scope(
 
 
 def get_mcp_config_by_name(name: str) -> ScopedMcpServerConfig | None:
+    # Enterprise lockdown parity with the aggregate path (#286): when a
+    # managed-mcp.json exists, get_all_mcp_configs returns enterprise
+    # servers ONLY — the by-name resolve (reconnect, OAuth flows) must
+    # not be a side door that hands out user/project/local/managed/
+    # dynamic configs the merge excluded (same reasoning as the C7
+    # approval gate below).
+    if _does_enterprise_mcp_config_exist():
+        servers, _ = get_mcp_configs_by_scope("enterprise")
+        return servers.get(name)
+
     # Order: highest-trust → lowest-trust. Enterprise managed wins over
     # user, which wins over project, which wins over local.
     for scope in ("enterprise", "user", "project", "local"):
@@ -573,26 +583,36 @@ def get_dynamic_mcp_configs() -> dict[str, ScopedMcpServerConfig]:
 def get_managed_mcp_configs() -> dict[str, ScopedMcpServerConfig]:
     """Return plugin-provided MCP server configs (``managed`` scope).
 
-    Phase 7 WI-7.4 (gap #9 subset). The integration point with the plugin
-    layer at ``src/plugins/mcp_integration.py`` is the
-    ``McpPluginWrapper`` registry — but as of today, ``McpPluginWrapper``
-    does not carry an ``McpServerConfig`` (only a ``server_name``,
-    ``plugin``, ``tools``, ``connected``). There is therefore nothing to
-    return: the wrapper holds tool metadata, not the launch config.
-
-    Returning ``{}`` here keeps the merge surface stable: callers (the
-    ``get_all_mcp_configs`` aggregator and ``get_mcp_config_by_name``)
-    can ask for managed configs without crashing, and the moment the
-    plugin layer is extended to carry an ``McpServerConfig`` per wrapper,
-    this loader can read it via ``wrapper.config`` (or whatever the
-    extended schema names it) and propagate.
-
-    TODO(Phase 7 follow-up): extend ``McpPluginWrapper`` with a
-    ``server_config: McpServerConfig`` field, set at registration time,
-    and surface it here. Until that lands, plugin-provided MCP servers
-    cannot participate in the per-name lookup or the merge.
+    Phase 7 WI-7.4 (gap #9 subset), wired by #286: reads the
+    ``McpPluginWrapper`` registry at ``src/plugins/mcp_integration.py``
+    and surfaces every wrapper that carries a ``server_config`` into the
+    merge — per-name lookup, the ``get_all_mcp_configs`` aggregator, and
+    ``filter_mcp_servers_by_policy`` (``allow_managed_only_mcp`` counts
+    these as managed) all see them like any other scope. Legacy
+    tools-only registrations (``server_config=None``) stay invisible to
+    the merge, exactly as before.
+
+    ``plugin_source`` records the providing plugin's name so listings
+    and dedup notices can attribute the entry.
     """
-    return {}
+    try:
+        # Lazy import: services/mcp must not import the plugins package
+        # at module level (plugins imports services.mcp.types).
+        from src.plugins.mcp_integration import get_all_mcp_plugins
+    except ImportError:
+        logger.warning("plugin registry unavailable; no managed MCP servers")
+        return {}
+
+    servers: dict[str, ScopedMcpServerConfig] = {}
+    for wrapper in get_all_mcp_plugins():
+        if wrapper.server_config is None:
+            continue
+        servers[wrapper.server_name] = ScopedMcpServerConfig(
+            config=wrapper.server_config,
+            scope="managed",
+            plugin_source=wrapper.plugin.name,
+        )
+    return servers
 
 
 def get_all_mcp_configs() -> tuple[dict[str, ScopedMcpServerConfig], list[ValidationError]]:
@@ -687,6 +707,33 @@ def get_all_mcp_configs() -> tuple[dict[str, ScopedMcpServerConfig], list[Valida
                 f"duplicate of manual server {rec.get('duplicateOf')!r}."
             )
 
+    # #286: plugin/manual dedup — a plugin server whose launch signature
+    # duplicates a manual entry is suppressed so the operator's explicit
+    # config wins (same policy as the claudeai dedup above — including
+    # the disabled-server carve-out: a DISABLED manual entry must not
+    # suppress its plugin twin, or disabling the manual copy would leave
+    # the user with zero working servers).
+    if managed_servers:
+        managed_only = {k: v for k, v in merged.items() if v.scope == "managed"}
+        manual_only = {
+            k: v
+            for k, v in merged.items()
+            if v.scope in ("user", "project", "local")
+            and not is_mcp_server_disabled(k)
+        }
+        kept_managed, suppressed_plugin = dedup_plugin_mcp_servers(
+            managed_only, manual_only
+        )
+        merged = {
+            **{k: v for k, v in merged.items() if v.scope != "managed"},
+            **kept_managed,
+        }
+        for rec in suppressed_plugin:
+            dedup_notice_strings.append(
+                f"Plugin MCP server {rec.get('name')!r} suppressed; "
+                f"duplicate of {rec.get('duplicateOf')!r}."
+            )
+
     filtered, notices = filter_mcp_servers_by_policy(merged)
 
     all_errors = (

src/settings/types.py

@@ -157,6 +157,14 @@ class SettingsSchema:
     # provider-scoped in a multi-provider config).
     model_provider: str = ""
 
+    # Query sources eligible for 1h prompt-cache TTL (#285) — the
+    # config-backed replacement for the TS GrowthBook allowlist
+    # (e.g. ["repl_main_thread"]). Empty = 1h caching dormant.
+    # CLAWCODEX_PROMPT_CACHE_1H_SOURCES (comma-separated) overrides when
+    # SET, including set-but-empty as a kill switch. Like every list
+    # setting, a more specific config layer REPLACES (not extends) this.
+    prompt_cache_1h_sources: list[str] = field(default_factory=list)
+
     # Disable dynamic workflows (also honored via CLAUDE_CODE_DISABLE_WORKFLOWS
     # and the camelCase ``disableWorkflows`` JSON key). See src/workflow/gating.py.
     disable_workflows: bool = False

src/state/cache_state.py

@@ -11,18 +11,24 @@
 Sacrificing mid-session toggleability buys cache stability worth far more in
 dollars per turn.
 
-Per the Phase 2 audit (M9-resolved):
-  * ``prompt_cache_1h_eligible`` — wired here; consumed by WI-2.2's
+Per the Phase 2 audit (M9-resolved; #285 wired the 1h path):
+  * ``prompt_cache_1h_eligible`` — latched at session start by
+    ``src/state/session_start.initialize_prompt_cache_state`` (called
+    from ``init.pre_action``); consumed by WI-2.2's
     ``should_1h_cache_ttl`` selector.
+  * ``prompt_cache_1h_allowlist`` — populated at the same session-start
+    site from ``settings.prompt_cache_1h_sources`` /
+    ``CLAWCODEX_PROMPT_CACHE_1H_SOURCES`` (the config-backed,
+    non-GrowthBook channel — #285).
   * ``fast_mode_header_latched`` — wired by ``src/utils/fast_mode.py`` on
     first true result of ``is_fast_mode_enabled()``.
-  * ``afk_mode_header_latched`` — DEAD STORE today (no AFK toggle in
-    Python TUI yet). Future TUI WI must add the trigger.
-  * ``cache_editing_header_latched`` — DEAD STORE today (cache-editing is
-    a TS GrowthBook treatment with no Python equivalent yet).
-  * ``thinking_clear_latched`` — DEAD STORE today (thinking-mode-flip event
-    is not exposed by ``src/utils/effort.py`` in the form needed for this
-    latch). Future WI must surface the event.
+  * ``afk_mode_header_latched`` — DEAD STORE: its source feature (the TUI
+    AFK toggle) is not built. Wire when that feature lands (#285 audit).
+  * ``cache_editing_header_latched`` — DEAD STORE: cache-editing is a TS
+    GrowthBook treatment with no Python equivalent. Wire with that port.
+  * ``thinking_clear_latched`` — DEAD STORE: the port has no extended-
+    thinking request parameter, so a thinking-flip-after-cache-miss event
+    cannot exist yet. Wire when thinking lands.
 """
 
 from __future__ import annotations
@@ -41,6 +47,7 @@
     "get_prompt_cache_1h_allowlist",
     "get_prompt_cache_1h_eligible",
     "is_first_party_provider",
+    "populate_prompt_cache_1h_allowlist",
     "reset_for_test_only",
     "should_1h_cache_ttl",
     "should_use_global_cache_scope",
@@ -78,10 +85,11 @@ class BetaHeaderLatches:
     # ``prompt_cache_1h_eligible`` is True, the per-call decision still
     # requires the ``query_source`` to appear in this list — mirrors TS
     # GrowthBook config at ``services/api/claude.ts:430-438``.
-    # Default empty list = no source emits 1h. Population of this list is
-    # left to a future WI that ports the GrowthBook integration; for now
-    # the allowlist remains empty and 1h caching is dormant (5m caching
-    # still works because Phase 1 already engaged it).
+    # Default empty list = no source emits 1h. Populated once per session
+    # by ``populate_prompt_cache_1h_allowlist`` from configuration
+    # (settings.prompt_cache_1h_sources / CLAWCODEX_PROMPT_CACHE_1H_SOURCES
+    # — the non-GrowthBook channel, #285); unconfigured installs stay
+    # dormant (5m caching still works from Phase 1).
     prompt_cache_1h_allowlist: list[str] = field(default_factory=list)
 
     # Toggle latches. Set on first toggle event; never reset.
@@ -133,15 +141,32 @@ def get_prompt_cache_1h_eligible() -> bool | None:
 def get_prompt_cache_1h_allowlist() -> list[str]:
     """Read the 1h-cache query-source allowlist.
 
-    Returns a copy to discourage caller mutation. The allowlist is
-    populated by future GrowthBook-port work (currently always empty in
-    the open build). Plain getter for parity with TS
-    ``getPromptCache1hAllowlist`` (``bootstrap/state.ts:1579``). **No
-    setter is exposed**.
+    Returns a copy to discourage caller mutation. Populated once per
+    session from configuration via
+    ``populate_prompt_cache_1h_allowlist`` (#285 — the non-GrowthBook
+    config channel). Plain getter for parity with TS
+    ``getPromptCache1hAllowlist`` (``bootstrap/state.ts:1579``).
     """
     return list(_LATCHES.prompt_cache_1h_allowlist)
 
 
+def populate_prompt_cache_1h_allowlist(sources: list[str]) -> bool:
+    """Populate the 1h-cache allowlist ONCE per session (#285).
+
+    The config-backed replacement for the TS GrowthBook channel: the
+    session-start wiring reads the configured query sources and installs
+    them here. Sticky like every other field in this module — a
+    non-empty allowlist is never replaced mid-session (a flip would bust
+    the cached prompt prefix this module exists to protect). Returns
+    True when the list was installed.
+    """
+    cleaned = [s.strip() for s in sources if isinstance(s, str) and s.strip()]
+    if not cleaned or _LATCHES.prompt_cache_1h_allowlist:
+        return False
+    _LATCHES.prompt_cache_1h_allowlist = cleaned
+    return True
+
+
 def evaluate_prompt_cache_1h_eligibility(
     *,
     is_ant_user: bool,
@@ -165,16 +190,14 @@ def evaluate_prompt_cache_1h_eligibility(
     keeps 1h caching dormant. When the porting WI for these inputs lands,
     1h caching activates without requiring code changes here.
 
-    **Status (Phase 2):** this primitive is implemented but has NO
-    production caller today. ``grep -rn "evaluate_prompt_cache_1h_eligibility"
-    src/`` returns only the definition. The 1h cache path is therefore
-    end-to-end dormant: ``prompt_cache_1h_eligible`` stays at ``None``,
-    ``should_1h_cache_ttl`` always returns False, every cache_control
-    emits ``ttl: '5m'``. Activating 1h requires (a) porting the user-type
-    /subscription/overage signals, (b) calling this function with real
-    inputs at session start, AND (c) populating
-    ``prompt_cache_1h_allowlist`` from a Python-equivalent of the TS
-    GrowthBook config. All three are deferred to a future WI.
+    **Status (#285 — wired):** called once per session via
+    ``src/state/session_start.initialize_prompt_cache_state`` (from
+    ``init.pre_action``, env-signal backed) and lazily from
+    ``should_1h_cache_ttl`` when the latch is unevaluated (SDK paths
+    that skip pre_action; a /clear that reset the latches). 1h engages
+    when the eligibility signals AND a configured allowlist
+    (``populate_prompt_cache_1h_allowlist``) are both present;
+    otherwise every cache_control stays at ``ttl: '5m'``.
     """
     latches = get_beta_header_latches()
     if latches.prompt_cache_1h_eligible is None:
@@ -191,12 +214,24 @@ def should_1h_cache_ttl(query_source: str) -> bool:
       1. ``prompt_cache_1h_eligible`` is latched True (the user is eligible).
       2. ``query_source`` is in the allowlist (this specific call is eligible).
 
-    The allowlist is empty by default — until a future WI populates it
-    from configuration, every call defaults to ``ttl: '5m'``. This is the
-    safe-default behavior: 5m caching is already engaged from Phase 1; 1h
-    is an opt-in extension for sessions that cross 5-minute idle gaps.
+    Unconfigured installs default every call to ``ttl: '5m'`` — the
+    safe behavior already engaged in Phase 1; 1h is an opt-in extension
+    (#285: settings.prompt_cache_1h_sources / the env override) for
+    sessions that cross 5-minute idle gaps.
     """
     latches = get_beta_header_latches()
+    if latches.prompt_cache_1h_eligible is None:
+        # Lazy (re-)initialization — TS evaluates at the consumer
+        # (claude.ts:420-425). Covers SDK paths that never ran
+        # init.pre_action AND a /clear / /compact that reset the latch
+        # singleton (without this, a cleared session silently downgrades
+        # to 5m for its remainder).
+        try:
+            from src.state.session_start import initialize_prompt_cache_state
+
+            initialize_prompt_cache_state()
+        except Exception:
+            pass  # fail-soft: 5m below
     if latches.prompt_cache_1h_eligible is not True:
         return False
     return query_source in latches.prompt_cache_1h_allowlist