diff --git a/src/query/query.py b/src/query/query.py index 5bfa1978c..cce804fb9 100644 --- a/src/query/query.py +++ b/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: diff --git a/src/state/app_state.py b/src/state/app_state.py index d13cad665..827e67e69 100644 --- a/src/state/app_state.py +++ b/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, } diff --git a/tests/integration/test_advisor_smoke.py b/tests/integration/test_advisor_smoke.py new file mode 100644 index 000000000..1d3984b6b --- /dev/null +++ b/tests/integration/test_advisor_smoke.py @@ -0,0 +1,323 @@ +"""End-to-end advisor smoke tests against a mocked Anthropic SDK stream. + +Two scenarios: + 1. Happy path — the stream emits text + advisor server_tool_use + + advisor_tool_result + text. Verify the assembled assistant message + preserves all four blocks AND the next-turn outbound request + includes the use/result pair in history (because the beta header + keeps going). + 2. Interrupt path — the stream emits text + advisor server_tool_use + without a matching result (simulating ESC mid-advisor). Verify + the next turn's outbound message has the orphan ``server_tool_use`` + stripped by ``ensure_tool_result_pairing`` so the API doesn't 400. +""" + +from __future__ import annotations + +import asyncio +import os +import tempfile +import unittest +from typing import Any +from unittest.mock import MagicMock, patch + +from src.providers.anthropic_provider import AnthropicProvider +from src.providers.base import ChatResponse +from src.query.query import _call_model_sync +from src.types.messages import AssistantMessage, UserMessage + + +class _Isolation: + """Override the module-level config-path constants to a tmp dir. + + See ``tests/test_advisor_request_wiring.py`` for the rationale — + ``src/config.py`` freezes the path constants at import time so a + plain ``HOME`` patch is too late. + """ + + def __init__(self) -> None: + self._tmp = tempfile.mkdtemp(prefix="advisor_smoke_") + self._saved_global = None + self._saved_history = None + self._saved_dir = None + + def enter(self) -> None: + import src.config as cfg_mod + from pathlib import Path as _P + self._saved_global = cfg_mod.GLOBAL_CONFIG_FILE + self._saved_history = cfg_mod.HISTORY_FILE + self._saved_dir = cfg_mod.GLOBAL_CONFIG_DIR + cfg_mod.GLOBAL_CONFIG_FILE = _P(self._tmp) / ".clawcodex" / "config.json" + cfg_mod.HISTORY_FILE = _P(self._tmp) / ".clawcodex" / "history.jsonl" + cfg_mod.GLOBAL_CONFIG_DIR = _P(self._tmp) / ".clawcodex" + cfg_mod._default_manager = None + from src.settings.settings import invalidate_settings_cache + invalidate_settings_cache() + + def exit(self) -> None: + import src.config as cfg_mod + cfg_mod.GLOBAL_CONFIG_FILE = self._saved_global + cfg_mod.HISTORY_FILE = self._saved_history + cfg_mod.GLOBAL_CONFIG_DIR = self._saved_dir + cfg_mod._default_manager = None + from src.settings.settings import invalidate_settings_cache + invalidate_settings_cache() + + +def _isolate_env(): + return _Isolation() + + +def _set_advisor(value: str) -> None: + import src.config as cfg_mod + from src.config import ConfigManager + from src.settings.settings import invalidate_settings_cache + cfg_mod._default_manager = None + mgr = ConfigManager() + cfg = mgr.load_global() + sub = cfg.get("settings") if isinstance(cfg.get("settings"), dict) else {} + sub["advisor_model"] = value + cfg["settings"] = sub + mgr.save_global(cfg) + invalidate_settings_cache() + + +def _build_fake_anthropic_response(content_blocks: list[dict]) -> Any: + """Build a fake ChatResponse with all the blocks the SDK would surface. + + We bypass ``messages.stream`` entirely and stub + ``chat_stream_response`` to return the assembled ChatResponse — the + SDK code path is exercised by ``test_provider_anthropic.py`` + elsewhere; here we want to verify the query layer assembles blocks + correctly when the provider does its job. + """ + raw_blocks: list[dict] = [] + text_content = "" + tool_uses: list[dict] = [] + for block in content_blocks: + btype = block.get("type") + if btype == "text": + text_content += block.get("text", "") + elif btype == "tool_use": + tool_uses.append({ + "id": block["id"], + "name": block["name"], + "input": block.get("input", {}), + }) + elif btype in ("server_tool_use", "advisor_tool_result"): + raw_blocks.append(dict(block)) + return ChatResponse( + content=text_content, + model="claude-opus-4-6", + usage={"input_tokens": 5, "output_tokens": 20}, + finish_reason="end_turn", + tool_uses=tool_uses or None, + raw_content_blocks=raw_blocks or None, + ) + + +class _Capture: + api_messages: list = None + call_kwargs: dict = None + + +def _make_provider(captured: _Capture, *, response_blocks: list[dict]): + provider = MagicMock(spec=AnthropicProvider) + provider.has_custom_endpoint = MagicMock(return_value=False) + provider.model = "claude-opus-4-6" + + def fake_chat_stream_response(api_messages, *, abort_signal=None, **kwargs): + captured.api_messages = list(api_messages) + captured.call_kwargs = dict(kwargs) + return _build_fake_anthropic_response(response_blocks) + + provider.chat_stream_response = fake_chat_stream_response + return provider + + +class TestAdvisorHappyPath(unittest.TestCase): + def setUp(self) -> None: + self._iso = _isolate_env() + self._iso.enter() + os.environ.pop("CLAUDE_CODE_DISABLE_ADVISOR_TOOL", None) + _set_advisor("claude-opus-4-6") + + def tearDown(self) -> None: + # _iso.exit() restores the real config paths AND clears the + # singleton cache; any write here would leak onto the real + # user's config file. The tmp dir from .enter() is its own + # world — no additional cleanup write needed. + self._iso.exit() + + def test_advisor_pair_preserved_in_history(self) -> None: + cap = _Capture() + response_blocks = [ + {"type": "text", "text": "Let me check with the advisor. "}, + { + "type": "server_tool_use", + "id": "srv_001", + "name": "advisor", + "input": {}, + }, + { + "type": "advisor_tool_result", + "tool_use_id": "srv_001", + "content": {"type": "advisor_result", "text": "Looks good."}, + }, + {"type": "text", "text": "Proceeding."}, + ] + provider = _make_provider(cap, response_blocks=response_blocks) + # Turn 1. + result_msgs, _ = asyncio.run(_call_model_sync( + provider=provider, + messages=[UserMessage(content="What should I do?")], + system_prompt="sys", + tools=[], + )) + # The assembled AssistantMessage MUST carry the advisor pair as + # raw passthrough dicts so the next turn can replay them. + self.assertEqual(len(result_msgs), 1) + asst = result_msgs[0] + types = [] + for b in asst.content if isinstance(asst.content, list) else []: + if isinstance(b, dict): + types.append(b.get("type")) + else: + types.append(getattr(b, "type", "?")) + self.assertIn("server_tool_use", types) + self.assertIn("advisor_tool_result", types) + + # Turn 2 — feed the assistant message back and confirm the API + # payload preserves the pair (beta header still going, so no + # strip should happen). + cap2 = _Capture() + provider2 = _make_provider(cap2, response_blocks=[ + {"type": "text", "text": "ok"}, + ]) + asyncio.run(_call_model_sync( + provider=provider2, + messages=[ + UserMessage(content="What should I do?"), + asst, + UserMessage(content="now what?"), + ], + system_prompt="sys", + tools=[], + )) + # Beta header still attached (advisor still active). + self.assertIn("advisor-tool-2026-03-01", cap2.call_kwargs.get("betas", [])) + # Advisor blocks survived through normalize → ensure_pairing. + asst_payload = next( + m for m in cap2.api_messages if m.get("role") == "assistant" + ) + api_types = [b.get("type") for b in asst_payload["content"]] + self.assertIn("server_tool_use", api_types) + self.assertIn("advisor_tool_result", api_types) + + +class TestAdvisorInterruptPath(unittest.TestCase): + def setUp(self) -> None: + self._iso = _isolate_env() + self._iso.enter() + os.environ.pop("CLAUDE_CODE_DISABLE_ADVISOR_TOOL", None) + _set_advisor("claude-opus-4-6") + + def tearDown(self) -> None: + # _iso.exit() restores the real config paths AND clears the + # singleton cache; any write here would leak onto the real + # user's config file. The tmp dir from .enter() is its own + # world — no additional cleanup write needed. + self._iso.exit() + + def test_orphan_use_stripped_next_turn(self) -> None: + cap = _Capture() + # Simulate an interrupted advisor — the use block landed but + # the result never did (ESC mid-call). The provider returns it + # as a raw passthrough block; the next turn must strip it. + response_blocks = [ + {"type": "text", "text": "consulting"}, + { + "type": "server_tool_use", + "id": "srv_orphan", + "name": "advisor", + "input": {}, + }, + ] + provider = _make_provider(cap, response_blocks=response_blocks) + result_msgs, _ = asyncio.run(_call_model_sync( + provider=provider, + messages=[UserMessage(content="Help me")], + system_prompt="sys", + tools=[], + )) + asst = result_msgs[0] + + # Turn 2 — the next request must drop the orphan + # ``server_tool_use`` (else the API rejects with "advisor tool + # use without corresponding advisor_tool_result"). + cap2 = _Capture() + provider2 = _make_provider(cap2, response_blocks=[ + {"type": "text", "text": "ok"}, + ]) + asyncio.run(_call_model_sync( + provider=provider2, + messages=[ + UserMessage(content="Help me"), + asst, + UserMessage(content="continue"), + ], + system_prompt="sys", + tools=[], + )) + asst_payload = next( + m for m in cap2.api_messages if m.get("role") == "assistant" + ) + api_types = [b.get("type") for b in asst_payload["content"]] + self.assertNotIn( + "server_tool_use", api_types, + "Orphan server_tool_use (advisor) MUST be stripped before send", + ) + + def test_orphan_stripped_even_with_beta_active(self) -> None: + # Critical: the strip pass in ensure_tool_result_pairing applies + # regardless of whether the beta header is going — it removes + # orphans because the API rejects them in ALL cases, not just + # when the header is absent. + cap = _Capture() + provider = _make_provider(cap, response_blocks=[ + { + "type": "server_tool_use", + "id": "srv_orphan_2", + "name": "advisor", + "input": {}, + }, + ]) + asst_msg = AssistantMessage(content=[ + { + "type": "server_tool_use", + "id": "srv_orphan_2", + "name": "advisor", + "input": {}, + }, + ]) + asyncio.run(_call_model_sync( + provider=provider, + messages=[ + UserMessage(content="hi"), + asst_msg, + UserMessage(content="x"), + ], + system_prompt="sys", + tools=[], + )) + # Beta IS going (active advisor) — but the orphan still strips. + self.assertIn("advisor-tool-2026-03-01", cap.call_kwargs.get("betas", [])) + asst_payload = next( + m for m in cap.api_messages if m.get("role") == "assistant" + ) + api_types = [b.get("type") for b in asst_payload["content"]] + self.assertNotIn("server_tool_use", api_types) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_advisor_request_wiring.py b/tests/test_advisor_request_wiring.py new file mode 100644 index 000000000..46f0f2149 --- /dev/null +++ b/tests/test_advisor_request_wiring.py @@ -0,0 +1,389 @@ +"""Verify the request-time wiring in ``src/query/query.py:_call_model_sync``. + +Three classes of assertion: + * Activation gate: advisor decision == (first-party Anthropic AND + settings.advisor_model AND model_supports_advisor AND + is_valid_advisor_model). + * On activation: schema appended after regular tools, beta header + set, instructions appended to system_prompt, advisor blocks NOT + stripped from outgoing messages. + * Off activation: schema absent, beta absent, instructions absent, + advisor blocks STRIPPED from outgoing messages (so the API doesn't + 400 on a stale server_tool_use without the beta header). + +Tests intercept ``provider.chat_stream_response`` to inspect the kwargs +the query layer would have sent to the SDK. This avoids opening any +network connection. +""" + +from __future__ import annotations + +import asyncio +import os +import tempfile +import unittest +from typing import Any +from unittest.mock import MagicMock, patch + +from src.providers.anthropic_provider import AnthropicProvider +from src.providers.base import ChatResponse +from src.query.query import _call_model_sync +from src.types.messages import AssistantMessage, UserMessage + + +class _Capture: + """Holds the kwargs captured from a single ``chat_stream_response`` call.""" + api_messages: list = None + call_kwargs: dict = None + + +def _stub_provider_class(provider_cls, captured: _Capture) -> Any: + """Create a provider instance that records the kwargs it was called with.""" + if provider_cls is AnthropicProvider: + provider = MagicMock(spec=AnthropicProvider) + provider.has_custom_endpoint = MagicMock(return_value=False) + else: + provider = MagicMock(spec=provider_cls) + provider.model = "claude-opus-4-6" + + def fake_chat_stream_response(api_messages, *, abort_signal=None, **kwargs): + captured.api_messages = list(api_messages) + captured.call_kwargs = dict(kwargs) + return ChatResponse( + content="ok", + model="claude-opus-4-6", + usage={"input_tokens": 1, "output_tokens": 1}, + finish_reason="end_turn", + tool_uses=None, + raw_content_blocks=None, + ) + + provider.chat_stream_response = fake_chat_stream_response + return provider + + +class _Isolation: + """Helper that monkeypatches the config-file paths to a tmp dir. + + ``src/config.py`` evaluates ``GLOBAL_CONFIG_FILE = Path.home() / + ".clawcodex/config.json"`` at import time, so patching ``HOME`` + after import is too late — writes would land on the real user's + config file. We override the module-level constant directly. + + The helper is used as a class with ``enter()`` / ``exit()`` from + test setUp / tearDown rather than via ``with`` so existing + setUp/tearDown shapes don't need restructuring. + """ + + def __init__(self) -> None: + self._tmp = tempfile.mkdtemp(prefix="advisor_wire_") + self._saved_global = None + self._saved_history = None + self._saved_dir = None + + def enter(self) -> None: + import src.config as cfg_mod + self._saved_global = cfg_mod.GLOBAL_CONFIG_FILE + self._saved_history = cfg_mod.HISTORY_FILE + self._saved_dir = cfg_mod.GLOBAL_CONFIG_DIR + from pathlib import Path as _P + cfg_mod.GLOBAL_CONFIG_FILE = _P(self._tmp) / ".clawcodex" / "config.json" + cfg_mod.HISTORY_FILE = _P(self._tmp) / ".clawcodex" / "history.jsonl" + cfg_mod.GLOBAL_CONFIG_DIR = _P(self._tmp) / ".clawcodex" + cfg_mod._default_manager = None + from src.settings.settings import invalidate_settings_cache + invalidate_settings_cache() + + def exit(self) -> None: + import src.config as cfg_mod + cfg_mod.GLOBAL_CONFIG_FILE = self._saved_global + cfg_mod.HISTORY_FILE = self._saved_history + cfg_mod.GLOBAL_CONFIG_DIR = self._saved_dir + cfg_mod._default_manager = None + from src.settings.settings import invalidate_settings_cache + invalidate_settings_cache() + + +def _isolate_env(): + """Build an _Isolation helper. setUp/tearDown call .enter/.exit.""" + return _Isolation() + + +def _set_settings(**kwargs): + """Write keys into the global settings file + invalidate cache.""" + import src.config as cfg_mod + from src.config import ConfigManager + from src.settings.settings import invalidate_settings_cache + cfg_mod._default_manager = None + mgr = ConfigManager() + cfg = mgr.load_global() + sub = cfg.get("settings") if isinstance(cfg.get("settings"), dict) else {} + sub.update(kwargs) + cfg["settings"] = sub + mgr.save_global(cfg) + invalidate_settings_cache() + + +def _run(provider, messages, system_prompt="hi", tools=None): + return asyncio.run( + _call_model_sync( + provider=provider, + messages=messages, + system_prompt=system_prompt, + tools=tools or [], + ) + ) + + +class TestAdvisorActiveOnFirstPartyAnthropic(unittest.TestCase): + """Full activation path: beta header + schema + instructions added.""" + + def setUp(self) -> None: + self._iso = _isolate_env() + self._iso.enter() + os.environ.pop("CLAUDE_CODE_DISABLE_ADVISOR_TOOL", None) + _set_settings(advisor_model="claude-opus-4-6") + + def tearDown(self) -> None: + # _iso.exit() restores the real config paths AND clears the + # singleton cache; the tmp dir from .enter() is its own world, + # so no additional cleanup write is needed (and any write here + # would land on the real user's config — see the isolation + # helper docstring above). + self._iso.exit() + + def test_beta_header_attached(self) -> None: + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + _run(provider, [UserMessage(content="please")]) + self.assertIn("betas", cap.call_kwargs) + self.assertIn("advisor-tool-2026-03-01", cap.call_kwargs["betas"]) + + def test_schema_appended_after_regular_tools(self) -> None: + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + + class _FakeTool: + name = "Bash" + input_schema = {"type": "object", "properties": {}} + def prompt(self) -> str: + return "fake bash tool" + + _run(provider, [UserMessage(content="x")], tools=[_FakeTool()]) + tools = cap.call_kwargs["tools"] + # Regular tool first, advisor LAST — preserves the cache_control + # marker position from any other tool. + self.assertEqual(tools[0]["name"], "Bash") + self.assertEqual(tools[-1]["type"], "advisor_20260301") + self.assertEqual(tools[-1]["name"], "advisor") + self.assertEqual(tools[-1]["model"], "claude-opus-4-6") + + def test_instructions_appended_to_string_system_prompt(self) -> None: + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + _run(provider, [UserMessage(content="x")], system_prompt="base prompt") + sysp = cap.call_kwargs.get("system") + self.assertIsInstance(sysp, str) + self.assertIn("base prompt", sysp) + self.assertIn("# Advisor Tool", sysp) + + def test_instructions_appended_to_block_list_system_prompt(self) -> None: + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + sysp_in = [ + {"type": "text", "text": "block 1"}, + {"type": "text", "text": "block 2", "cache_control": {"type": "ephemeral"}}, + ] + _run(provider, [UserMessage(content="x")], system_prompt=sysp_in) + sysp_out = cap.call_kwargs.get("system") + self.assertIsInstance(sysp_out, list) + # Instructions land at the END so the cache_control marker on + # block 2 keeps its position relative to the prefix. + self.assertEqual(sysp_out[-1]["type"], "text") + self.assertIn("# Advisor Tool", sysp_out[-1]["text"]) + # Earlier blocks (including the cached one) are unchanged. + self.assertEqual(sysp_out[0]["text"], "block 1") + self.assertEqual(sysp_out[1]["text"], "block 2") + self.assertIn("cache_control", sysp_out[1]) + + def test_advisor_blocks_in_history_are_kept(self) -> None: + # When the request will carry the beta header, historical + # advisor blocks must round-trip back to the API as a valid + # use/result pair. + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + prior = AssistantMessage( + content=[ + {"type": "text", "text": "before"}, + { + "type": "server_tool_use", + "id": "srv_1", + "name": "advisor", + "input": {}, + }, + { + "type": "advisor_tool_result", + "tool_use_id": "srv_1", + "content": {"type": "advisor_result", "text": "advice"}, + }, + ], + ) + _run(provider, [UserMessage(content="hi"), prior, UserMessage(content="next")]) + # Inspect the assistant message in the API payload. + asst = next( + m for m in cap.api_messages + if m.get("role") == "assistant" + ) + types = [b["type"] for b in asst["content"]] + self.assertIn("server_tool_use", types) + self.assertIn("advisor_tool_result", types) + + +class TestAdvisorInactivePaths(unittest.TestCase): + """Negative cases — schema/header/instructions MUST NOT be sent.""" + + def setUp(self) -> None: + self._iso = _isolate_env() + self._iso.enter() + os.environ.pop("CLAUDE_CODE_DISABLE_ADVISOR_TOOL", None) + + def tearDown(self) -> None: + # _iso.exit() restores the real config paths AND clears the + # singleton cache; the tmp dir from .enter() is its own world, + # so no additional cleanup write is needed (and any write here + # would land on the real user's config — see the isolation + # helper docstring above). + self._iso.exit() + + def _assert_inactive(self, cap: _Capture) -> None: + self.assertNotIn("betas", cap.call_kwargs) + tools = cap.call_kwargs.get("tools") or [] + for t in tools: + self.assertNotEqual( + t.get("type"), "advisor_20260301", + "advisor schema must not be sent when inactive", + ) + self.assertNotEqual( + t.get("name"), "advisor", + "advisor name must not appear in tools when inactive", + ) + sysp = cap.call_kwargs.get("system", "") or "" + if isinstance(sysp, list): + sysp_text = "\n".join( + b.get("text", "") for b in sysp if isinstance(b, dict) + ) + else: + sysp_text = sysp + self.assertNotIn("# Advisor Tool", sysp_text) + + def test_no_advisor_when_settings_unset(self) -> None: + _set_settings(advisor_model="") + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + _run(provider, [UserMessage(content="x")]) + self._assert_inactive(cap) + + def test_no_advisor_when_env_disabled(self) -> None: + _set_settings(advisor_model="claude-opus-4-6") + with patch.dict( + os.environ, {"CLAUDE_CODE_DISABLE_ADVISOR_TOOL": "1"}, clear=False + ): + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + _run(provider, [UserMessage(content="x")]) + self._assert_inactive(cap) + + def test_no_advisor_when_anthropic_has_custom_endpoint(self) -> None: + _set_settings(advisor_model="claude-opus-4-6") + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + # Simulate a custom base_url (Bedrock shim, self-hosted proxy). + provider.has_custom_endpoint = MagicMock(return_value=True) + _run(provider, [UserMessage(content="x")]) + self._assert_inactive(cap) + + def test_no_advisor_when_base_model_unsupported(self) -> None: + _set_settings(advisor_model="claude-opus-4-6") + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + # Stale main model — advisor must not piggy-back on an older model. + provider.model = "claude-opus-4-5" + _run(provider, [UserMessage(content="x")]) + self._assert_inactive(cap) + + def test_no_advisor_when_advisor_model_invalid(self) -> None: + _set_settings(advisor_model="claude-haiku-4-5") + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + _run(provider, [UserMessage(content="x")]) + self._assert_inactive(cap) + + def test_advisor_blocks_stripped_when_inactive(self) -> None: + # The previous turn left advisor blocks in history, but the + # current request will NOT carry the beta header (e.g. user + # ran /advisor unset). The API would 400 if we sent them. + _set_settings(advisor_model="") # inactive + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + prior = AssistantMessage( + content=[ + {"type": "text", "text": "before"}, + { + "type": "server_tool_use", + "id": "srv_1", + "name": "advisor", + "input": {}, + }, + { + "type": "advisor_tool_result", + "tool_use_id": "srv_1", + "content": {"type": "advisor_result", "text": "advice"}, + }, + {"type": "text", "text": "after"}, + ], + ) + _run(provider, [UserMessage(content="hi"), prior, UserMessage(content="next")]) + asst = next( + m for m in cap.api_messages if m.get("role") == "assistant" + ) + types = [b["type"] for b in asst["content"]] + self.assertNotIn("server_tool_use", types) + self.assertNotIn("advisor_tool_result", types) + + +class TestAdvisorActivationDefensive(unittest.TestCase): + """The activation predicate is wrapped in a single try/except so any + failure (transient import, future provider that throws on + `has_custom_endpoint`, settings cache contention) defaults to + "advisor inactive" rather than failing the turn. Critic-flagged + nit: the fix had no dedicated test — adding one here. + """ + + def test_activation_exception_does_not_kill_turn(self) -> None: + # Force `is_advisor_enabled` to raise inside `_call_model_sync`. + # The expected behavior is: caught, advisor_active=False, the + # request proceeds without the beta header or schema. + iso = _isolate_env() + iso.enter() + try: + _set_settings(advisor_model="claude-opus-4-6") + cap = _Capture() + provider = _stub_provider_class(AnthropicProvider, cap) + with patch( + "src.utils.advisor.is_advisor_enabled", + side_effect=RuntimeError("synthetic gate failure"), + ): + # Must not raise; should fall through to a non-advisor + # request. + result, _ = _run(provider, [UserMessage(content="hi")]) + self.assertEqual(len(result), 1) + self.assertNotIn("betas", cap.call_kwargs) + tools = cap.call_kwargs.get("tools") or [] + for t in tools: + self.assertNotEqual(t.get("type"), "advisor_20260301") + finally: + iso.exit() + + +if __name__ == "__main__": + unittest.main()