feat(advisor) 3/4: wire activation into the query layer

src/query/query.py

@@ -286,9 +286,68 @@ async def _call_model_sync(
     abort_signal: Any = None,
 ) -> tuple[list[AssistantMessage], list[ToolUseBlock]]:
     from ..types.messages import normalize_messages_for_api
+    from ..utils.advisor import (
+        ADVISOR_BETA_HEADER,
+        ADVISOR_TOOL_INSTRUCTIONS,
+        build_advisor_tool_schema,
+        is_advisor_enabled,
+        is_valid_advisor_model,
+        model_supports_advisor,
+        strip_advisor_blocks,
+    )
+
+    # Advisor activation decision — server-side parity with TS
+    # claude.ts:1094-1130. Activation requires ALL of:
+    # 1. Provider is first-party Anthropic (no custom base_url) and the
+    #    CLAUDE_CODE_DISABLE_ADVISOR_TOOL env switch is not set.
+    # 2. ``settings.advisor_model`` is non-empty (the user opted in via
+    #    /advisor; ``_on_advisor_model_change`` persists writes here).
+    # 3. The main-loop model supports calling the advisor tool
+    #    (opus-4-6 / sonnet-4-6 — older models reject the schema).
+    # 4. The configured advisor model itself is a valid advisor target.
+    # A stale advisor_model setting under a non-supporting model is
+    # silently ignored — never sent to the API. Mirrors the TS
+    # "skipping advisor — base model X does not support advisor" branch.
+    main_loop_model = getattr(provider, "model", "") or ""
+    advisor_active = False
+    advisor_model_normalized: str | None = None
+    # Wrap the ENTIRE activation predicate so any failure (transient
+    # import, future provider type that throws on the first-party
+    # check, settings cache contention) defaults to "advisor inactive"
+    # instead of failing the turn. Critic M1: previously the env/
+    # provider check was outside the inner try, so an exception in
+    # ``is_first_party_provider`` would propagate to the caller.
+    try:
+        if is_advisor_enabled(provider):
+            from ..settings.settings import get_settings
+            configured = (get_settings().advisor_model or "").strip()
+            if configured and model_supports_advisor(main_loop_model):
+                from ..models.model import canonical_model_name
+                candidate = canonical_model_name(configured)
+                if is_valid_advisor_model(candidate):
+                    advisor_active = True
+                    advisor_model_normalized = candidate
+    except Exception:
+        # Don't let advisor activation issues kill the turn. The
+        # historical advisor blocks (if any) will still be stripped
+        # below since advisor_active is False.
+        logger.exception(
+            "Advisor activation check failed; treating advisor as inactive"
+        )
+        advisor_active = False
+        advisor_model_normalized = None
 
     api_messages = normalize_messages_for_api(messages)
 
+    # When the advisor beta header is NOT going on this request, strip
+    # any historical advisor blocks — the API 400s on
+    # ``server_tool_use(name=advisor)`` and ``advisor_tool_result`` blocks
+    # without the matching beta. Mirror of TS claude.ts:1332-1334.
+    # Always-on when advisor is inactive (provider switch, env disable,
+    # base model unsupported, or user ran /advisor unset).
+    if not advisor_active:
+        api_messages = strip_advisor_blocks(api_messages)
+
     # --- Diagnostic tracing ---
     _diag = os.environ.get("CLAWCODEX_DEBUG", "").lower() in ("1", "true", "yes")
     if _diag:
@@ -339,8 +398,26 @@ async def _call_model_sync(
             "input_schema": dict(tool.input_schema),
         })
 
+    # Append the advisor schema AFTER the regular tools so the
+    # ``cache_control`` marker (which conventionally lives on the last
+    # cached tool — the final entry in ``tool_schemas`` before this
+    # append) stays in place. If we prepended or interleaved, toggling
+    # /advisor would shift the marker and bust the prompt cache. Mirrors
+    # TS claude.ts:1411-1421 explicitly.
+    if advisor_active:
+        tool_schemas.append(build_advisor_tool_schema(advisor_model_normalized))
+
     call_kwargs: dict[str, Any] = {"tools": tool_schemas}
 
+    if advisor_active:
+        # Opt into the server-side advisor tool. ``betas`` lives outside
+        # ``extra_headers`` because the SDK auto-converts it into the
+        # ``anthropic-beta`` header AND filters out 3P-incompatible
+        # entries on Bedrock/Vertex transports. Currently we send only
+        # the advisor beta; if other betas are introduced, change this
+        # to ``call_kwargs.setdefault("betas", []).append(...)``.
+        call_kwargs["betas"] = [ADVISOR_BETA_HEADER]
+
     from ..providers.anthropic_provider import AnthropicProvider
     from ..providers.minimax_provider import MinimaxProvider
 
