feat: add isolated provider-pinned /deep-plan workflow

amplifier_app_cli/interrupt.py

@@ -0,0 +1,66 @@
+"""Shared interactive SIGINT handling for asynchronous CLI work."""
+
+from __future__ import annotations
+
+import asyncio
+import signal
+from collections.abc import Awaitable
+from contextlib import suppress
+from typing import Any, TypeVar
+
+_ResultT = TypeVar("_ResultT")
+
+
+async def run_with_interrupt(
+    awaitable: Awaitable[_ResultT],
+    *,
+    cancellation: Any,
+    console: Any,
+) -> _ResultT:
+    """Await work with the CLI's graceful-then-immediate Ctrl+C behavior.
+
+    The first interrupt updates the coordinator cancellation token, allowing it
+    to propagate to registered child sessions. A second interrupt cancels the
+    local task immediately. Callers remain responsible for interpreting a
+    graceful cancellation after the awaitable returns.
+    """
+
+    cancellation.reset()
+
+    def _handle_sigint(_signum: int, _frame: Any) -> None:
+        # CancellationToken updates are intentionally synchronous. Scheduling
+        # these writes would race when a user presses Ctrl+C twice quickly.
+        if cancellation.is_cancelled:
+            cancellation.request_immediate()
+            console.print("\n[bold red]Cancelling immediately...[/bold red]")
+            return
+
+        cancellation.request_graceful()
+        running_tools = cancellation.running_tool_names
+        if running_tools:
+            tools = ", ".join(running_tools)
+            console.print(
+                "\n[yellow]Stopping after current operation in "
+                f"[bold]{tools}[/bold]... (Ctrl+C again to force)[/yellow]"
+            )
+        else:
+            console.print(
+                "\n[yellow]Stopping after current operation completes... "
+                "(Ctrl+C again to force)[/yellow]"
+            )
+
+    task = asyncio.ensure_future(awaitable)
+    original_handler = signal.signal(signal.SIGINT, _handle_sigint)
+    try:
+        while not task.done():
+            if cancellation.is_immediate:
+                task.cancel()
+                break
+            await asyncio.sleep(0.05)
+        return await task
+    finally:
+        signal.signal(signal.SIGINT, original_handler)
+        if not task.done():
+            task.cancel()
+            with suppress(asyncio.CancelledError):
+                await task

amplifier_app_cli/lib/settings.py

@@ -9,12 +9,10 @@
 import tempfile
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
-from typing import Literal
+from typing import Any, Literal
 
 import yaml
-from filelock import BaseFileLock
-from filelock import FileLock
+from filelock import BaseFileLock, FileLock
 
 Scope = Literal["local", "project", "global", "session"]
 
@@ -136,6 +134,20 @@ def get_merged_settings(self) -> dict[str, Any]:
                     pass  # Skip malformed files
         return result
 
+    def get_deep_plan_provider(self) -> str:
+        """Return the configured deep-plan provider, validating explicit values."""
+
+        from amplifier_app_cli.deep_plan import resolve_planner_provider
+
+        return resolve_planner_provider(self.get_merged_settings())
+
+    def get_deep_plan_config(self) -> Any:
+        """Return validated provider, exact model, and effort for deep planning."""
+
+        from amplifier_app_cli.deep_plan import resolve_planner_config
+
+        return resolve_planner_config(self.get_merged_settings())
+
     # ----- Bundle settings -----
 
     def get_active_bundle(self) -> str | None:

amplifier_app_cli/main.py

@@ -67,6 +67,7 @@
 from .console import Markdown, console
 from .dedicated_tty_input import close_dedicated_tty_input, get_dedicated_tty_input
 from .effective_config import get_effective_config_summary
+from .interrupt import run_with_interrupt
 from .key_manager import KeyManager
 from .provider_diagnostics import DEFAULT_TIMEOUT_S as _PROVIDER_DIAGNOSTIC_TIMEOUT_S
 from .provider_diagnostics import format_model_line
@@ -524,6 +525,12 @@ class CommandProcessor:
                 "/provider test <name> | /provider models <name>"
             ),
         },
