From 6a4bb2da2838a692df2ac7c8a9e030bc328fb194 Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Thu, 13 Aug 2026 01:12:10 -0700 Subject: [PATCH] feat: add /provider command for mid-session model pinning (experimental) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - New /provider (status), /provider use (pin), /provider auto (unpin) - Refuses loudly when orchestrator doesn't register capability (never silently fails) - Persistent [📌 ] prompt indicator, composing with existing [mode] indicator - (experimental) tagging on help/status/confirmations; usage-accuracy caveat in status view - Fixed latent bug: _create_prompt_session carried duplicate get_prompt closure that ignored pin getter * 48 unit tests passed while feature did not work * Added wiring tests that fail against broken code (6 of 11 fail when bug reintroduced) * Deleted duplicate; now properly invokes pin getter - Tests: 1372 passing Verified end-to-end in DTU with real interactive TUI and multi-vendor API calls. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- amplifier_app_cli/main.py | 367 ++++++++++++- tests/test_provider_command.py | 956 +++++++++++++++++++++++++++++++++ 2 files changed, 1312 insertions(+), 11 deletions(-) create mode 100644 tests/test_provider_command.py diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py index de93a2ed..a65b622e 100644 --- a/amplifier_app_cli/main.py +++ b/amplifier_app_cli/main.py @@ -1,6 +1,7 @@ """Amplifier CLI - Command-line interface for the Amplifier platform.""" import asyncio +import html import json import logging import os @@ -463,6 +464,13 @@ class CommandProcessor: "for a hard cap; or /goal clear)" ), }, + "/provider": { + "action": "handle_provider", + "description": ( + "(experimental) Show/pin the conversation-scope provider: " + "/provider (status) | /provider use | /provider auto" + ), + }, } # Dynamic shortcuts for modes (populated from mode definitions) @@ -473,6 +481,32 @@ class CommandProcessor: # Kept for backward compatibility; the canonical copy lives in dashboard_renderer. _SENSITIVE_KEY_PATTERNS = ("key", "token", "secret", "password", "api_key") + # /provider: caveat on the per-turn usage figures, shown in the STATUS + # view only -- deliberately NOT on the /provider use / auto confirmations, + # which are already 320 and 284 chars (5 and 4 wrapped lines at 80 cols). + # Repeating this there would push the every-time confirmation past 7 lines + # for a condition that is a property of the CLI's usage display, not of + # pinning. Status is the command's reference surface and has the room. + # + # ACCURACY GUARD -- do not reword into "costs are wrong". Measured across + # four providers against raw events.jsonl: per-vendor rates are CORRECT + # and are never cross-applied (exact to 8 decimals). The real defects are + # display/precision ones, all pre-existing and all being fixed in the + # provider and streaming-UI repos: + # * one vendor's Input token count is inflated (cost unaffected) + # * another omits thinking/reasoning tokens from count and cost + # * costs >= $0.01 are rounded to 2dp (can understate) + # * the newest models are missing from the rate table (no cost shown) + # "over- or under-reported" covers the first two without naming vendors + # that will shortly be fixed; the closing sentence is what keeps this + # honest rather than alarming. + _PROVIDER_USAGE_CAVEAT = ( + "Usage figures: per-turn token and cost numbers are indicative, not " + "billing-grade -- counts can be over- or under-reported, costs are " + "rounded, and the newest models may show no cost. Per-provider rates " + "themselves are correct." + ) + # /goal: aliases that clear an active goal. The turn cap is optional and # None (unlimited) by default -- deliberately, see # docs/decisions/ADR-0005-goal-unlimited-by-default.md. A positive int @@ -681,6 +715,9 @@ async def handle_command(self, action: str, data: dict[str, Any]) -> str: if action == "handle_goal": return await self._handle_goal(data.get("args", "")) + if action == "handle_provider": + return await self._handle_provider(data.get("args", "")) + if action == "list_modes": return await self._list_modes() @@ -1242,6 +1279,211 @@ async def _handle_goal(self, args: str) -> str: cap_suffix = f" (max {cap} turns)" if cap else " (unlimited turns)" return f"Goal set{cap_suffix}." + # === /provider: pin/unpin the conversation-scope provider === + # + # Provider selection is ORCHESTRATOR policy, not app policy (see + # amplifier_module_loop_streaming.ConversationProviderPin). This app + # layer only asks the 'conversation.provider_pin' capability and + # reports what it says -- it never selects a provider itself, and + # never claims a switch has happened before the capability confirms + # it. If the capability isn't registered, this orchestrator doesn't + # support pinning and we must say so instead of pretending to succeed. + + def _current_orchestrator_name(self) -> str | None: + """Best-effort orchestrator module name, for error messages only.""" + raw_config = self.session.coordinator.config + session_config = ( + raw_config.get("session", {}) if isinstance(raw_config, dict) else {} + ) + if not isinstance(session_config, dict): + return None + value = session_config.get("orchestrator") + if isinstance(value, dict): + module = value.get("module") + return module if isinstance(module, str) else None + if isinstance(value, str): + return value + return None + + def _provider_pin_unavailable_message(self) -> str: + """Refusal text when 'conversation.provider_pin' isn't registered. + + Must be shown instead of any success/status claim -- see + REQUIRED BEHAVIORS #1: do not write config, do not report success. + """ + orchestrator_name = self._current_orchestrator_name() + where = f" ('{orchestrator_name}')" if orchestrator_name else "" + return ( + f"Provider pinning is not supported by this session's " + f"orchestrator{where}: the 'conversation.provider_pin' capability " + f"is not registered, so /provider cannot pin, unpin, or confirm " + f"an active provider mid-session.\n" + f"Providers can only be changed by restarting with a different " + f"configuration." + ) + + @staticmethod + def _provider_model_for_display(provider: Any) -> str | None: + """Best-effort default model name via the Provider protocol's + synchronous ``get_info()`` (no network I/O) -- display only, never + used to make a routing decision.""" + try: + info = provider.get_info() + except Exception: + return None + defaults = getattr(info, "defaults", None) + if not isinstance(defaults, dict): + return None + model = defaults.get("model") + return model if isinstance(model, str) else None + + @staticmethod + def _provider_priority_for_display(provider: Any) -> int: + """Mirrors the tie-break amplifier_module_loop_streaming's + ``_select_provider`` reads (attribute, then config, then default + 100) -- display only, so the status view matches automatic + selection without this app layer making that selection itself.""" + if hasattr(provider, "priority"): + return provider.priority + config = getattr(provider, "config", None) + if isinstance(config, dict): + return config.get("priority", 100) + return 100 + + def _render_provider_status(self, pin: Any) -> str: + """Render mounted providers, their models/priorities, and whether + the conversation is pinned or automatic (REQUIRED BEHAVIORS: show + mounted providers, mark which is active, state pin state).""" + mounted = self.session.coordinator.get("providers") or {} + lines = ["Conversation providers (experimental):"] + + if not mounted: + lines.append(" (none mounted)") + if pin is None: + lines.append("") + lines.append(self._provider_pin_unavailable_message()) + return "\n".join(lines) + + pinned_name: str | None = None + if pin is not None: + try: + pinned_name = pin.current() + except Exception: + pinned_name = None + + rows = [ + ( + name, + self._provider_model_for_display(provider), + self._provider_priority_for_display(provider), + ) + for name, provider in sorted(mounted.items()) + ] + + # Display-only: which mount priority ordering would pick right now, + # when unpinned. This mirrors the orchestrator's own tie-break so + # the status view is informative, but it is never used to decide + # anything -- the per-turn usage line remains the real confirmation. + priority_winner = None + if pin is not None and pinned_name is None and rows: + priority_winner = min(rows, key=lambda r: r[2])[0] + + for name, model, priority in rows: + model_label = model or "(unknown)" + if name == pinned_name: + marker, suffix = "\u2605 ", " [pinned, active]" + elif name == priority_winner: + marker, suffix = "\u2605 ", " [active by priority]" + else: + marker, suffix = " ", "" + lines.append( + f"{marker}{name:<24} model={model_label:<28} priority={priority}{suffix}" + ) + + lines.append("") + if pin is None: + # Capability absent: the refusal is the whole message. Do NOT + # append the usage caveat here -- same principle that keeps the + # (experimental) tag off the error paths. + lines.append(self._provider_pin_unavailable_message()) + return "\n".join(lines) + + if pinned_name is not None: + lines.append(f"Selection: pinned to '{pinned_name}'.") + else: + lines.append( + "Selection: automatic (priority order). The \u2605 above shows " + "what priority ordering currently favors; the orchestrator " + "resolves the actual provider each turn -- confirm via the " + "per-turn usage line." + ) + lines.append(self._PROVIDER_USAGE_CAVEAT) + + return "\n".join(lines) + + async def _handle_provider(self, args: str) -> str: + """Handle /provider: status (no args), 'use ' to pin, or + 'auto' to unpin. See REQUIRED BEHAVIORS in the task spec this + command was built from: + + 1. Capability absent -> refuse loudly, no config write, no success. + 2. pin() ValueError -> clean user-facing error, not a traceback. + 3. Never claim a switch before it's confirmed (next turn, not now). + 4. Report scope accurately: top-level conversation only. + """ + args = args.strip() + parts = args.split(maxsplit=1) + subcmd = parts[0].lower() if parts else "" + + pin = self.session.coordinator.get_capability("conversation.provider_pin") + + if not subcmd: + return self._render_provider_status(pin) + + if subcmd in ("use", "auto") and pin is None: + return self._provider_pin_unavailable_message() + + if subcmd == "use": + name = parts[1].strip() if len(parts) > 1 else "" + if not name: + return ( + "Usage: /provider use . Run /provider to see " + "mounted providers." + ) + try: + pin.pin(name) + except ValueError as e: + return f"\u2717 {e}" + return ( + f"(experimental) Pinned conversation provider to '{name}'. " + f"This takes effect on the NEXT turn, not this one -- the " + f"token-usage line after your next message is your " + f"confirmation of which model actually answered. Scope: " + f"top-level conversation only; model-role routing, " + f"sub-agents, and the /goal loop are unaffected." + ) + + if subcmd == "auto": + previous = pin.unpin() + if previous is None: + return ( + "(experimental) Conversation provider is already " + "automatic (priority order). Nothing to unpin." + ) + return ( + f"(experimental) Unpinned conversation provider (was " + f"'{previous}'). Priority-based selection resumes on the " + f"NEXT turn -- confirm via the token-usage line after your " + f"next message. Scope: top-level conversation only; " + f"model-role routing, sub-agents, and the /goal loop are " + f"unaffected." + ) + + return ( + f"Unknown /provider subcommand: {subcmd!r}. " + f"Usage: /provider | /provider use | /provider auto" + ) + async def _rename_session(self, new_name: str) -> str: """Rename the current session.""" new_name = new_name.strip() @@ -2703,12 +2945,109 @@ async def process_runtime_mentions(session: AmplifierSession, prompt: str) -> st ) -def _create_prompt_session(get_active_mode: Callable | None = None) -> PromptSession: +def _escape_prompt_text(text: str) -> str: + """Escape a value for literal interpolation into prompt_toolkit ``HTML()``. + + ``HTML()`` parses its argument as markup, so a mode or provider name + containing ``<`` or ``&`` would otherwise raise -- inside a callable that + runs on every keystroke. Quotes are deliberately left alone: these values + are interpolated as element text, never as attribute values. + """ + return html.escape(text, quote=False) + + +def _pinned_provider_name(session: Any) -> str | None: + """The pinned conversation provider's mount name, or None. + + Reads the orchestrator's ``conversation.provider_pin`` capability live + (see ``amplifier_module_loop_streaming.ConversationProviderPin``) so the + prompt indicator tracks ``/provider use`` and ``/provider auto`` without + the app keeping its own copy of the pin state -- a copy could disagree + with the orchestrator, which is exactly what this feature exists to + prevent. + + Returns None when the capability is absent (this orchestrator does not + support pinning) or when nothing is pinned. Both cases mean "no + indicator", so the prompt renders byte-for-byte as it did before this + feature existed. + + Cost: one capability lookup plus one attribute read. This is called on + every keystroke render -- see ``_build_prompt_message``. + """ + pin = session.coordinator.get_capability("conversation.provider_pin") + if pin is None: + return None + name = pin.current() + return name if isinstance(name, str) and name else None + + +def _build_prompt_message( + get_active_mode: Callable | None = None, + get_pinned_provider: Callable | None = None, +) -> HTML: + """Build the REPL prompt, composing the optional indicators. + + Renders (mode leftmost, then pin, then the prompt caret):: + + unpinned: > + pinned: [pin anthropic-haiku]> + pinned + mode: [plan][pin anthropic-haiku]> + + (``pin`` above is the U+1F4CC pushpin glyph.) The pin indicator shows + the FULL provider mount name, never truncated: the user pinned by mount + name and several names share a model family (anthropic-fable / + anthropic-opus / anthropic-sonnet / anthropic-haiku), so shortening it + would destroy the only information the indicator carries. + + MUST NOT RAISE. This runs on every keystroke render, so an exception + here would make the session unusable. Every getter is called defensively + and a failing one degrades to "no indicator" -- independently, so one + broken getter cannot suppress the other's indicator. + + When neither indicator applies, the returned markup is byte-for-byte + identical to the pre-pin prompt. + """ + + def _indicator(getter: Callable | None) -> str | None: + if getter is None: + return None + try: + value = getter() + except Exception: + # A broken getter costs its indicator, never the prompt. + logger.debug("prompt indicator getter failed", exc_info=True) + return None + if not value: + return None + # Escape before interpolating into HTML(): an unescaped '<' or '&' + # in a mode or provider name would make HTML() raise at render time. + return _escape_prompt_text(str(value)) + + indicators = "" + active_mode = _indicator(get_active_mode) + if active_mode: + indicators += f"[{active_mode}]" + pinned_provider = _indicator(get_pinned_provider) + if pinned_provider: + indicators += f"[\U0001f4cc {pinned_provider}]" + + try: + return HTML(f"\n{indicators}> ") + except Exception: + # Last resort: a bare prompt always beats an unusable session. + logger.debug("prompt markup failed to build", exc_info=True) + return HTML("\n> ") + + +def _create_prompt_session( + get_active_mode: Callable | None = None, + get_pinned_provider: Callable | None = None, +) -> PromptSession: """Create configured PromptSession for REPL. Provides: - Persistent history at ~/.amplifier/projects//repl_history - - Dynamic prompt that shows [mode] indicator when a mode is active + - Dynamic prompt that shows [mode] and [pin] indicators when active - Green prompt styling matching Rich console - History search with Ctrl-R - Multi-line input with Ctrl-J @@ -2716,6 +3055,8 @@ def _create_prompt_session(get_active_mode: Callable | None = None) -> PromptSes Args: get_active_mode: Optional callable that returns the current active mode name + get_pinned_provider: Optional callable that returns the pinned + conversation provider's mount name (see ``_pinned_provider_name``) Returns: Configured PromptSession instance @@ -2759,15 +3100,18 @@ def accept_input(event): """Submit input on Enter.""" event.current_buffer.validate_and_handle() - # Dynamic prompt that shows [mode] indicator when a mode is active + # Dynamic prompt that shows [mode] and [pin] indicators when active. + # + # Delegates entirely to _build_prompt_message() -- the composer that + # actually handles both indicators, escaping, and defensive per-getter + # exception handling (see its docstring). This closure previously + # reimplemented mode-only rendering inline and never referenced + # get_pinned_provider at all, so the pin indicator was silently never + # rendered even though the pin feature itself worked correctly and + # _build_prompt_message was fully implemented and unit-tested. Do not + # reintroduce a second implementation here -- call the single composer. def get_prompt(): - if get_active_mode: - active_mode = get_active_mode() - if active_mode: - return HTML( - f"\n[{active_mode}]> " - ) - return HTML("\n> ") + return _build_prompt_message(get_active_mode, get_pinned_provider) return PromptSession( message=get_prompt, # Callable for dynamic prompt @@ -2878,7 +3222,8 @@ async def interactive_chat( prompt_session = _create_prompt_session( get_active_mode=lambda: command_processor.session.coordinator.session_state.get( "active_mode" - ) + ), + get_pinned_provider=lambda: _pinned_provider_name(command_processor.session), ) # Helper to extract model name from config diff --git a/tests/test_provider_command.py b/tests/test_provider_command.py new file mode 100644 index 00000000..d6a7e8d2 --- /dev/null +++ b/tests/test_provider_command.py @@ -0,0 +1,956 @@ +"""Tests for the /provider command: pin/unpin the conversation-scope provider +mid-session via the orchestrator's 'conversation.provider_pin' capability. + +See amplifier_module_loop_streaming.ConversationProviderPin for the capability +contract this command is built against. This app layer only asks and reports +-- it never selects a provider itself (REQUIRED BEHAVIORS in the task spec). +""" + +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +sys.path.insert(0, str(Path(__file__).parent)) + +from helpers import _make_command_processor + + +def _make_provider(model=None, priority=None, config_priority=None): + """Build a mock Provider-protocol object for display-only reads. + + - model: value returned via get_info().defaults["model"] + - priority: sets a `.priority` attribute directly (highest precedence, + mirrors amplifier_module_loop_streaming._select_provider's own read + order) + - config_priority: sets `.config = {"priority": N}` instead + """ + provider = MagicMock() + info = SimpleNamespace(defaults={"model": model} if model else {}) + provider.get_info.return_value = info + + # MagicMock auto-creates attributes, so hasattr(provider, "priority") is + # always True unless we explicitly remove it to test the fallback paths. + if priority is not None: + provider.priority = priority + else: + del provider.priority + + if config_priority is not None: + provider.config = {"priority": config_priority} + else: + provider.config = {} + + return provider + + +def _make_pin(available=None, current=None, pin_side_effect=None): + """Build a mock 'conversation.provider_pin' capability object.""" + pin = MagicMock() + pin.available.return_value = available or [] + pin.current.return_value = current + pin.unpin.return_value = None # idempotent-unpin default; override per-test + if pin_side_effect is not None: + pin.pin.side_effect = pin_side_effect + return pin + + +def _cp_with(pin=None, providers=None, orchestrator=None): + """CommandProcessor whose coordinator returns `pin` for + get_capability('conversation.provider_pin') and `providers` for + get('providers'), with an optional orchestrator name in coordinator.config. + """ + cp = _make_command_processor() + coordinator = cp.session.coordinator + + def _get_capability(key): + if key == "conversation.provider_pin": + return pin + return None + + coordinator.get_capability = _get_capability + + def _get(key): + if key == "providers": + return providers or {} + return None + + coordinator.get = _get + + coordinator.config = ( + {"session": {"orchestrator": orchestrator}} if orchestrator else {} + ) + return cp + + +# === Capability absent: refuse loudly, never report success === + + +class TestCapabilityAbsent: + @pytest.mark.asyncio + async def test_status_names_orchestrator_and_says_restart_required(self): + cp = _cp_with(pin=None, providers={}, orchestrator="loop-basic") + result = await cp._handle_provider("") + assert "loop-basic" in result + assert "conversation.provider_pin" in result + assert "not registered" in result + assert "restarting" in result + + @pytest.mark.asyncio + async def test_status_without_known_orchestrator_name_still_refuses(self): + cp = _cp_with(pin=None, providers={}) + result = await cp._handle_provider("") + assert "not supported" in result + assert "restarting" in result + + @pytest.mark.asyncio + async def test_use_refuses_and_never_calls_pin(self): + cp = _cp_with(pin=None, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("use anthropic-fable") + assert "not registered" in result + assert "restarting" in result + # No success language anywhere in the refusal. + assert "Pinned" not in result + + @pytest.mark.asyncio + async def test_auto_refuses_and_never_claims_unpin(self): + cp = _cp_with(pin=None, providers={}) + result = await cp._handle_provider("auto") + assert "not registered" in result + assert "Unpinned" not in result + + @pytest.mark.asyncio + async def test_status_still_lists_mounted_providers_when_capability_absent(self): + """Even without pinning support, /provider is useful for seeing what's + mounted -- that read comes straight from the kernel 'providers' mount + point and doesn't require the capability.""" + cp = _cp_with( + pin=None, + providers={"anthropic-fable": _make_provider(model="claude-sonnet-4-5")}, + ) + result = await cp._handle_provider("") + assert "anthropic-fable" in result + assert "claude-sonnet-4-5" in result + + +# === Status display when capability IS present === + + +class TestStatusDisplay: + @pytest.mark.asyncio + async def test_no_providers_mounted(self): + cp = _cp_with(pin=_make_pin(), providers={}) + result = await cp._handle_provider("") + assert "none mounted" in result + + @pytest.mark.asyncio + async def test_shows_model_and_priority_for_each_provider(self): + providers = { + "anthropic-fable": _make_provider(model="claude-sonnet-4-5", priority=1), + "openai-gpt5": _make_provider(model="gpt-5", priority=2), + } + pin = _make_pin(available=list(providers), current=None) + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert "anthropic-fable" in result + assert "claude-sonnet-4-5" in result + assert "priority=1" in result + assert "openai-gpt5" in result + assert "gpt-5" in result + assert "priority=2" in result + + @pytest.mark.asyncio + async def test_unpinned_marks_priority_winner_active_and_states_automatic(self): + providers = { + "anthropic-fable": _make_provider(model="claude-sonnet-4-5", priority=1), + "openai-gpt5": _make_provider(model="gpt-5", priority=2), + } + pin = _make_pin(available=list(providers), current=None) + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert "automatic" in result + assert "active by priority" in result + # The lower-priority-number provider is the one marked active. + lines = result.splitlines() + winner_line = next(line_ for line_ in lines if "anthropic-fable" in line_) + assert "active by priority" in winner_line + loser_line = next(line_ for line_ in lines if "openai-gpt5" in line_) + assert "active by priority" not in loser_line + + @pytest.mark.asyncio + async def test_pinned_marks_pinned_provider_active_not_priority_winner(self): + providers = { + "anthropic-fable": _make_provider(model="claude-sonnet-4-5", priority=1), + "openai-gpt5": _make_provider(model="gpt-5", priority=2), + } + # openai-gpt5 is pinned despite NOT having priority-winning rank. + pin = _make_pin(available=list(providers), current="openai-gpt5") + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert "pinned to 'openai-gpt5'" in result + lines = result.splitlines() + pinned_line = next(line_ for line_ in lines if "openai-gpt5" in line_) + assert "[pinned, active]" in pinned_line + other_line = next(line_ for line_ in lines if "anthropic-fable" in line_) + assert "active" not in other_line + + @pytest.mark.asyncio + async def test_falls_back_to_config_priority_when_no_priority_attr(self): + providers = { + "anthropic-fable": _make_provider( + model="claude-sonnet-4-5", config_priority=5 + ), + } + pin = _make_pin(available=list(providers), current=None) + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert "priority=5" in result + + @pytest.mark.asyncio + async def test_defaults_priority_to_100_when_unspecified(self): + providers = {"anthropic-fable": _make_provider(model="claude-sonnet-4-5")} + pin = _make_pin(available=list(providers), current=None) + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert "priority=100" in result + + @pytest.mark.asyncio + async def test_unknown_model_shown_when_get_info_lacks_it(self): + providers = {"anthropic-fable": _make_provider(model=None)} + pin = _make_pin(available=list(providers), current=None) + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert "model=(unknown)" in result + + +# === /provider use : pin === + + +class TestProviderUse: + @pytest.mark.asyncio + async def test_success_calls_pin_and_reports_next_turn_not_now(self): + pin = _make_pin(available=["anthropic-fable"]) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("use anthropic-fable") + pin.pin.assert_called_once_with("anthropic-fable") + assert "Pinned conversation provider to 'anthropic-fable'" in result + assert "NEXT turn" in result + # Must not claim the switch already happened, and must not compete + # with the per-turn usage line's "now using X" confirmation. + assert "now using" not in result.lower() + + @pytest.mark.asyncio + async def test_states_scope_is_top_level_conversation_only(self): + pin = _make_pin(available=["anthropic-fable"]) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("use anthropic-fable") + assert "top-level conversation only" in result + assert "model-role routing" in result + assert "sub-agents" in result + assert "/goal loop" in result + + @pytest.mark.asyncio + async def test_invalid_name_renders_clean_error_not_traceback(self): + def _raise(name): + raise ValueError( + f"cannot pin conversation provider {name!r}: it is not mounted " + f"in this session. Mounted providers: anthropic-fable" + ) + + pin = _make_pin(available=["anthropic-fable"], pin_side_effect=_raise) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("use nonexistent-provider") + assert "Traceback" not in result + assert "nonexistent-provider" in result + assert "anthropic-fable" in result # lists what IS available + + @pytest.mark.asyncio + async def test_missing_name_shows_usage(self): + pin = _make_pin(available=["anthropic-fable"]) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("use") + assert "Usage: /provider use " in result + pin.pin.assert_not_called() + + +# === /provider auto: unpin === + + +class TestProviderAuto: + @pytest.mark.asyncio + async def test_success_calls_unpin_and_reports_previous(self): + pin = _make_pin(available=["anthropic-fable"]) + pin.unpin.return_value = "anthropic-fable" + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("auto") + pin.unpin.assert_called_once() + assert "Unpinned conversation provider (was 'anthropic-fable')" in result + assert "NEXT turn" in result + + @pytest.mark.asyncio + async def test_already_unpinned_is_idempotent_and_says_so(self): + pin = _make_pin(available=["anthropic-fable"]) + pin.unpin.return_value = None + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("auto") + assert "already automatic" in result + assert "Nothing to unpin" in result + + @pytest.mark.asyncio + async def test_states_scope_is_top_level_conversation_only(self): + pin = _make_pin(available=["anthropic-fable"]) + pin.unpin.return_value = "anthropic-fable" + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("auto") + assert "top-level conversation only" in result + assert "/goal loop" in result + + +# === Unknown subcommand === + + +class TestUnknownSubcommand: + @pytest.mark.asyncio + async def test_unknown_subcommand_shows_usage(self): + pin = _make_pin(available=["anthropic-fable"]) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("frobnicate") + assert "Unknown /provider subcommand" in result + assert "/provider use " in result + assert "/provider auto" in result + + +# === Registration / discoverability === + + +class TestCommandRegistration: + def test_provider_registered_in_commands_dict(self): + from amplifier_app_cli.main import CommandProcessor + + assert "/provider" in CommandProcessor.COMMANDS + assert CommandProcessor.COMMANDS["/provider"]["action"] == "handle_provider" + + @pytest.mark.asyncio + async def test_help_output_includes_provider_command(self): + cp = _make_command_processor() + help_text = cp._format_help() + assert "/provider" in help_text + + @pytest.mark.asyncio + async def test_handle_command_dispatches_to_handle_provider(self): + pin = _make_pin(available=["anthropic-fable"]) + cp = _cp_with(pin=pin, providers={}) + result = await cp.handle_command("handle_provider", {"args": "auto"}) + assert "already automatic" in result + + +# === (experimental) tagging === +# +# The tag goes where the user sees the feature working -- help, status, and +# the confirmations -- and deliberately NOT on the error paths, which are +# already loud and would only be diluted by it. + + +class TestExperimentalTag: + def test_help_entry_is_tagged(self): + from amplifier_app_cli.main import CommandProcessor + + assert "(experimental)" in CommandProcessor.COMMANDS["/provider"]["description"] + + @pytest.mark.asyncio + async def test_help_output_shows_tag(self): + cp = _make_command_processor() + help_text = cp._format_help() + provider_line = next( + line for line in help_text.splitlines() if line.startswith(" /provider") + ) + assert "(experimental)" in provider_line + + @pytest.mark.asyncio + async def test_status_header_is_tagged(self): + providers = {"anthropic-fable": _make_provider(model="claude-sonnet-4-5")} + pin = _make_pin(available=list(providers), current=None) + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert result.splitlines()[0] == "Conversation providers (experimental):" + + @pytest.mark.asyncio + async def test_status_header_tagged_even_with_no_providers(self): + cp = _cp_with(pin=_make_pin(), providers={}) + result = await cp._handle_provider("") + assert result.splitlines()[0] == "Conversation providers (experimental):" + + @pytest.mark.asyncio + async def test_use_confirmation_is_tagged(self): + pin = _make_pin(available=["anthropic-fable"]) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("use anthropic-fable") + assert result.startswith("(experimental) ") + + @pytest.mark.asyncio + async def test_auto_confirmation_is_tagged(self): + pin = _make_pin(available=["anthropic-fable"]) + pin.unpin.return_value = "anthropic-fable" + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("auto") + assert result.startswith("(experimental) ") + + @pytest.mark.asyncio + async def test_auto_already_unpinned_confirmation_is_tagged(self): + """Both /provider auto outcomes carry the tag -- otherwise the same + command would appear tagged or untagged depending on prior state, + which reads like a bug.""" + pin = _make_pin(available=["anthropic-fable"]) + pin.unpin.return_value = None + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("auto") + assert result.startswith("(experimental) ") + + @pytest.mark.asyncio + async def test_capability_absent_error_is_NOT_tagged(self): + cp = _cp_with(pin=None, providers={}, orchestrator="loop-basic") + result = await cp._handle_provider("use anthropic-fable") + assert "(experimental)" not in result + + @pytest.mark.asyncio + async def test_unmounted_provider_error_is_NOT_tagged(self): + def _raise(name): + raise ValueError( + f"cannot pin conversation provider {name!r}: it is not mounted " + f"in this session. Mounted providers: anthropic-fable" + ) + + pin = _make_pin(available=["anthropic-fable"], pin_side_effect=_raise) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("use nonexistent-provider") + assert "(experimental)" not in result + + +# === Usage-figure caveat === +# +# PLACEMENT DECISION (locked in by these tests): the caveat lives in the +# /provider STATUS view only, never on the per-use confirmations. The +# confirmations are already 320 and 284 chars -- 5 and 4 wrapped lines at +# 80 cols -- and the condition being disclaimed is a property of the CLI's +# usage display, not of pinning, so repeating it on every pin would grow +# the noisiest string for the least benefit. +# +# ACCURACY: per-vendor rates are correct and never cross-applied. The caveat +# must stay framed as precision/rounding/coverage, never "costs are wrong". + + +def _caveat() -> str: + from amplifier_app_cli.main import CommandProcessor + + return CommandProcessor._PROVIDER_USAGE_CAVEAT + + +class TestUsageCaveat: + @pytest.mark.asyncio + async def test_status_unpinned_shows_caveat(self): + providers = {"anthropic-fable": _make_provider(model="claude-sonnet-4-5")} + pin = _make_pin(available=list(providers), current=None) + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert _caveat() in result + + @pytest.mark.asyncio + async def test_status_pinned_shows_caveat(self): + providers = {"anthropic-fable": _make_provider(model="claude-sonnet-4-5")} + pin = _make_pin(available=list(providers), current="anthropic-fable") + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert _caveat() in result + + @pytest.mark.asyncio + async def test_capability_absent_status_does_NOT_show_caveat(self): + """When pinning isn't supported the refusal is the whole message -- + a cost footnote there dilutes it, same principle that keeps the + (experimental) tag off the error paths.""" + providers = {"anthropic-fable": _make_provider(model="claude-sonnet-4-5")} + cp = _cp_with(pin=None, providers=providers, orchestrator="loop-basic") + result = await cp._handle_provider("") + assert _caveat() not in result + assert "Usage figures:" not in result + # The refusal itself must still be fully intact. + assert "not registered" in result + assert "restarting" in result + + @pytest.mark.asyncio + async def test_no_providers_mounted_does_NOT_show_caveat(self): + """No providers means no usage figures to disclaim.""" + cp = _cp_with(pin=_make_pin(), providers={}) + result = await cp._handle_provider("") + assert "Usage figures:" not in result + + @pytest.mark.asyncio + async def test_use_confirmation_does_NOT_carry_caveat(self): + pin = _make_pin(available=["anthropic-fable"]) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("use anthropic-fable") + assert "Usage figures:" not in result + assert "billing-grade" not in result + + @pytest.mark.asyncio + async def test_auto_confirmations_do_NOT_carry_caveat(self): + for unpin_result in ("anthropic-fable", None): + pin = _make_pin(available=["anthropic-fable"]) + pin.unpin.return_value = unpin_result + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + result = await cp._handle_provider("auto") + assert "Usage figures:" not in result + assert "billing-grade" not in result + + @pytest.mark.asyncio + async def test_confirmations_do_not_grow_past_their_current_wrapping(self): + """The confirmations already wrap to 5 (use) and 4 (auto) lines at + 80 cols. This is a proxy guard (textwrap, not a real terminal) but + it fails loudly if someone appends the caveat -- or any other + paragraph -- to the strings the user sees on EVERY pin.""" + import textwrap + + pin = _make_pin(available=["anthropic-fable"]) + cp = _cp_with(pin=pin, providers={"anthropic-fable": _make_provider()}) + use_msg = await cp._handle_provider("use anthropic-fable") + + pin2 = _make_pin(available=["anthropic-fable"]) + pin2.unpin.return_value = "anthropic-fable" + cp2 = _cp_with(pin=pin2, providers={"anthropic-fable": _make_provider()}) + auto_msg = await cp2._handle_provider("auto") + + assert len(textwrap.wrap(use_msg, 80)) <= 5, ( + "/provider use confirmation grew past 5 wrapped lines at 80 cols" + ) + assert len(textwrap.wrap(auto_msg, 80)) <= 4, ( + "/provider auto confirmation grew past 4 wrapped lines at 80 cols" + ) + + def test_caveat_does_not_claim_rates_are_wrong_or_cross_applied(self): + """Measured across four providers against raw events.jsonl: + per-vendor rates ARE correct and are never cross-applied. Saying + otherwise would be inaccurate and needlessly alarming.""" + caveat = _caveat().lower() + for forbidden in ( + "wrong", + "incorrect", + "inaccurate", + "cross-appl", + "mispriced", + "overcharg", + "unreliable", + ): + assert forbidden not in caveat, ( + f"caveat overstates the defect with {forbidden!r} -- the " + f"honest framing is precision/rounding/coverage" + ) + + def test_caveat_states_rates_themselves_are_correct(self): + """The anti-alarm clause is load-bearing: without it the caveat + reads as 'costs are wrong', which is not what was measured.""" + caveat = _caveat().lower() + assert "rates" in caveat + assert "correct" in caveat + + def test_caveat_covers_the_four_measured_defect_classes(self): + """Counts (over- and under-reporting), rounding, and missing + coverage for the newest models.""" + caveat = _caveat().lower() + assert "over- or under-reported" in caveat + assert "rounded" in caveat + assert "newest models" in caveat + + def test_caveat_stays_short(self): + """It is a footnote, not a paragraph. Kept under 250 chars so the + status view stays scannable.""" + assert len(_caveat()) <= 250, ( + f"usage caveat grew to {len(_caveat())} chars -- keep it a footnote" + ) + + @pytest.mark.asyncio + async def test_caveat_is_the_last_line_of_status(self): + """Reads as a footnote under the Selection line, not as a banner + competing with the provider table.""" + providers = {"anthropic-fable": _make_provider(model="claude-sonnet-4-5")} + pin = _make_pin(available=list(providers), current=None) + cp = _cp_with(pin=pin, providers=providers) + result = await cp._handle_provider("") + assert result.splitlines()[-1] == _caveat() + + +# === Prompt indicator === +# +# Approved mockup (mode leftmost, pin second, caret last): +# +# unpinned: > +# pinned: [PIN anthropic-haiku]> +# pinned + mode: [plan][PIN anthropic-haiku]> +# +# ("PIN" above is the U+1F4CC pushpin glyph.) + +_PUSHPIN = "\U0001f4cc" + + +def _rendered(message): + """The visible text of a prompt_toolkit HTML message, markup stripped. + + Also proves HTML() actually parses -- a markup error would raise here. + """ + from prompt_toolkit.formatted_text import to_formatted_text + + return "".join(fragment[1] for fragment in to_formatted_text(message)) + + +class TestPromptIndicator: + def test_unpinned_no_mode_renders_bare_caret(self): + from amplifier_app_cli.main import _build_prompt_message + + message = _build_prompt_message(lambda: None, lambda: None) + assert _rendered(message) == "\n> " + + def test_unpinned_no_mode_markup_is_byte_for_byte_unchanged(self): + """The pre-pin prompt markup, exactly. If this drifts, every user + who never touches /provider sees a changed prompt.""" + from amplifier_app_cli.main import _build_prompt_message + + message = _build_prompt_message(lambda: None, lambda: None) + assert message.value == "\n> " + + def test_no_getters_at_all_renders_bare_caret(self): + """Absent getters (the pre-feature call signature) must behave + exactly like getters that return None.""" + from amplifier_app_cli.main import _build_prompt_message + + message = _build_prompt_message() + assert message.value == "\n> " + + def test_mode_only_markup_is_byte_for_byte_unchanged(self): + from amplifier_app_cli.main import _build_prompt_message + + message = _build_prompt_message(lambda: "plan", lambda: None) + assert ( + message.value + == "\n[plan]> " + ) + assert _rendered(message) == "\n[plan]> " + + def test_pinned_only_matches_mockup(self): + from amplifier_app_cli.main import _build_prompt_message + + message = _build_prompt_message(lambda: None, lambda: "anthropic-haiku") + assert _rendered(message) == f"\n[{_PUSHPIN} anthropic-haiku]> " + + def test_pinned_plus_mode_matches_mockup_with_mode_leftmost(self): + from amplifier_app_cli.main import _build_prompt_message + + message = _build_prompt_message(lambda: "plan", lambda: "anthropic-haiku") + assert _rendered(message) == f"\n[plan][{_PUSHPIN} anthropic-haiku]> " + + def test_provider_name_is_never_truncated(self): + """Several mount names share a model family (anthropic-fable / + -opus / -sonnet / -haiku); shortening destroys the distinction the + indicator exists to show.""" + from amplifier_app_cli.main import _build_prompt_message + + long_name = "anthropic-fable-experimental-long-mount-name" + message = _build_prompt_message(lambda: None, lambda: long_name) + assert long_name in _rendered(message) + + def test_pin_getter_raising_does_not_break_prompt(self): + """The prompt callable runs on every keystroke -- it must never + raise, or the session becomes unusable.""" + from amplifier_app_cli.main import _build_prompt_message + + def _boom(): + raise RuntimeError("capability exploded") + + message = _build_prompt_message(lambda: None, _boom) + assert _rendered(message) == "\n> " + + def test_mode_getter_raising_does_not_suppress_pin_indicator(self): + """Each indicator degrades independently -- one broken getter must + not take the other's indicator down with it.""" + from amplifier_app_cli.main import _build_prompt_message + + def _boom(): + raise RuntimeError("mode lookup exploded") + + message = _build_prompt_message(_boom, lambda: "anthropic-haiku") + assert _rendered(message) == f"\n[{_PUSHPIN} anthropic-haiku]> " + + def test_markup_special_characters_in_name_do_not_raise(self): + """A '<' or '&' in a name would otherwise make HTML() raise.""" + from amplifier_app_cli.main import _build_prompt_message + + message = _build_prompt_message(lambda: None, lambda: "weird<&>name") + assert _rendered(message) == f"\n[{_PUSHPIN} weird<&>name]> " + + def test_empty_string_pin_is_treated_as_unpinned(self): + from amplifier_app_cli.main import _build_prompt_message + + message = _build_prompt_message(lambda: None, lambda: "") + assert message.value == "\n> " + + +# === _pinned_provider_name: the prompt's source of truth === + + +class TestPinnedProviderName: + def test_returns_none_when_capability_absent(self): + from amplifier_app_cli.main import _pinned_provider_name + + cp = _cp_with(pin=None, providers={}) + assert _pinned_provider_name(cp.session) is None + + def test_returns_none_when_nothing_pinned(self): + from amplifier_app_cli.main import _pinned_provider_name + + cp = _cp_with(pin=_make_pin(current=None), providers={}) + assert _pinned_provider_name(cp.session) is None + + def test_returns_pinned_name(self): + from amplifier_app_cli.main import _pinned_provider_name + + cp = _cp_with(pin=_make_pin(current="anthropic-haiku"), providers={}) + assert _pinned_provider_name(cp.session) == "anthropic-haiku" + + def test_non_string_current_is_treated_as_unpinned(self): + from amplifier_app_cli.main import _pinned_provider_name + + cp = _cp_with(pin=_make_pin(current=object()), providers={}) + assert _pinned_provider_name(cp.session) is None + + @pytest.mark.asyncio + async def test_tracks_pin_and_unpin_through_the_command(self): + """End-to-end within the app layer: the indicator source follows + /provider use and /provider auto without any app-side copy of the + pin state.""" + from amplifier_app_cli.main import _build_prompt_message, _pinned_provider_name + + state = {"pinned": None} + pin = _make_pin(available=["anthropic-haiku"]) + pin.current.side_effect = lambda: state["pinned"] + + def _pin(name): + state["pinned"] = name + return name + + def _unpin(): + previous = state["pinned"] + state["pinned"] = None + return previous + + pin.pin.side_effect = _pin + pin.unpin.side_effect = _unpin + + cp = _cp_with(pin=pin, providers={"anthropic-haiku": _make_provider()}) + + def getter(): + return _pinned_provider_name(cp.session) + + assert _rendered(_build_prompt_message(lambda: None, getter)) == "\n> " + + await cp._handle_provider("use anthropic-haiku") + assert ( + _rendered(_build_prompt_message(lambda: None, getter)) + == f"\n[{_PUSHPIN} anthropic-haiku]> " + ) + + await cp._handle_provider("auto") + assert _rendered(_build_prompt_message(lambda: None, getter)) == "\n> " + + +# === WIRING: the prompt callable PromptSession actually renders === +# +# REGRESSION GUARD. Every test above this line calls _build_prompt_message +# directly. That is exactly the coverage that already existed and did NOT +# catch the real bug: _create_prompt_session() carried its own duplicate +# get_prompt() closure that handled the mode indicator but never referenced +# get_pinned_provider, and PromptSession(message=...) was wired to THAT copy. +# The composer was correct, fully unit-tested, and never called at runtime -- +# 48 passing tests, zero of which touched _create_prompt_session. +# +# So these tests deliberately do NOT call the composer. They construct the +# real PromptSession through _create_prompt_session() and resolve the message +# through the constructed object, the same way prompt_toolkit does at render +# time. prompt_toolkit's own render path is: +# +# PromptSession._get_prompt(self): +# return to_formatted_text(self.message, style="class:prompt") +# +# (prompt_toolkit/shortcuts/prompt.py -- `self.message = message` at __init__, +# resolved via to_formatted_text at render). So `session.message` IS the seam, +# and resolving it through to_formatted_text is what the library itself does. +# +# A real PromptSession CAN be constructed headless -- no TTY needed. It emits +# "Warning: Input is not a terminal (fd=0)" and works, so there is no need to +# settle for anything short of the real object here. + + +@pytest.fixture +def isolated_home(tmp_path, monkeypatch): + """Redirect Path.home() so constructing a real PromptSession cannot touch + the developer's actual ~/.amplifier/projects//repl_history.""" + from pathlib import Path as _Path + + monkeypatch.setattr(_Path, "home", staticmethod(lambda: tmp_path)) + return tmp_path + + +def _prompt_session(mode=None, pinned=None): + """Build the REAL PromptSession through the REAL factory.""" + from amplifier_app_cli.main import _create_prompt_session + + return _create_prompt_session( + get_active_mode=lambda: mode, + get_pinned_provider=lambda: pinned, + ) + + +def _render_via_prompt_session(session) -> str: + """Resolve the prompt the way prompt_toolkit does at render time. + + Mirrors PromptSession._get_prompt() -- to_formatted_text(self.message) -- + so this exercises the wiring, not a composer called in isolation. + """ + from prompt_toolkit.formatted_text import to_formatted_text + + return "".join(fragment[1] for fragment in to_formatted_text(session.message)) + + +@pytest.mark.usefixtures("isolated_home") +class TestPromptSessionWiring: + def test_message_is_a_callable_not_a_prebuilt_value(self): + """A static value would freeze the indicator at construction time -- + the pin must be re-read on every render.""" + session = _prompt_session(mode=None, pinned="anthropic-haiku") + assert callable(session.message) + + def test_unpinned_renders_bare_prompt_through_session(self): + session = _prompt_session(mode=None, pinned=None) + assert _render_via_prompt_session(session) == "\n> " + + def test_pinned_renders_pin_indicator_through_session(self): + """THE regression: this is the assertion the old suite never made. + It fails against the duplicate closure that ignored the pin getter.""" + session = _prompt_session(mode=None, pinned="anthropic-haiku") + assert ( + _render_via_prompt_session(session) == f"\n[{_PUSHPIN} anthropic-haiku]> " + ) + + def test_mode_only_renders_mode_indicator_through_session(self): + session = _prompt_session(mode="plan", pinned=None) + assert _render_via_prompt_session(session) == "\n[plan]> " + + def test_pinned_plus_mode_renders_both_with_mode_leftmost(self): + session = _prompt_session(mode="plan", pinned="anthropic-haiku") + assert ( + _render_via_prompt_session(session) + == f"\n[plan][{_PUSHPIN} anthropic-haiku]> " + ) + + def test_prompt_toolkits_own_get_prompt_path_renders_the_indicators(self): + """Belt and braces: drive prompt_toolkit's ACTUAL internal render + method rather than our reimplementation of it. If PromptSession stops + routing through _get_prompt, this skips rather than lying.""" + session = _prompt_session(mode="plan", pinned="anthropic-haiku") + get_prompt = getattr(session, "_get_prompt", None) + if get_prompt is None: # pragma: no cover - prompt_toolkit API drift + pytest.skip("prompt_toolkit no longer exposes PromptSession._get_prompt") + rendered = "".join(fragment[1] for fragment in get_prompt()) + assert rendered == f"\n[plan][{_PUSHPIN} anthropic-haiku]> " + + def test_message_is_re_evaluated_on_every_render(self): + """The pin can change mid-session (/provider use, /provider auto), so + a cached first render would show a stale indicator forever.""" + from amplifier_app_cli.main import _create_prompt_session + + state: dict[str, str | None] = {"pinned": None} + session = _create_prompt_session( + get_active_mode=lambda: None, + get_pinned_provider=lambda: state["pinned"], + ) + assert _render_via_prompt_session(session) == "\n> " + state["pinned"] = "anthropic-haiku" + assert ( + _render_via_prompt_session(session) == f"\n[{_PUSHPIN} anthropic-haiku]> " + ) + state["pinned"] = None + assert _render_via_prompt_session(session) == "\n> " + + def test_every_getter_passed_to_the_factory_is_actually_consulted(self): + """Generalized guard for the whole bug class, independent of what the + getters return: a partial closure that never references one of them + records zero calls on that spy and fails here. This is what makes + 'someone reintroduces a closure that ignores a getter' non-silent + even for a getter whose value happens to be empty.""" + from amplifier_app_cli.main import _create_prompt_session + + calls = {"mode": 0, "pin": 0} + + # Both deliberately yield an EMPTY value (implicit None): the spy + # proves consultation, so this catches an ignored getter even when + # its value would not have shown an indicator anyway. + def _mode(): + calls["mode"] += 1 + + def _pin(): + calls["pin"] += 1 + + session = _create_prompt_session( + get_active_mode=_mode, get_pinned_provider=_pin + ) + _render_via_prompt_session(session) + + assert calls["mode"] > 0, ( + "get_active_mode was never consulted when the prompt rendered -- " + "PromptSession.message is not wired to the composer" + ) + assert calls["pin"] > 0, ( + "get_pinned_provider was never consulted when the prompt rendered " + "-- this is the exact bug: a duplicate closure that ignores it" + ) + + def test_factory_delegates_to_the_single_composer(self): + """Structural backstop: the factory's message callable must route + through _build_prompt_message. Patching the composer must change what + the constructed session renders -- if it doesn't, a second + implementation has been reintroduced somewhere in the factory.""" + from unittest.mock import patch + + from prompt_toolkit.formatted_text import HTML + + with patch( + "amplifier_app_cli.main._build_prompt_message", + return_value=HTML("SENTINEL"), + ) as composer: + session = _prompt_session(mode="plan", pinned="anthropic-haiku") + rendered = _render_via_prompt_session(session) + + assert rendered == "SENTINEL", ( + "the constructed PromptSession did not render through " + "_build_prompt_message -- a duplicate prompt implementation exists" + ) + assert composer.called + + def test_factory_still_works_with_no_getters(self): + """The pre-feature call signature must keep working -- a user who + never touches /provider sees the byte-for-byte original prompt.""" + from amplifier_app_cli.main import _create_prompt_session + + session = _create_prompt_session() + assert _render_via_prompt_session(session) == "\n> " + + def test_raising_getter_does_not_break_the_constructed_prompt(self): + """Through the real wiring, not just the composer: the render path + runs on every keystroke and must never raise.""" + from amplifier_app_cli.main import _create_prompt_session + + def _boom(): + raise RuntimeError("capability exploded") + + session = _create_prompt_session( + get_active_mode=lambda: None, get_pinned_provider=_boom + ) + assert _render_via_prompt_session(session) == "\n> "