@@ -349,6 +426,33 @@ async def _call_model_sync(
         # Forward whatever shape the engine produced — str or list[dict].
         # The SDK's ``system`` param accepts ``Union[str, Iterable[TextBlockParam]]``;
         # cache_control markers on blocks engage server-side prompt caching.
+        #
+        # When the advisor is active, append ``ADVISOR_TOOL_INSTRUCTIONS``
+        # AFTER the existing system prompt blocks. This mirrors TS
+        # claude.ts:1395 (the advisor instructions come AFTER the cached
+        # system blocks, so they land in the request-scope partition and
+        # toggling /advisor doesn't churn the cached prefix).
+        if advisor_active:
+            if isinstance(system_prompt, list):
+                system_prompt = list(system_prompt) + [
+                    {"type": "text", "text": ADVISOR_TOOL_INSTRUCTIONS}
+                ]
+            elif isinstance(system_prompt, str):
+                system_prompt = (
+                    f"{system_prompt}\n\n{ADVISOR_TOOL_INSTRUCTIONS}"
+                    if system_prompt
+                    else ADVISOR_TOOL_INSTRUCTIONS
+                )
+            else:
+                # Defensive: the upstream contract is
+                # ``str | list[dict[str, Any]]``. A future caller that
+                # passes something else (e.g. None, a TextBlock object)
+                # silently loses the instructions if we don't warn.
+                logger.warning(
+                    "Advisor active but system_prompt has unexpected type "
+                    "%s — ADVISOR_TOOL_INSTRUCTIONS NOT injected",
+                    type(system_prompt).__name__,
+                )
         call_kwargs["system"] = system_prompt
     else:
         # Non-Anthropic providers (OpenAI-compat, GLM, etc.) consume the
@@ -498,6 +602,18 @@ async def _call_model_sync(
             assistant_blocks.append(block)
             tool_use_blocks.append(block)
 
+    # Preserve advisor server-tool blocks as passthrough dicts so the
+    # next turn can replay them to the API as a matched use/result pair.
+    # ``normalize_messages_for_api`` round-trips dict blocks unchanged
+    # (via ``content_block_to_dict``), and ``ensure_tool_result_pairing``
+    # treats the advisor pair as a self-contained server-side
+    # use/result on the assistant message (already paired in-message).
+    # Stripping happens centrally in this function when ``advisor_active``
+    # is False on a future turn.
+    if response.raw_content_blocks:
+        for raw in response.raw_content_blocks:
+            assistant_blocks.append(dict(raw))
+
     stop_reason = response.finish_reason or "end_turn"
 
     if _diag:

src/state/app_state.py

@@ -90,6 +90,13 @@ class AppState:
     # to process at startup. TS: AppStateStore.ts:406.
     initial_message: str | None = None
 
+    # Advisor model (TS: AppStateStore.ts advisorModel). None = no /advisor.
+    # Writes here fire ``_on_advisor_model_change``, which persists into
+    # ``settings.advisor_model`` (the read channel — see
+    # ``src/utils/advisor.py`` and ``src/query/query.py``) and invalidates
+    # the settings cache so the next API call picks up the new value.
+    advisor_model: str | None = None
+
 
 def replace_state(state: AppState, **changes: Any) -> AppState:
     """Return a copy of ``state`` with ``changes`` applied. Equivalent
@@ -211,6 +218,55 @@ def _on_initial_message_change(old: AppState, new: AppState) -> None:
     return
 
 
+def _on_advisor_model_change(old: AppState, new: AppState) -> None:
+    """Persist advisor_model to user settings + invalidate the read cache.
+
+    Mirrors TS ``commands/advisor.ts:49`` (``updateSettingsForSource(
+    'userSettings', { advisorModel })``). The persistence pattern matches
+    the existing TODO on ``_on_main_loop_model_change`` — settings are
+    layered as ``global > project > local``; we write the user-scoped
+    global level only. After persisting, the in-memory settings cache must
+    be invalidated so the next ``get_settings()`` call (which
+    ``_call_model_sync`` makes per-turn) reflects the new value
+    immediately. Without the invalidate, mid-session toggles would only
+    take effect after a process restart.
+    """
+    if old.advisor_model == new.advisor_model:
+        return
+    # Local imports avoid making this module's import time pay for the
+    # settings stack on cold start. Use the SHARED default ConfigManager
+    # so callers reading via ``load_config()`` / the global cache see
+    # the new value, not a stale in-memory snapshot from before the
+    # write.
+    from src import config as cfg_mod
+    from src.settings.settings import invalidate_settings_cache
+    try:
+        mgr = cfg_mod._get_default_manager()
+        cfg = mgr.load_global()
+        settings_section = cfg.get("settings")
+        if not isinstance(settings_section, dict):
+            settings_section = {}
+        # None / "" both map to "unset" — write empty string for
+        # round-trip fidelity with the SettingsSchema default.
+        settings_section["advisor_model"] = new.advisor_model or ""
+        cfg["settings"] = settings_section
+        mgr.save_global(cfg)
+        invalidate_settings_cache()
+        logger.debug(
+            "AppState.advisor_model %s -> %s — persisted + cache invalidated",
+            old.advisor_model,
+            new.advisor_model,
+        )
+    except Exception:
+        # The slash command already reported success to the user based on
+        # the in-memory store update; surface persistence failures via the
+        # log but don't propagate (the in-memory change still works for
+        # the current process; only re-launches would lose the setting).
+        logger.exception(
+            "Failed to persist advisor_model to settings; in-memory value still active"
+        )
+
+
 # Registry. EVERY field in AppState must appear here as a handler
 # (function-form, including explicit no-ops). The coverage test enforces
 # this — adding a new AppState field without a handler entry fails
@@ -221,6 +277,7 @@ def _on_initial_message_change(old: AppState, new: AppState) -> None:
     "expanded_view": _on_expanded_view_change,
     "permission_mode": _on_permission_mode_change,
     "initial_message": _on_initial_message_change,
+    "advisor_model": _on_advisor_model_change,
 }