+        "/deep-plan": {
+            "action": "deep_plan",
+            "description": (
+                "Plan with the configured premium provider, then execute the task"
+            ),
+        },
     }
 
     # Dynamic shortcuts for modes (populated from mode definitions)
@@ -807,6 +814,9 @@ async def handle_command(self, action: str, data: dict[str, Any]) -> str:
         if action == "handle_provider":
             return await self._handle_provider(data.get("args", ""))
 
+        if action == "deep_plan":
+            return "Use /deep-plan <task>."
+
         if action == "list_modes":
             return await self._list_modes()
 
@@ -3637,51 +3647,19 @@ async def _repair_transcript_if_needed():
             logger.debug("Pre-turn transcript repair failed: %s", e)
 
     # Helper to execute a prompt with Ctrl+C handling
-    async def _execute_with_interrupt(prompt_text: str) -> bool:
-        """Execute prompt with interrupt handling. Returns True if completed, False if cancelled."""
+    async def _execute_with_interrupt(
+        prompt_text: str, *, manage_interrupt: bool = True
+    ) -> bool:
+        """Execute one prompt, optionally installing this turn's SIGINT handler.
+
+        ``/deep-plan`` wraps both its planner and parent turn in one outer
+        interrupt scope, so its parent execution must not reset cancellation
+        or replace that handler at the planning-to-execution handoff.
+        """
         # Pre-turn transcript repair: detect and fix any orphaned tool calls,
         # ordering violations, or incomplete turns before the next LLM call.
         await _repair_transcript_if_needed()
 
-        # Reset cancellation state for new execution
-        session.coordinator.cancellation.reset()
-
-        def sigint_handler(signum, frame):
-            """Handle Ctrl+C with graceful/immediate cancellation.
-
-            CRITICAL: State updates must be SYNCHRONOUS to avoid race conditions.
-            If we used async scheduling (call_soon_threadsafe + create_task), rapid
-            double Ctrl+C could be mishandled because the first state update might
-            not complete before the second signal arrives.
-
-            The CancellationToken's request_graceful() and request_immediate() methods
-            are synchronous, so we call them directly here.
-            """
-            cancellation = session.coordinator.cancellation
-
-            if cancellation.is_cancelled:
-                # Second Ctrl+C - request immediate cancellation
-                # SYNC state update to avoid race condition with rapid double Ctrl+C
-                cancellation.request_immediate()
-                console.print("\n[bold red]Cancelling immediately...[/bold red]")
-            else:
-                # First Ctrl+C - request graceful cancellation
-                # SYNC state update to ensure state is set before any second signal
-                cancellation.request_graceful()
-                # Show what's running
-                running_tools = cancellation.running_tool_names
-                if running_tools:
-                    tools_str = ", ".join(running_tools)
-                    console.print(
-                        f"\n[yellow]Stopping after current operation in [bold]{tools_str}[/bold]... (Ctrl+C again to force)[/yellow]"
-                    )
-                else:
-                    console.print(
-                        "\n[yellow]Stopping after current operation completes... (Ctrl+C again to force)[/yellow]"
-                    )
-
-        original_handler = signal.signal(signal.SIGINT, sigint_handler)
-
         # Mid-turn steering: create the anchored-input manager.
         # patch_stdout() (below) ensures all Rich console.print calls that
         # originate from session.execute() or hooks appear ABOVE the pinned
@@ -3765,18 +3743,17 @@ def sigint_handler(signum, frame):
                 _reader_task = asyncio.create_task(_manager.run())
 
                 try:
-                    execute_task = asyncio.create_task(session.execute(prompt_text))
-
-                    # Poll task while checking for cancellation
-                    while not execute_task.done():
-                        # Check for immediate cancellation - cancel the task
-                        if session.coordinator.cancellation.is_immediate:
-                            execute_task.cancel()
-                            break
-                        await asyncio.sleep(0.05)
-
                     try:
-                        response = await execute_task
+                        if manage_interrupt:
+                            response = await run_with_interrupt(
+                                session.execute(prompt_text),
+                                cancellation=session.coordinator.cancellation,
+                                console=console,
+                            )
+                        else:
+                            if session.coordinator.cancellation.is_cancelled:
+                                raise asyncio.CancelledError
+                            response = await session.execute(prompt_text)
 
                         # Get hooks early for observability around render + prompt:complete + store
                         hooks = session.coordinator.get("hooks")
@@ -3856,7 +3833,6 @@ def sigint_handler(signum, frame):
                         pass
 
         finally:
-            signal.signal(signal.SIGINT, original_handler)
             # Don't reset cancellation here - session.py handles status
             # Unregister this turn's badge hook so callbacks bound to this
             # finished per-turn manager don't accumulate on the shared hooks
@@ -3923,7 +3899,9 @@ def sigint_handler(signum, frame):
                 # freeze risk applies to any background Rich writes that
                 # land while the user is composing input.
                 with patch_stdout():
-                    user_input = await prompt_session.prompt_async()
+                    user_input = await prompt_session.prompt_async(
+                        set_exception_handler=False
+                    )
 
                 if user_input.lower() in ["exit", "quit"]:
                     break
@@ -3943,6 +3921,85 @@ def sigint_handler(signum, frame):
                         # see the note at the initial_prompt call site above.
                         await _execute_with_interrupt(_expanded_text)
 
+                    elif action == "deep_plan":
+                        task = data.get("args", "").strip()
+                        if not task:
+                            console.print("[cyan]Usage: /deep-plan <task>[/cyan]")
+                            continue
+
+                        from .deep_plan import (
+                            DeepPlanError,
+                            execute_deep_plan_turn,
+                        )
+                        from .lib.settings import AppSettings
+                        from .project_utils import get_project_slug
+                        from .ui import render_message
+
+                        def _display_deep_plan(deep_plan) -> None:
+                            console.print(
+                                f"\n[bold cyan]Deep plan[/bold cyan] "
+                                f"[dim]({escape_markup(deep_plan.attribution)})[/dim]"
+                            )
+                            render_message(
+                                {"role": "assistant", "content": deep_plan.plan},
+                                console,
+                                show_label=False,
+                            )
+
+                        try:
+
+                            async def _run_deep_plan_turn(task_snapshot: str) -> None:
+                                settings = AppSettings().with_session(
+                                    session.session_id, get_project_slug()
+                                )
+                                deep_plan_config = settings.get_deep_plan_config()
+                                expanded_task = await process_runtime_mentions(
+                                    session, task_snapshot
+                                )
+                                console.print(
+                                    "\n[dim]Planning with "
+                                    f"{escape_markup(deep_plan_config.description)}...[/dim]"
+                                )
+
+                                async def _execute_planned_parent(prompt: str) -> bool:
+                                    console.print(
+                                        "\n[dim]Executing with normal session "
+                                        "routing...[/dim]"
+                                    )
+                                    return await _execute_with_interrupt(
+                                        prompt, manage_interrupt=False
+                                    )
+
+                                await execute_deep_plan_turn(
+                                    session,
+                                    expanded_task,
+                                    deep_plan_config,
+                                    planner_runner=lambda awaitable: awaitable,
+                                    parent_executor=_execute_planned_parent,
+                                    on_plan=_display_deep_plan,
+                                )
+
+                            await run_with_interrupt(
+                                _run_deep_plan_turn(task),
+                                cancellation=session.coordinator.cancellation,
+                                console=console,
+                            )
+                        except asyncio.CancelledError:
+                            console.print(
+                                "[yellow]Deep planning cancelled; normal execution was stopped.[/yellow]"
+                            )
+                            continue
+                        except DeepPlanError as error:
+                            console.print(f"[red]{escape_markup(str(error))}[/red]")
+                            continue
+                        except Exception as error:
+                            logger.exception("Deep planning failed")
+                            console.print(
+                                f"[red]Deep-plan command failed: "
+                                f"{escape_markup(str(error))}[/red]"
+                            )
+                            continue
+
                     else:
                         if action == "load_skill":
                             # Call _load_skill() directly to get is_prompt flag —