diff --git a/navi/core/agent.py b/navi/core/agent.py
index 70ca720..b51ca07 100644
--- a/navi/core/agent.py
+++ b/navi/core/agent.py
@@ -18,7 +18,6 @@
 import asyncio
 import base64
 import io
-import re
 import time
 from datetime import datetime, timezone
 from pathlib import Path
@@ -41,7 +40,7 @@
     Tool,
     ToolContext,
     current_event_sink,
-    current_replan_runner,
+    current_plan_runner,
     current_stop_event,
     current_user_role,
     current_user_info,
@@ -52,7 +51,7 @@
 from .compressor import ContextCompressor, should_compress
 from .context_builder import ContextBuilder
 from .planning import PlanningEngine
-from navi.tools.replan import ReplanRunner
+from navi.tools.plan import PlanRunner
 from .stream_guard import _iter_stream_guarded
 from .subagent_runner import SubAgentRunner
 from .tool_utils import build_tool_list
@@ -61,7 +60,6 @@
     AIHelperTokensUsed,
     CompressionStarted,
     ModelInfo,
-    PlanningDebugData,
     StreamEnd,
     StreamStopped,
     SubagentComplete,
@@ -89,166 +87,6 @@
 _TOOL_DONE = object()
 
 
-_CASUAL_WORDS = frozenset(
-    {
-        # Russian greetings/social
-        "привет",
-        "здравствуй",
-        "здравствуйте",
-        "хай",
-        "хелло",
-        "хеллоу",
-        "как",
-        "дела",
-        "делишки",
-        "ты",
-        "вы",
-        "поживаешь",
-        "поживаете",
-        "жизнь",
-        "сам",
-        "сама",
-        "спасибо",
-        "спс",
-        "пока",
-        "bye",
-        "goodbye",
-        "доброе",
-        "утро",
-        "добрый",
-        "день",
-        "вечер",
-        "спокойной",
-        "ночи",
-        "ок",
-        "окей",
-        "ладно",
-        "давай",
-        # English greetings/social
-        "hi",
-        "hello",
-        "hey",
-        "hola",
-        "bonjour",
-        "how",
-        "are",
-        "you",
-        "it",
-        "going",
-        "things",
-        "what",
-        "up",
-        "s",
-        "thanks",
-        "thank",
-        "thx",
-        "good",
-        "morning",
-        "afternoon",
-        "evening",
-        "night",
-        "see",
-        "cya",
-        "ok",
-        "okay",
-        # Common fillers that keep a social phrase social
-        "a",
-        "an",
-        "the",
-        "and",
-        "too",
-        "very",
-        "much",
-        "today",
-        "now",
-        "there",
-        "here",
-        "again",
-        "well",
-        "oh",
-        "ah",
-        "um",
-        "в",
-        "и",
-        "а",
-        "но",
-        "же",
-        "тоже",
-        "очень",
-        "сегодня",
-        "сейчас",
-        "ну",
-        "вот",
-        "тут",
-        "ещё",
-        "раз",
-        "ка",
-        "is",
-        "am",
-        "are",
-        "do",
-        "does",
-        "did",
-        "be",
-        "been",
-        "being",
-        "man",
-        "dude",
-        "bro",
-        "mate",
-        "friend",
-        "dear",
-    }
-)
-
-
-def _is_casual_message(text: str) -> bool:
-    """Fast heuristic: obvious social/greeting chat that doesn't need planning.
-
-    Conservative by design. Returns True only for short, tool-free social
-    utterances (e.g. 'привет', 'как дела', 'спасибо'). Anything that looks
-    like a command, URL, path, or multi-step request is treated as non-casual.
-    """
-    if not text:
-        return True
-    # Tool/action markers and URLs disqualify the message immediately.
-    if any(marker in text for marker in ("@", "!", "http://", "https://", "file://")):
-        return False
-    # Path-like or command-like fragments.
-    if any(
-        fragment in text
-        for fragment in (
-            "/home",
-            "/tmp",
-            "/etc",
-            "./",
-            "../",
-            "~/",
-            "\\",
-            ".py ",
-            ".txt",
-            ".md",
-            ".json",
-        )
-    ):
-        return False
-    # A bare leading slash is a command/path indicator.
-    if text.strip().startswith("/"):
-        return False
-    # Long messages are very unlikely to be pure social greetings.
-    if len(text) > 100:
-        return False
-    stripped = text.strip()
-    # Very short messages are treated as casual by default.
-    if len(stripped) <= 10:
-        return True
-    words = [w for w in re.findall(r"\b\w+\b", stripped.lower()) if len(w) > 1]
-    if not words:
-        return True
-    casual_count = sum(1 for w in words if w in _CASUAL_WORDS)
-    return casual_count / len(words) >= 0.5
-
-
 async def _todo_progress_message(
     session_id: str, *, first_iteration: bool = False
 ) -> "Message | None":
@@ -266,11 +104,12 @@
 _FINAL_INTERCEPT_NUDGES = (
     "[Final-turn check] You ended the turn with text only, but the todo list still has open steps "
     "(pending/in_progress). Do not stop with bare text while work remains. Either mark finished "
-    "steps done/skipped via the todo tool, reflect on what is blocking, replan if the approach is "
-    "dead, or call a tool to continue — then keep going.",
+    "steps done/skipped via the todo tool, reflect on what is blocking, re-plan with the `plan` "
+    "tool if the approach is dead, or call a tool to continue — then keep going.",
     "[Final-turn check — second stop] You already stopped once without acting and the todo still "
     "has open steps. You must act now: call a tool — todo (close steps as done/skipped/failed with "
-    "validation), reflect, replan, or continue execution. Do not output text without a tool call.",
+    "validation), reflect, `plan` (re-plan with a reason), or continue execution. Do not output "
+    "text without a tool call.",
 )
 
 
@@ -560,53 +399,6 @@
         # nudge from the intercept site to the next turn's context).
         pending_final_nudge: Message | None = None
 
-        # Planning phase — always runs on the first user message in a session;
-        # on subsequent messages uses the profile's planning_enabled flag.
-        # force_plan suppresses the DIRECT shortcut: first message is always forced,
-        # and planning_mandatory extends that to every subsequent message.
-        # Casual greetings are exempt from planning even on the first message.
-        _is_first_message = sum(1 for m in session.messages if m.role == "user") == 1
-        _is_casual = _is_casual_message(context_content) and not profile.planning_mandatory
-        _force_plan = (_is_first_message and not _is_casual) or profile.planning_mandatory
-        if (_is_first_message or profile.planning_enabled) and not _is_casual:
-            log.debug(
-                "agent.planning_enter",
-                session_id=session_id,
-                first_message=_is_first_message,
-                planning_enabled=profile.planning_enabled,
-                force_plan=_force_plan,
-            )
-            async for _ev in self._planning.run(
-                session.context,
-                profile,
-                llm,
-                mem,
-                tool_schemas,
-                messages=session.messages,
-                force_plan=_force_plan,
-            ):
-                if isinstance(_ev, AIHelperTokensUsed):
-                    turn_ctx.subagent_tokens += _ev.completion_tokens
-                elif isinstance(_ev, PlanningDebugData):
-                    session.planning_logs.append(_ev.log)
-                    # Cap to prevent unbounded DB growth on long sessions
-                    _MAX_PLANNING_LOGS = 20
-                    if len(session.planning_logs) > _MAX_PLANNING_LOGS:
-                        session.planning_logs = session.planning_logs[-_MAX_PLANNING_LOGS:]
-                else:
-                    yield _ev
-
-            # Persist planning output (plan_ctx_msg + prompt_msg appended in
-            # planning.py) immediately — otherwise it sits in memory until the
-            # first save inside the tool loop and is lost on a mid-turn crash /
-            # CancelledError (server restart).
-            await self._sessions.save(session)
-
-            # Planning auto-populates the todo (planning.set_tasks) when Phase 3
-            # produced steps. Push the fresh state to the UI so the side panel
-            # reflects the plan before the tool-calling loop starts.
-            yield await self._todo_updated(session)
-
         ctx_task = asyncio.create_task(self._ctx_builder._collect_context_injections(profile))
         mem_facts_task = asyncio.create_task(
             self._ctx_builder._memory_facts_msg(
@@ -630,7 +422,7 @@
             log.debug("agent.memory_facts_none", session_id=session_id)
 
         anti_stall = AntiStallMonitor(profile)
-        if profile.anti_stall_enabled or profile.adaptive_replan_enabled:
+        if profile.anti_stall_enabled:
             await anti_stall.init(session_id)
 
         # Tool-calling loop — uses stream_complete() for every turn so thinking
@@ -831,22 +623,22 @@
                 user_info=current_user_info.get(),
                 cwd=_cwd_var.get(),
             )
-            # Expose a per-iteration ReplanRunner so the `replan` tool can re-run
-            # the planner over the live session. Constructed inside the loop (not
+            # Expose a per-iteration PlanRunner so the `plan` tool can run the
+            # planner over the live session. Constructed inside the loop (not
             # once before) using this iteration's profile/llm/tool_schemas so a
-            # post-switch_profile replan uses the reloaded profile. reset on exit
-            # so the runner never leaks beyond the tool-execution window.
-            _replan_runner = ReplanRunner(
+            # post-switch_profile plan call uses the reloaded profile. reset on
+            # exit so the runner never leaks beyond the tool-execution window.
+            _plan_runner = PlanRunner(
                 self._planning, session, profile, llm, mem, tool_schemas
             )
-            _replan_token = current_replan_runner.set(_replan_runner)
+            _plan_token = current_plan_runner.set(_plan_runner)
             try:
                 async for _ev in self._execute_tools_with_sink(
                     turn_tool_calls, tools, turn_ctx, session, stop_event, tool_ctx
                 ):
                     yield _ev
             finally:
-                current_replan_runner.reset(_replan_token)
+                current_plan_runner.reset(_plan_token)
 
             # The todo tool may have been called this turn (set/update/clear).
             # Re-read and push the latest state to the UI's side panel.
@@ -861,7 +653,7 @@
                 yield StreamStopped()
                 return
 
-            # Update anti-stall counters and adaptive-replan state
+            # Update anti-stall counters
             await anti_stall.post_turn(session_id, turn_tool_calls)
 
             # 7. If switch_profile was called this iteration, reload profile + tools.
diff --git a/navi/core/anti_stall.py b/navi/core/anti_stall.py
index bd672c8..a5777f6 100644
--- a/navi/core/anti_stall.py
+++ b/navi/core/anti_stall.py
@@ -1,4 +1,4 @@
-"""Anti-stall and adaptive re-plan monitoring for the Agent loop."""
+"""Anti-stall monitoring for the Agent loop."""
 
 from __future__ import annotations
 
@@ -16,21 +16,16 @@
     - No todo progress: consecutive iterations without a todo status change.
     - Repeated tool calls: identical tool signatures across consecutive turns.
 
-    Also handles adaptive re-plan: when a todo step is newly marked failed,
-    a re-planning message is queued for injection on the next iteration.
-    A second adaptive signal — the "long step" nudge — fires when the current
-    step stays in_progress for ``adaptive_long_step_threshold`` iterations
-    without a todo status change (and the model is still issuing non-repeating
-    tool calls), asking it to split the step before the general anti-stall
-    warning.
+    The former adaptive re-plan nudges (queued messages on failed steps / long
+    steps) were retired together with the mandatory planning gate: the model
+    is trusted to call `plan` (or revise the todo) on its own when execution
+    shows the approach is wrong.
     """
 
     profile: object  # AgentProfile — avoid circular import
     stall_no_todo: int = 0
     stall_repeat_tools: int = 0
     prev_tool_sigs: frozenset = field(default_factory=frozenset)
-    known_failed: frozenset = field(default_factory=frozenset)
-    replan_msg: str | None = None
     _todo_snapshot: frozenset | None = field(default=None, repr=False)
 
     async def init(self, session_id: str) -> None:
@@ -40,13 +35,6 @@
 
     async def pre_turn(self, session_id: str, iteration: int) -> Message | None:
         """Return a system message to inject before the LLM call, or None."""
-        # Adaptive re-plan: inject queued message from previous iteration
-        if self.profile.adaptive_replan_enabled and self.replan_msg:
-            msg = self.replan_msg
-            self.replan_msg = None
-            return Message(role="system", content=msg)
-
-        # Anti-stall warning
         if self.profile.anti_stall_enabled and iteration > 0:
             stalled = (
                 self.stall_no_todo >= self.profile.anti_stall_threshold
@@ -71,79 +59,29 @@
         return None
 
     async def post_turn(self, session_id: str, tool_calls: list[ToolCallRequest]) -> None:
-        """Update stall counters and adaptive-replan state after tool execution."""
-        from navi.tools.todo import get_failed_steps, get_task_snapshot
+        """Update stall counters after tool execution."""
+        from navi.tools.todo import get_task_snapshot
 
-        # --- Anti-stall: todo progress signal ---
-        if self.profile.anti_stall_enabled:
-            before = self._todo_snapshot
-            current = await get_task_snapshot(session_id)
-            if before is not None:
-                if current != before:
-                    self.stall_no_todo = 0
-                else:
-                    self.stall_no_todo += 1
-            self._todo_snapshot = current
+        if not self.profile.anti_stall_enabled:
+            return
 
-            # Repeated tool call signal
-            cur_sigs = frozenset(
-                (tc.name, json.dumps(tc.arguments, sort_keys=True))
-                for tc in (tool_calls or [])
-            )
-            if cur_sigs and cur_sigs == self.prev_tool_sigs:
-                self.stall_repeat_tools += 1
+        # Todo progress signal
+        before = self._todo_snapshot
+        current = await get_task_snapshot(session_id)
+        if before is not None:
+            if current != before:
+                self.stall_no_todo = 0
             else:
-                self.stall_repeat_tools = 0
-            self.prev_tool_sigs = cur_sigs
+                self.stall_no_todo += 1
+        self._todo_snapshot = current
 
-        # --- Adaptive re-plan: detect newly-failed steps, or a long-running step ---
-        if self.profile.adaptive_replan_enabled:
-            current_failed = await get_failed_steps(session_id)
-            new_failures = current_failed - self.known_failed
-            self.known_failed = current_failed
-            if new_failures:
-                import structlog
-                log = structlog.get_logger()
-                failed_labels = ", ".join(
-                    f'step {idx} ("{text}")'
-                    for idx, text in sorted(new_failures)
-                )
-                self.replan_msg = (
-                    f"[Adaptive re-plan] {failed_labels} just failed. "
-                    "Before continuing, revise your plan with the todo tool: either replace the remaining "
-                    "pending steps or mark failed/skipped steps with validation. Then continue execution "
-                    "with an approach that accounts for what went wrong."
-                )
-                log.info(
-                    "agent.adaptive_replan_queued",
-                    failures=len(new_failures),
-                    session_id=session_id,
-                )
-            elif (
-                self.profile.adaptive_long_step_threshold
-                and tool_calls
-                and self.stall_repeat_tools == 0
-                and self.stall_no_todo == self.profile.adaptive_long_step_threshold
-            ):
-                # The current step has been in_progress for several iterations
-                # without a todo status change, yet the model is still working
-                # (non-repeating tool calls) — nudge it to split the step before
-                # the general anti-stall warning fires. ``stall_no_todo`` is only
-                # maintained while ``anti_stall_enabled`` is on, so this nudge is
-                # part of the anti-stall family. Fires once, at the threshold:
-                # re-newed todo progress resets the counter to 0, so the nudge
-                # rewards real movement rather than repeating every iteration.
-                self.replan_msg = (
-                    f"[Adaptive re-plan] the current step has been in_progress for "
-                    f"{self.stall_no_todo} tool iterations without completing. If it is larger than the "
-                    "plan anticipated, split it now: mark the finished part 'done' via todo (with "
-                    "validation) and add the remainder as new steps, then continue. If you are genuinely "
-                    "close to finishing, ignore this."
-                )
-                import structlog
-                log = structlog.get_logger()
-                log.info(
-                    "agent.adaptive_long_step_queued",
-                    iterations=self.stall_no_todo,
-                    session_id=session_id,
-                )
+        # Repeated tool call signal
+        cur_sigs = frozenset(
+            (tc.name, json.dumps(tc.arguments, sort_keys=True))
+            for tc in (tool_calls or [])
+        )
+        if cur_sigs and cur_sigs == self.prev_tool_sigs:
+            self.stall_repeat_tools += 1
+        else:
+            self.stall_repeat_tools = 0
+        self.prev_tool_sigs = cur_sigs
\ No newline at end of file
diff --git a/navi/core/planning.py b/navi/core/planning.py
index 8bb57a8..05d857d 100644
--- a/navi/core/planning.py
+++ b/navi/core/planning.py
@@ -1,6 +1,10 @@
 """Planning pipeline — extracted from agent.py.
 
-Async generator that runs 3-phase planning (analysis → review → execution plan).
+Async generator that runs 2-phase planning (analysis → execution plan).
+Top-level planning is agent-invoked via the `plan` tool; sub-agents run it
+automatically before their tool loop. The former Phase 2 (structured review)
+was retired — it fired in under 10% of production plans and rarely changed
+the outcome.
 """
 
 import asyncio
@@ -23,6 +27,50 @@
 
 log = structlog.get_logger()
 
+# Conversation char budget for the Phase 1 prompt (≈5k tokens). Mid-session
+# plan/re-plan calls arrive after potentially dozens of tool results; sending
+# the full transcript chokes cloud models (measured: 28.7k prompt tokens →
+# empty output). The window keeps the newest messages and pins the first user
+# message (the original task statement).
+_PHASE1_CONTEXT_MAX_CHARS = 20_000
+
+# gemma4 on ollama-cloud leaks its thinking-channel header into content on
+# non-streaming calls ("thought\n<channel|>"); strip the artifact so the
+# analysis/plan parse cleanly.
+_CHANNEL_ARTIFACT = re.compile(r"^\s*thought\s*<channel\|>\s*", re.IGNORECASE)
+
+
+def _strip_channel_artifact(text: str) -> str:
+    return _CHANNEL_ARTIFACT.sub("", text, count=1)
+
+
+def _window_for_planner(msgs: list[Message], max_chars: int) -> list[Message]:
+    """Newest-first char-budget window over the conversation.
+
+    Returns all messages when they fit the budget. Otherwise keeps the newest
+    messages (at least one, even if a single message exceeds the budget) and
+    pins the first user message at the front so a re-plan keeps the original
+    task statement.
+    """
+    if not msgs:
+        return []
+    total = sum(len(m.content or "") for m in msgs)
+    if total <= max_chars:
+        return list(msgs)
+    keep: list[Message] = []
+    used = 0
+    for m in reversed(msgs):
+        size = len(m.content or "")
+        if keep and used + size > max_chars:
+            break
+        keep.append(m)
+        used += size
+    keep.reverse()
+    first_user_idx = next((i for i, m in enumerate(msgs) if m.role == "user"), None)
+    if first_user_idx is not None and first_user_idx < len(msgs) - len(keep):
+        keep.insert(0, msgs[first_user_idx])
+    return keep
+
 
 def _parse_plan_steps(plan_text: str) -> list[tuple[str, str]]:
     """Extract numbered step lines from the **Steps:** section of a plan.
@@ -53,10 +101,14 @@
 
 
 class PlanningEngine:
-    """Runs the 3-phase planning pipeline."""
+    """Runs the 2-phase planning pipeline."""
 
     def __init__(self, ctx_builder) -> None:
         self._ctx_builder = ctx_builder
+        # COMPLEXITY parsed from the latest Phase 1 output ("simple"/"medium"/
+        # "complex"). The `plan` tool reads it to pick the confirmation
+        # instruction — complex tasks wait for the user, simpler ones proceed.
+        self.last_complexity: str = ""
 
     async def run(
         self,
@@ -68,16 +120,21 @@
         messages: list[Message] | None = None,
         system_prompt_override: str | None = None,
         is_subagent: bool = False,
-        force_plan: bool = False,
         is_replan: bool = False,
         replan_context: str | None = None,
     ) -> AsyncGenerator:
         """Planning pipeline (async generator):
 
-        Phase 1 — Analysis (think=profile.think_enabled): reformulate the task,
-                   identify subtasks and unknowns. Outputs DIRECT for simple requests.
-        Phase 2 — Structured review (conditional, think=False): one critique pass.
+        Phase 1 — Analysis (think=False): reformulate the task, identify
+                   subtasks, unknowns and COMPLEXITY. Sub-agents may output
+                   DIRECT to skip planning for trivial subtasks.
         Phase 3 — Execution plan (think=False): assigns each subtask to TOOL / AGENT / SELF.
+
+        Both calls are non-streaming helpers: think=False keeps reasoning
+        models from dumping their chain-of-thought into the structured output
+        (gemma4 leaks "thought<channel|>" headers, glm-flash returns empty
+        content with think=True). Consistent with compressor / ai_helper.
+        The Phase 1 conversation is windowed to _PHASE1_CONTEXT_MAX_CHARS.
         """
         # ── Build compact tool list for Phase 2 / Phase 3 ─────────────────────
         if tool_schemas:
@@ -103,11 +160,11 @@
         if _mcp_msg:
             _base_sys = _base_sys + "\n\n---\n\n" + (_mcp_msg.content or "")
 
-        # Re-plan framing: when the agent calls `replan` mid-task, ReplanRunner packs
-        # the reason + current todo + scratchpad findings/errors into replan_context.
-        # We frame it here so Phase 1 revises the existing plan instead of starting
-        # fresh, while reusing the same analysis → review → plan pipeline. The DIRECT
-        # shortcut and observe-skip are suppressed for re-plans (gated above/below on
+        # Re-plan framing: when the agent calls `plan` with a reason mid-task,
+        # PlanRunner packs the reason + current todo + scratchpad findings/errors
+        # into replan_context. We frame it here so Phase 1 revises the existing
+        # plan instead of starting fresh, while reusing the same analysis → plan
+        # pipeline. The DIRECT shortcut is suppressed for re-plans (gated below on
         # is_replan) so a stale plan always yields a new plan.
         _replan_block = ""
         _has_replan_ctx = bool(is_replan and replan_context)
@@ -142,7 +199,7 @@
                     )
                     + (
                         ""
-                        if (force_plan or is_replan) else
+                        if not (is_subagent and not is_replan) else
                         "CRITICAL DIRECT shortcut — use it whenever possible:\n"
                         "- If the user's message is a greeting (hello, hi, thanks, good morning, "
                         "'how are you', 'what\\'s up', 'привет', 'как дела', 'спасибо') or any "
@@ -167,11 +224,6 @@
                     "Analyse the request and output:\n\n"
                     "TASK: [one clear sentence — what actually needs to be done]\n"
                     "GOAL: [how you will know the task is complete]\n"
-                    "MODE: observe | act — observe = the user wants to look at / read / "
-                    "explain / inspect / list / find / show something (no changes to be made); "
-                    "act = the user wants to build / change / fix / create / run / modify "
-                    "something. Classify from the user's actual intent, not from the tools you "
-                    "happen to use — reading files to answer is still observe.\n"
                     "UNKNOWNS: [genuine uncertainties that could block execution, or NONE]\n"
                     "RESOURCES:\n"
                     "- [tool_name]: [what it does] — [limitation if any] — [alternative if limitation blocks the goal]\n"
@@ -192,9 +244,6 @@
                     "ATOMICITY: For each subtask that requires multiple actions — if it fails halfway, "
                     "is any partial result still useful? If not, split it into smaller steps where "
                     "each one delivers an independent, usable result on its own.\n"
-                    "REFLECT: yes — if the task is complex (multiple unknowns, external APIs, "
-                    "research required, or high-stakes/irreversible actions); "
-                    "no — if it is straightforward and the path is clear.\n"
                     "COMMITMENTS: [follow the plan step by step using the todo tool; gather missing context independently before asking the user; before the final answer, run a knowledge persistence checkpoint]\n\n"
                     "Rules: list enough subtasks to make execution unambiguous. "
                     "Simple tasks usually need 1-3 subtasks; medium tasks 5-9; complex or autonomous tasks 8-15. "
@@ -205,14 +254,19 @@
             phase1_ctx: list[Message] = [phase1_system]
             if mem:
                 phase1_ctx.append(mem)
-            phase1_ctx.extend(m for m in context if m.role != "system")
+            phase1_ctx.extend(
+                _window_for_planner(
+                    [m for m in context if m.role != "system"],
+                    _PHASE1_CONTEXT_MAX_CHARS,
+                )
+            )
 
             try:
                 r1 = await asyncio.wait_for(
-                    llm.complete(phase1_ctx, tools=None, temperature=0.3, model=profile.model, think=profile.think_enabled),
+                    llm.complete(phase1_ctx, tools=None, temperature=0.3, model=profile.model, think=False),
                     timeout=settings.llm_complete_timeout,
                 )
-                analysis = (r1.content or "").strip()
+                analysis = _strip_channel_artifact((r1.content or "").strip())
             except asyncio.TimeoutError:
                 log.warning("agent.planning_phase1_timeout", timeout=settings.llm_complete_timeout)
                 _dbg["result"] = "phase1_timeout"
@@ -249,88 +303,14 @@
                 log.debug("agent.planning_stopped", phase=1)
                 return
 
-            # Observe vs act: an observe request (look/read/explain/inspect) needs
-            # no multi-step execution plan — skip Phase 2/3, no auto-todo, no
-            # "execute step by step" prompt. The agent just gathers info and answers.
-            # Independent of force_plan (force_plan only suppresses the DIRECT
-            # shortcut); "look at X" on the first message should still not plan.
-            _is_observe = bool(re.search(r"MODE\s*:\s*observe", analysis, re.IGNORECASE))
-            if profile.observe_skips_plan_enabled and _is_observe and not is_subagent and not is_replan:
-                log.debug("agent.planning_observe_skip")
-                _dbg["result"] = "observe_skip"
-                if not is_subagent:
-                    yield PlanningDebugData(log=_dbg)
-                return
+            # Expose the COMPLEXITY classification to the caller (the `plan` tool
+            # picks the confirmation instruction from it — complex tasks wait for
+            # the user's go-ahead, simpler ones proceed straight to execution).
+            _cx = re.search(r"COMPLEXITY\s*:\s*(simple|medium|complex)", analysis, re.IGNORECASE)
+            self.last_complexity = _cx.group(1).lower() if _cx else ""
         else:
             log.debug("agent.planning_phase1_skipped")
 
-        # ── Phase 2: Structured review (conditional) ───────────────────────────
-        advisor_feedback: str = ""
-        needs_reflect = bool(re.search(r"REFLECT\s*:\s*yes", analysis, re.IGNORECASE))
-
-        if profile.planning_phase2_enabled and needs_reflect and not is_subagent:
-            yield PlanningStatus(phase=2, label="Reviewing plan...", is_subagent=is_subagent)
-
-            review_system = Message(
-                role="system",
-                content=(
-                    _base_sys
-                    + "\n\n---\n\n"
-                    "[PLANNING - PHASE 2: STRUCTURED REVIEW]\n\n"
-                    "Review the phase 1 task analysis before execution. "
-                    "Do not change the user's goal. Do not invent facts. "
-                    "Prefer resolving missing information through connected knowledge servers, docs, manuals, memory, files, "
-                    "tool schemas, command output, or web research before asking the user. "
-                    "Check that the proposed knowledge source and persistence target match the fact scope: "
-                    "memory only for personal user facts; connected MCP knowledge servers for their own canonical domains; "
-                    "docs/manuals for project-wide documentation. Flag any plan that stores infrastructure facts in memory.\n\n"
-                    "Return exactly these sections:\n\n"
-                    "## Critic\n"
-                    "- 3-5 bullets: wrong or unverified assumptions, ignored risks, contradictions, "
-                    "and facts that must be verified before acting.\n\n"
-                    "## Pragmatist\n"
-                    "- 3-5 bullets: simpler path, unnecessary steps, mergeable steps, better executor choices, "
-                    "and cheaper ways to reach the user's actual goal.\n\n"
-                    "## Detailer\n"
-                    "- 3-5 bullets: missing requirements, missing docs/files/tools to inspect, edge cases, "
-                    "and validation steps.\n\n"
-                    "## Plan Adjustments\n"
-                    "- Concrete changes Phase 3 must apply: add/remove/split/merge/reorder steps, "
-                    "change TOOL/AGENT/SELF executor, verify specific facts, correct the persistence target, add a knowledge persistence checkpoint, or defer user questions "
-                    "until available sources are checked.\n\n"
-                    "Keep output concise. No prose outside these sections.\n\n"
-                    f"PHASE 1 ANALYSIS:\n{analysis}"
-                ),
-            )
-            review_ctx: list[Message] = [review_system]
-            if mem:
-                review_ctx.append(mem)
-            review_ctx.extend(m for m in context if m.role != "system")
-            try:
-                r_review = await asyncio.wait_for(
-                    llm.complete(review_ctx, tools=None, temperature=0.35, model=profile.model, think=False),
-                    timeout=settings.llm_complete_timeout,
-                )
-                advisor_feedback = (r_review.content or "").strip()
-                if r_review.prompt_tokens or r_review.completion_tokens:
-                    yield AIHelperTokensUsed(
-                        prompt_tokens=r_review.prompt_tokens or 0,
-                        completion_tokens=r_review.completion_tokens or 0,
-                    )
-                _dbg["phases"]["2"] = {
-                    "output": advisor_feedback,
-                    "prompt_tokens": r_review.prompt_tokens or 0,
-                    "completion_tokens": r_review.completion_tokens or 0,
-                }
-                log.debug("agent.planning_review_done", has_output=bool(advisor_feedback))
-            except Exception:
-                log.warning("agent.planning_review_failed", exc_info=True)
-                _dbg["phases"]["2"] = {"output": "", "prompt_tokens": 0, "completion_tokens": 0}
-
-            if _stop and _stop.is_set():
-                log.debug("agent.planning_stopped", phase=2)
-                return
-
         # ── Phase 3: Execution plan ────────────────────────────────────────────
         if not profile.planning_phase3_enabled:
             log.debug("agent.planning_phase3_skipped")
@@ -341,12 +321,6 @@
 
         yield PlanningStatus(phase=3, label="Building execution plan...", is_subagent=is_subagent)
 
-        advisor_block = (
-            "Structured review feedback — apply the Plan Adjustments in your plan:\n\n"
-            + advisor_feedback
-            + "\n\n---\n\n"
-        ) if advisor_feedback else ""
-
         phase3_system = Message(
             role="system",
             content=(
@@ -356,7 +330,6 @@
                 "Task analysis:\n\n"
                 f"{analysis}\n\n"
                 "---\n\n"
-                + advisor_block
                 + available_tools_block
                 + "Now write the execution plan. For each subtask assign a specific executor:\n"
                 "- TOOL: <tool_name>  — a single tool call is enough; use exact tool names from the list above\n"
@@ -443,7 +416,7 @@
                 llm.complete(phase3_ctx, tools=None, temperature=0.3, model=profile.model, think=False),
                 timeout=settings.llm_complete_timeout,
             )
-            plan_text = (r2.content or "").strip()
+            plan_text = _strip_channel_artifact((r2.content or "").strip())
         except asyncio.TimeoutError:
             log.warning("agent.planning_phase3_timeout", timeout=settings.llm_complete_timeout)
             _dbg["result"] = "phase3_timeout"
@@ -491,14 +464,19 @@
             messages.append(plan_ctx_msg)
             messages.append(Message(role="assistant", content=plan_text, is_plan=True, is_context=False))
 
-        prompt_msg = Message(
-            role="user",
-            content="Plan is ready. Execute it now step by step, starting with step 1. Use the todo tool to track progress.",
-            is_display=False,
-        )
-        context.append(prompt_msg)
-        if messages is not None:
-            messages.append(prompt_msg)
+        # Sub-agents plan and execute without confirmation — inject the go-ahead.
+        # The top-level path is the `plan` tool: IT composes the follow-up
+        # instruction (confirmation for complex tasks, go-ahead otherwise) in
+        # its tool result, so no prompt is injected here.
+        if is_subagent:
+            prompt_msg = Message(
+                role="user",
+                content="Plan is ready. Execute it now step by step, starting with step 1. Use the todo tool to track progress.",
+                is_display=False,
+            )
+            context.append(prompt_msg)
+            if messages is not None:
+                messages.append(prompt_msg)
 
         _todo_steps = _parse_plan_steps(plan_text)
         if _todo_steps:
@@ -516,7 +494,7 @@
             except Exception:
                 log.warning("agent.todo_auto_populate_failed", exc_info=True)
 
-        log.debug("agent.plan_ready", phases=3 if advisor_feedback else 2, length=len(plan_text))
+        log.debug("agent.plan_ready", complexity=self.last_complexity or None, length=len(plan_text))
         if not is_subagent:
             yield PlanningDebugData(log=_dbg)
         yield PlanReady(plan=plan_text, is_subagent=is_subagent)
diff --git a/navi/core/registry.py b/navi/core/registry.py
index 28bf849..c54edce 100644
--- a/navi/core/registry.py
+++ b/navi/core/registry.py
@@ -17,7 +17,7 @@
     ManageRecallTool,
     MemoryTool,
     ReflectTool,
-    ReplanTool,
+    PlanTool,
     ScheduleRecallTool,
     SpawnAgentTool,
     SshExecTool,
@@ -240,7 +240,7 @@
                 ShareFileTool(), ContentPublishTool(),
                 TodoTool(kv_store=kv_store), ScratchpadTool(kv_store=kv_store),
                 ReflectTool(ai_helper=ai_helper),
-                ReplanTool(),
+                PlanTool(),
                 reload_tool, list_tool, manual_tool,
                 mcp_status_tool, create_mcp_server_tool, test_mcp_tool_tool,
                 schedule_recall_tool, manage_recall_tool,
diff --git a/navi/tools/__init__.py b/navi/tools/__init__.py
index 268a66f..0251975 100644
--- a/navi/tools/__init__.py
+++ b/navi/tools/__init__.py
@@ -15,7 +15,7 @@
 from .switch_profile import SwitchProfileTool
 from .list_profiles import ListProfilesTool
 from .reflect import ReflectTool
-from .replan import ReplanRunner, ReplanTool
+from .plan import PlanRunner, PlanTool
 
 __all__ = [
     "Tool",
@@ -36,6 +36,6 @@
     "SwitchProfileTool",
     "ListProfilesTool",
     "ReflectTool",
-    "ReplanRunner",
-    "ReplanTool",
+    "PlanRunner",
+    "PlanTool",
 ]
diff --git a/navi/tools/_internal/base.py b/navi/tools/_internal/base.py
index be83b30..a70ac9d 100644
--- a/navi/tools/_internal/base.py
+++ b/navi/tools/_internal/base.py
@@ -60,12 +60,13 @@
     "current_working_directory", default=None
 )
 
-# Set by run_stream() inside the tool-execution window. Holds the per-run ReplanRunner
-# bound to the current session/profile/llm, so the `replan` tool can re-run the planner
-# over the live session context + todo + scratchpad. None outside an agent run (e.g.
-# sub-agent runs, which don't expose replan) — the tool reports unavailable then.
-current_replan_runner: ContextVar[object | None] = ContextVar(
-    "current_replan_runner", default=None
+# Set by run_stream() inside the tool-execution window. Holds the per-run PlanRunner
+# bound to the current session/profile/llm, so the `plan` tool can run the planner
+# over the live session context (+ todo + scratchpad in re-plan mode). None outside
+# an agent run (e.g. sub-agent runs, which don't expose plan) — the tool reports
+# unavailable then.
+current_plan_runner: ContextVar[object | None] = ContextVar(
+    "current_plan_runner", default=None
 )
 
 
diff --git a/navi/tools/plan.py b/navi/tools/plan.py
new file mode 100644
index 0000000..6d52994
--- /dev/null
+++ b/navi/tools/plan.py
@@ -0,0 +1,229 @@
+"""
+plan — agent-invoked planning: think through a task before executing it.
+
+The planner runs as a tool, not as a mandatory pre-turn gate: the agent calls
+`plan` itself when a task is non-trivial. It produces a structured execution
+plan (milestones + steps with TOOL/AGENT/SELF executors) and auto-populates
+the todo.
+
+Fresh planning: call with no arguments before starting execution. Revision:
+pass `reason` (what you discovered that invalidates the remaining plan) and
+optionally `updated_goal` — the planner then frames the run as a re-plan and
+replaces the todo, preserving completed work.
+
+PlanningStatus / PlanReady events are forwarded to the event sink so the UI
+shows planning progress and the plan card mid-turn (pattern: switch_profile).
+The tool result carries the follow-up instruction: for COMPLEXITY=complex the
+agent must present the plan and wait for the user's confirmation; otherwise it
+proceeds straight to execution.
+"""
+
+from __future__ import annotations
+
+import structlog
+
+from navi.tools._internal.base import (
+    Tool,
+    ToolContext,
+    ToolResult,
+    current_event_sink,
+    current_plan_runner,
+)
+from navi.tools.scratchpad import get_section as scratchpad_get_section
+from navi.tools.todo import render_todo_lines
+
+log = structlog.get_logger()
+
+# Mirror agent.py's cap so planning debug logs don't grow session.planning_logs unbounded.
+_MAX_PLANNING_LOGS = 20
+
+
+class PlanRunner:
+    """Per-run helper bound to the live session/profile/llm.
+
+    Constructed inside the agent run_stream loop (so it sees the current
+    iteration's profile/llm/tool_schemas — correct after switch_profile) and
+    exposed to the `plan` tool via the `current_plan_runner` ContextVar.
+    """
+
+    def __init__(self, planning_engine, session, profile, llm, mem, tool_schemas) -> None:
+        self.planning = planning_engine
+        self._session = session
+        self._profile = profile
+        self._llm = llm
+        self._mem = mem
+        self._tool_schemas = tool_schemas
+
+    @property
+    def complexity(self) -> str:
+        """COMPLEXITY parsed from the latest Phase 1 output ("" when unknown)."""
+        return self.planning.last_complexity
+
+    async def plan(
+        self, reason: str | None, updated_goal: str | None, event_sink
+    ) -> str | None:
+        """Run the planner over the live session context.
+
+        With `reason` — revision mode: the reason + current todo + scratchpad
+        findings/errors are packed into the re-plan context, Phase 1 frames the
+        run as a revision of the existing plan and the todo is replaced.
+        Without — fresh planning.
+
+        Returns the new plan text, or None if planning produced no plan.
+        """
+        session = self._session
+
+        if reason:
+            todo_lines = await render_todo_lines(session.id)
+            findings = await scratchpad_get_section(session.id, "findings", user_id=session.user_id)
+            errors = await scratchpad_get_section(session.id, "errors", user_id=session.user_id)
+
+            todo_block = "\n".join(todo_lines) if todo_lines else "(empty — no todo steps)"
+            findings_block = findings.strip() if findings and findings.strip() else "(none)"
+            errors_block = errors.strip() if errors and errors.strip() else "(none)"
+            goal_line = (
+                updated_goal.strip()
+                if updated_goal and updated_goal.strip()
+                else "(not specified — keep the original goal)"
+            )
+
+            replan_context = (
+                f"Reason for re-plan: {reason}\n"
+                f"Updated goal: {goal_line}\n\n"
+                f"Current todo:\n{todo_block}\n\n"
+                f"Scratchpad — findings:\n{findings_block}\n\n"
+                f"Scratchpad — errors:\n{errors_block}"
+            )
+        else:
+            replan_context = None
+
+        plan_text: str | None = None
+        try:
+            # Lazy import: navi.core.events pulls in navi.core.__init__ → registry →
+            # navi.tools, which would cycle if done at module load. At runtime every
+            # module is already imported, so this is safe.
+            from navi.core.events import (
+                AIHelperTokensUsed,
+                PlanReady,
+                PlanningDebugData,
+                PlanningStatus,
+            )
+
+            async for ev in self.planning.run(
+                session.context,
+                self._profile,
+                self._llm,
+                self._mem,
+                self._tool_schemas,
+                messages=session.messages,
+                is_replan=bool(reason),
+                replan_context=replan_context,
+            ):
+                if isinstance(ev, PlanReady):
+                    plan_text = ev.plan
+                    if event_sink is not None:
+                        await event_sink.put(ev)
+                elif isinstance(ev, PlanningStatus):
+                    if event_sink is not None:
+                        await event_sink.put(ev)
+                elif isinstance(ev, PlanningDebugData):
+                    session.planning_logs.append(ev.log)
+                    if len(session.planning_logs) > _MAX_PLANNING_LOGS:
+                        session.planning_logs = session.planning_logs[-_MAX_PLANNING_LOGS:]
+                # AIHelperTokensUsed is internal (wire-None) — no UI consumer
+                # in the tool path; token usage stays visible via planning_logs.
+        except Exception:
+            log.warning("plan.runner_failed", exc_info=True)
+            return None
+
+        return plan_text
+
+
+class PlanTool(Tool):
+    name = "plan"
+    description = (
+        "Run the planner: decompose a task into a structured execution plan (milestones + steps "
+        "with TOOL/AGENT/SELF executors) and auto-populate the todo.\n\n"
+        "Call this BEFORE starting execution when:\n"
+        "- The task is non-trivial: multiple steps, several files/systems, research, or real risk.\n"
+        "- The work needs decomposition, ordering, or sub-agent scoping decided up front.\n\n"
+        "Skip it for trivial work: single-file edits, one-off commands, questions, casual chat.\n\n"
+        "Re-plan mid-task by passing `reason` — what you discovered that invalidates the remaining "
+        "plan (a step turned out unnecessary, the real problem differs from the assumed one, new "
+        "constraints appeared) — plus optionally `updated_goal`. The new plan replaces the todo; "
+        "completed work is preserved in the scratchpad/conversation.\n\n"
+        "Costs 2 LLM calls (analysis + execution plan) — use selectively, like `reflect`.\n\n"
+        "Do NOT call plan for a single failed step: revise the todo inline instead. For a small "
+        "adjustment (drop/merge/reorder 1-2 steps) edit the `todo` directly — plan is for when the "
+        "remaining plan's overall structure is wrong (or missing). Use `reflect` when you're unsure "
+        "what's wrong (it surfaces assumptions, no plan change); use `plan` when you need a plan."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "reason": {
+                "type": "string",
+                "description": (
+                    "Re-plan mode: what changed — the discovery that makes the current plan stale. "
+                    "One or two sentences, concrete (e.g. 'the config is TOML not JSON, so the parser "
+                    "step is wrong'). Omit for fresh planning."
+                ),
+            },
+            "updated_goal": {
+                "type": "string",
+                "description": (
+                    "Optional, only with `reason`: the new success criterion if the goal itself "
+                    "shifted. Omit to keep the original goal."
+                ),
+            },
+        },
+        "required": [],
+    }
+
+    async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult:
+        reason = (params.get("reason") or "").strip() or None
+        updated_goal = (params.get("updated_goal") or "").strip() or None
+
+        runner = current_plan_runner.get(None)
+        if runner is None:
+            return ToolResult(
+                success=False,
+                output="",
+                error="plan is not available in this context (no active agent run)",
+            )
+
+        # Forward planning UI events mid-turn (status line + plan card) —
+        # the agent loop drains the sink into the WS stream (switch_profile pattern).
+        # tool_ctx.event_sink is None in the agent loop (the ContextVar is the real
+        # channel), so fall through to current_event_sink whenever ctx has no sink.
+        sink = ctx.event_sink if (ctx and ctx.event_sink) else current_event_sink.get()
+
+        plan_text = await runner.plan(reason, updated_goal, sink)
+        if not plan_text:
+            if reason:
+                return ToolResult(
+                    success=False,
+                    output="",
+                    error="re-planning produced no plan; keep the current plan and continue, or revise the todo inline",
+                )
+            return ToolResult(
+                success=False,
+                output="",
+                error="planning produced no plan; proceed directly without a plan",
+            )
+
+        title = "# Revised plan" if reason else "# Plan"
+        if runner.complexity == "complex":
+            tail = (
+                "The task is COMPLEX. Briefly present this plan to the user — milestones and key "
+                "steps in a few sentences — and WAIT for their confirmation before executing. "
+                "Do not start executing yet."
+            )
+        else:
+            tail = (
+                "The plan is ready and the todo has been populated with its steps. "
+                "Proceed with execution now, starting from step 1, tracking progress with the todo "
+                "tool. You may summarize the plan in one or two sentences before starting."
+            )
+
+        return ToolResult(success=True, output=f"{title}\n\n{plan_text}\n\n---\n{tail}")
\ No newline at end of file
diff --git a/navi/tools/replan.py b/navi/tools/replan.py
deleted file mode 100644
index 7f5c8d5..0000000
--- a/navi/tools/replan.py
+++ /dev/null
@@ -1,179 +0,0 @@
-"""
-replan — re-run the planner mid-task when the current plan is stale.
-
-Use when discoveries during execution invalidate the original plan (a step turned
-out to be unnecessary, the real problem is different from the assumed one, new
-constraints appeared). This is NOT for failed steps — those are handled by the
-[Adaptive re-plan] signal, which nudges you to revise the todo inline without
-re-running the planner.
-
-Integrated approach: replan reuses the same PlanningEngine pipeline (analysis →
-review → execution plan) with `is_replan=True`, which frames Phase 1 as a revision
-of the existing plan and suppresses the DIRECT shortcut and observe-skip so a
-new plan is always produced. The current todo + scratchpad findings/errors are
-packed into the re-plan context so the new plan accounts for real progress. The
-new plan replaces the todo (set_tasks) — completed work is preserved in the
-scratchpad/conversation, but the remaining steps are overwritten.
-
-Costs 1–3 LLM calls (Phase 1 always; Phase 2 only if REFLECT: yes; Phase 3
-always). Use selectively, like `reflect`.
-"""
-
-from __future__ import annotations
-
-import structlog
-
-from navi.tools._internal.base import Tool, ToolContext, ToolResult, current_replan_runner
-from navi.tools.scratchpad import get_section as scratchpad_get_section
-from navi.tools.todo import render_todo_lines
-
-log = structlog.get_logger()
-
-# Mirror agent.py's cap so re-plan debug logs don't grow session.planning_logs unbounded.
-_MAX_PLANNING_LOGS = 20
-
-
-class ReplanRunner:
-    """Per-run helper bound to the live session/profile/llm.
-
-    Constructed inside the agent run_stream loop (so it sees the current
-    iteration's profile/llm/tool_schemas — correct after switch_profile) and
-    exposed to the `replan` tool via the `current_replan_runner` ContextVar.
-    """
-
-    def __init__(self, planning_engine, session, profile, llm, mem, tool_schemas) -> None:
-        self._planning = planning_engine
-        self._session = session
-        self._profile = profile
-        self._llm = llm
-        self._mem = mem
-        self._tool_schemas = tool_schemas
-
-    async def replan(self, reason: str, updated_goal: str | None) -> str | None:
-        """Re-run the planner over the live session context + todo + scratchpad.
-
-        Returns the new plan text, or None if planning produced no plan. The
-        planner mutates session.context/messages (appends the new plan + execute
-        prompt) and replaces the todo via set_tasks.
-        """
-        session = self._session
-        todo_lines = await render_todo_lines(session.id)
-        findings = await scratchpad_get_section(session.id, "findings", user_id=session.user_id)
-        errors = await scratchpad_get_section(session.id, "errors", user_id=session.user_id)
-
-        todo_block = "\n".join(todo_lines) if todo_lines else "(empty — no todo steps)"
-        findings_block = findings.strip() if findings and findings.strip() else "(none)"
-        errors_block = errors.strip() if errors and errors.strip() else "(none)"
-        goal_line = updated_goal.strip() if updated_goal and updated_goal.strip() else "(not specified — keep the original goal)"
-
-        replan_context = (
-            f"Reason for re-plan: {reason}\n"
-            f"Updated goal: {goal_line}\n\n"
-            f"Current todo:\n{todo_block}\n\n"
-            f"Scratchpad — findings:\n{findings_block}\n\n"
-            f"Scratchpad — errors:\n{errors_block}"
-        )
-
-        plan_text: str | None = None
-        try:
-            # Lazy import: navi.core.events pulls in navi.core.__init__ → registry →
-            # navi.tools, which would cycle if done at module load. At runtime every
-            # module is already imported, so this is safe.
-            from navi.core.events import (
-                AIHelperTokensUsed,
-                PlanReady,
-                PlanningDebugData,
-                PlanningStatus,
-            )
-
-            async for ev in self._planning.run(
-                session.context,
-                self._profile,
-                self._llm,
-                self._mem,
-                self._tool_schemas,
-                messages=session.messages,
-                force_plan=True,
-                is_replan=True,
-                replan_context=replan_context,
-            ):
-                if isinstance(ev, PlanReady):
-                    plan_text = ev.plan
-                elif isinstance(ev, PlanningDebugData):
-                    session.planning_logs.append(ev.log)
-                    if len(session.planning_logs) > _MAX_PLANNING_LOGS:
-                        session.planning_logs = session.planning_logs[-_MAX_PLANNING_LOGS:]
-                # PlanningStatus / AIHelperTokensUsed are swallowed — replan is a
-                # tool call, not a top-level turn, so we don't re-yield UI events.
-        except Exception:
-            log.warning("replan.runner_failed", exc_info=True)
-            return None
-
-        return plan_text
-
-
-class ReplanTool(Tool):
-    name = "replan"
-    description = (
-        "Re-plan mid-task when the current plan is stale because of what you discovered during execution — "
-        "NOT because a step failed (that is handled by the [Adaptive re-plan] signal).\n\n"
-        "Call this when:\n"
-        "- A step turned out to be unnecessary or the real problem differs from the assumed one.\n"
-        "- New constraints, files, or facts surfaced that change the remaining work.\n"
-        "- The remaining steps no longer lead to the goal.\n\n"
-        "It re-runs the planner over your current context + todo + scratchpad findings/errors and "
-        "REPLACES the plan and todo with the new steps. Costs 1–3 LLM calls — use selectively, like `reflect`.\n\n"
-        "Do NOT call replan for a single failed step: revise the todo inline instead. "
-        "For a small adjustment (drop/merge/reorder 1-2 steps) edit the `todo` directly — replan is for when the remaining plan's overall structure is wrong. "
-        "Use `reflect` when you're unsure what's wrong (it surfaces assumptions, no plan change); use `replan` when you know what changed and need a new plan."
-    )
-    parameters = {
-        "type": "object",
-        "properties": {
-            "reason": {
-                "type": "string",
-                "description": (
-                    "What changed — the discovery that makes the current plan stale. "
-                    "One or two sentences, concrete (e.g. 'the config is TOML not JSON, so the parser step is wrong')."
-                ),
-            },
-            "updated_goal": {
-                "type": "string",
-                "description": (
-                    "Optional. The new success criterion if the goal itself shifted. "
-                    "Omit to keep the original goal."
-                ),
-            },
-        },
-        "required": ["reason"],
-    }
-
-    async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult:
-        reason = (params.get("reason") or "").strip()
-        updated_goal = (params.get("updated_goal") or "").strip() or None
-
-        if not reason:
-            return ToolResult(success=False, output="", error="reason is required")
-
-        runner = current_replan_runner.get(None)
-        if runner is None:
-            return ToolResult(
-                success=False,
-                output="",
-                error="replan is not available in this context (no active agent run)",
-            )
-
-        plan_text = await runner.replan(reason, updated_goal)
-        if not plan_text:
-            return ToolResult(
-                success=False,
-                output="",
-                error="re-planning produced no plan; keep the current plan and continue, or revise the todo inline",
-            )
-
-        output = (
-            f"# Revised plan\n\n{plan_text}\n\n---\n"
-            "Plan revised. The todo has been replaced with the new steps. "
-            "Continue from step 1 of the new plan; do not resume the old steps."
-        )
-        return ToolResult(success=True, output=output)
\ No newline at end of file
diff --git a/navi/tools/todo.py b/navi/tools/todo.py
index 539005c..4b1da81 100644
--- a/navi/tools/todo.py
+++ b/navi/tools/todo.py
@@ -317,7 +317,7 @@
             )
         elif failed > 1:
             lines.append(
-                "Discipline: multiple failures detected. Consider replanning or delegating."
+                "Discipline: multiple failures detected. Consider re-planning (`plan` with a reason) or delegating."
             )
 
         if first_iteration:
diff --git a/tests/unit/core/test_agent.py b/tests/unit/core/test_agent.py
index 760217a..30307cc 100644
--- a/tests/unit/core/test_agent.py
+++ b/tests/unit/core/test_agent.py
@@ -11,7 +11,7 @@
 import pytest
 import pytest_asyncio
 
-from navi.core.agent import Agent, _is_casual_message
+from navi.core.agent import Agent
 from navi.core.events import (
     CompressionStarted,
     ContextCompressed,
@@ -31,9 +31,6 @@
     sessions = InMemorySessionStore()
     profiles = ProfileRegistry()
     profile = make_profile("test")
-    profile.planning_phase1_enabled = False
-    profile.planning_phase2_enabled = False
-    profile.planning_phase3_enabled = False
     profiles.register(profile)
     tools = make_registry_with_tools()
     backends = BackendRegistry()
@@ -161,6 +158,29 @@
         assert saved.messages[-1].content == "streamed hello"
 
     @pytest.mark.asyncio
+    async def test_first_message_does_not_force_planning(self, agent, session):
+        """Regression: the pre-turn planning gate is gone. Even the FIRST message
+        of a session must not run the planner — planning only happens when the
+        agent calls the `plan` tool."""
+        from navi.core.events import PlanReady, PlanningStatus
+
+        backend = FakeLLMBackend(responses=["streamed hello"])
+        agent._backends.register("ollama", backend)
+
+        events = []
+        async for ev in agent.run_stream(session.id, "please build the new feature"):
+            events.append(ev)
+
+        # No planning events, no plan messages.
+        assert not any(isinstance(ev, (PlanningStatus, PlanReady)) for ev in events)
+        saved = await agent._sessions.get(session.id)
+        assert not any(getattr(m, "is_plan", False) for m in saved.messages)
+        # The planner's analysis call never ran (complete() untouched) — only
+        # the main loop's stream call hit the backend.
+        assert backend._call_idx == 0
+        assert backend._stream_idx == 1
+
+    @pytest.mark.asyncio
     async def test_run_stream_emits_model_info(self, agent, session):
         """The agent emits a ModelInfo event carrying the resolved model."""
         from typing import AsyncGenerator
@@ -739,50 +759,6 @@
         assert "thinking" in result.lower() or "stall" in result.lower()
 
 
-# ─── _is_casual_message heuristic tests ──────────────────────────────────────
-
-
-class TestIsCasualMessage:
-    def test_greetings_are_casual(self):
-        assert _is_casual_message("привет")
-        assert _is_casual_message("hi")
-        assert _is_casual_message("Hello")
-        assert _is_casual_message("здравствуйте")
-
-    def test_social_phrases_are_casual(self):
-        assert _is_casual_message("как дела?")
-        assert _is_casual_message("how are you")
-        assert _is_casual_message("спасибо")
-        assert _is_casual_message("thanks")
-        assert _is_casual_message("пока")
-
-    def test_very_short_messages_are_casual(self):
-        assert _is_casual_message("ok")
-        assert _is_casual_message("да")
-        assert _is_casual_message("yo")
-
-    def test_tool_or_command_markers_are_not_casual(self):
-        assert not _is_casual_message("/help")
-        assert not _is_casual_message("!restart")
-        assert not _is_casual_message("ping @user")
-
-    def test_urls_and_paths_are_not_casual(self):
-        assert not _is_casual_message("https://example.com")
-        assert not _is_casual_message("read /home/user/file.py")
-        assert not _is_casual_message("~/notes.md")
-
-    def test_long_messages_are_not_casual(self):
-        assert not _is_casual_message("привет, расскажи подробно как настроить сервер и что для этого нужно сделать")
-
-    def test_actual_task_requests_are_not_casual(self):
-        assert not _is_casual_message("напиши скрипт для бэкапа")
-        assert not _is_casual_message("проверь почему не работает ssh")
-        assert not _is_casual_message("what is the weather in Berlin tomorrow")
-
-
-# ─── Persistence-on-crash tests (B1: incremental flush) ─────────────────────
-
-
 class _SnapshotSessionStore(InMemorySessionStore):
     """Session store that mimics a real DB boundary.
 
@@ -853,11 +829,6 @@
 def _build_persistence_agent(store: InMemorySessionStore, backend) -> Agent:
     profiles = ProfileRegistry()
     profile = make_profile("test")
-    # Disable planning so it neither appends plan messages nor consumes LLM calls —
-    # the test controls backend calls precisely via stream_complete.
-    profile.planning_phase1_enabled = False
-    profile.planning_phase2_enabled = False
-    profile.planning_phase3_enabled = False
     profiles.register(profile)
     tools = make_registry_with_tools()
     backends = BackendRegistry()
diff --git a/tests/unit/core/test_anti_stall.py b/tests/unit/core/test_anti_stall.py
index 715410b..7545c78 100644
--- a/tests/unit/core/test_anti_stall.py
+++ b/tests/unit/core/test_anti_stall.py
@@ -46,24 +46,8 @@
         assert msg is None
 
     @pytest.mark.asyncio
-    async def test_adaptive_replan_injected_before_anti_stall(self):
-        """When both replan_msg and stall are queued, replan takes precedence."""
-        profile = make_profile(
-            anti_stall_enabled=True,
-            anti_stall_threshold=1,
-            adaptive_replan_enabled=True,
-        )
-        monitor = AntiStallMonitor(profile)
-        monitor.replan_msg = "Please replan."
-        monitor.stall_no_todo = 5
-        msg = await monitor.pre_turn("s1", iteration=1)
-        assert msg is not None
-        assert "Please replan." in msg.content
-        assert monitor.replan_msg is None  # consumed
-
-    @pytest.mark.asyncio
     async def test_disabled_anti_stall_returns_none(self):
-        profile = make_profile(anti_stall_enabled=False, adaptive_replan_enabled=False)
+        profile = make_profile(anti_stall_enabled=False)
         monitor = AntiStallMonitor(profile)
         monitor.stall_no_todo = 100
         msg = await monitor.pre_turn("s1", iteration=5)
@@ -132,138 +116,8 @@
         assert monitor.stall_repeat_tools == 0
 
     @pytest.mark.asyncio
-    async def test_adaptive_replan_queues_message(self):
-        profile = make_profile(adaptive_replan_enabled=True)
-        monitor = AntiStallMonitor(profile)
-        with patch(
-            "navi.tools.todo.get_task_snapshot",
-            new=AsyncMock(return_value=frozenset()),
-        ), patch(
-            "navi.tools.todo.get_failed_steps",
-            new=AsyncMock(return_value=frozenset({(1, "step A")})),
-        ):
-            await monitor.post_turn("s1", [])
-        assert monitor.replan_msg is not None
-        assert "step 1" in monitor.replan_msg
-        assert "step A" in monitor.replan_msg
-
-    @pytest.mark.asyncio
-    async def test_adaptive_replan_no_new_failures(self):
-        profile = make_profile(adaptive_replan_enabled=True)
-        monitor = AntiStallMonitor(profile)
-        monitor.known_failed = frozenset({(1, "step A")})
-        with patch(
-            "navi.tools.todo.get_task_snapshot",
-            new=AsyncMock(return_value=frozenset()),
-        ), patch(
-            "navi.tools.todo.get_failed_steps",
-            new=AsyncMock(return_value=frozenset({(1, "step A")})),
-        ):
-            await monitor.post_turn("s1", [])
-        assert monitor.replan_msg is None
-
-    @pytest.mark.asyncio
-    async def test_long_step_nudge_queued_at_threshold(self):
-        """adaptive_replan + a step in_progress for ``adaptive_long_step_threshold``
-        iterations (non-repeating tool calls, no new failures) → nudge the model
-        to split the step. Fires earlier than the general anti-stall warning."""
-        profile = make_profile(
-            anti_stall_enabled=True,
-            anti_stall_threshold=8,
-            adaptive_replan_enabled=True,
-            adaptive_long_step_threshold=4,
-        )
-        monitor = AntiStallMonitor(profile)
-        snapshot = frozenset({("task1", "in_progress")})
-        monitor._todo_snapshot = snapshot
-        monitor.stall_no_todo = 3  # one more no-progress iteration -> 4 == threshold
-        tc = ToolCallRequest(id="1", name="fs", arguments={"path": "/tmp"})
-        with patch(
-            "navi.tools.todo.get_task_snapshot",
-            new=AsyncMock(return_value=snapshot),
-        ), patch(
-            "navi.tools.todo.get_failed_steps",
-            new=AsyncMock(return_value=frozenset()),
-        ):
-            await monitor.post_turn("s1", [tc])
-        assert monitor.replan_msg is not None
-        assert "split it now" in monitor.replan_msg
-        assert "Adaptive re-plan" in monitor.replan_msg
-
-    @pytest.mark.asyncio
-    async def test_long_step_silent_below_threshold(self):
-        """Below the long-step threshold, no nudge is queued."""
-        profile = make_profile(
-            anti_stall_enabled=True,
-            adaptive_replan_enabled=True,
-            adaptive_long_step_threshold=4,
-        )
-        monitor = AntiStallMonitor(profile)
-        snapshot = frozenset({("task1", "in_progress")})
-        monitor._todo_snapshot = snapshot
-        monitor.stall_no_todo = 1  # -> 2, below threshold 4
-        tc = ToolCallRequest(id="1", name="fs", arguments={"path": "/tmp"})
-        with patch(
-            "navi.tools.todo.get_task_snapshot",
-            new=AsyncMock(return_value=snapshot),
-        ), patch(
-            "navi.tools.todo.get_failed_steps",
-            new=AsyncMock(return_value=frozenset()),
-        ):
-            await monitor.post_turn("s1", [tc])
-        assert monitor.replan_msg is None
-
-    @pytest.mark.asyncio
-    async def test_long_step_silent_when_model_silent(self):
-        """No tool calls this turn (model produced text only) → the long-step
-        nudge stays silent; the anti-stall path handles a genuinely stuck run."""
-        profile = make_profile(
-            anti_stall_enabled=True,
-            adaptive_replan_enabled=True,
-            adaptive_long_step_threshold=4,
-        )
-        monitor = AntiStallMonitor(profile)
-        snapshot = frozenset({("task1", "in_progress")})
-        monitor._todo_snapshot = snapshot
-        monitor.stall_no_todo = 3  # -> 4 == threshold, but no tool calls
-        with patch(
-            "navi.tools.todo.get_task_snapshot",
-            new=AsyncMock(return_value=snapshot),
-        ), patch(
-            "navi.tools.todo.get_failed_steps",
-            new=AsyncMock(return_value=frozenset()),
-        ):
-            await monitor.post_turn("s1", [])
-        assert monitor.replan_msg is None
-
-    @pytest.mark.asyncio
-    async def test_long_step_failed_takes_precedence(self):
-        """New failures take precedence over the long-step nudge."""
-        profile = make_profile(
-            anti_stall_enabled=True,
-            adaptive_replan_enabled=True,
-            adaptive_long_step_threshold=4,
-        )
-        monitor = AntiStallMonitor(profile)
-        snapshot = frozenset({("task1", "in_progress")})
-        monitor._todo_snapshot = snapshot
-        monitor.stall_no_todo = 3  # -> 4 == threshold
-        tc = ToolCallRequest(id="1", name="fs", arguments={"path": "/tmp"})
-        with patch(
-            "navi.tools.todo.get_task_snapshot",
-            new=AsyncMock(return_value=snapshot),
-        ), patch(
-            "navi.tools.todo.get_failed_steps",
-            new=AsyncMock(return_value=frozenset({(2, "step B")})),
-        ):
-            await monitor.post_turn("s1", [tc])
-        assert monitor.replan_msg is not None
-        assert "just failed" in monitor.replan_msg
-        assert "split it now" not in monitor.replan_msg
-
-    @pytest.mark.asyncio
     async def test_disabled_anti_stall_does_not_fetch_snapshot(self):
-        profile = make_profile(anti_stall_enabled=False, adaptive_replan_enabled=False)
+        profile = make_profile(anti_stall_enabled=False)
         monitor = AntiStallMonitor(profile)
         with patch("navi.tools.todo.get_task_snapshot") as mock_snap:
             await monitor.post_turn("s1", [])
diff --git a/tests/unit/core/test_planning.py b/tests/unit/core/test_planning.py
index cbd2f91..3945902 100644
--- a/tests/unit/core/test_planning.py
+++ b/tests/unit/core/test_planning.py
@@ -10,9 +10,11 @@
     def __init__(self, responses):
         self.responses = list(responses)
         self.calls = []
+        self.kwargs = []
 
     async def complete(self, messages, **kwargs):
         self.calls.append(messages)
+        self.kwargs.append(kwargs)
         return LLMResponse(
             content=self.responses.pop(0),
             tool_calls=None,
@@ -75,7 +77,6 @@
     async def test_planning_prompt_includes_profile_mcp_and_persistence_rules(self):
         profile = make_profile(
             "server_admin",
-            planning_phase2_enabled=False,
             mcp_servers={"gnexus-book": ["read", "write"]},
         )
         llm = RecordingLLM([
@@ -98,7 +99,6 @@
             "SUBTASKS:\n"
             "1. Search docs\n"
             "2. Persist facts\n"
-            "REFLECT: no\n"
             "COMMITMENTS: checkpoint",
             "## Plan\n\n"
             "**Task:** document infra\n"
@@ -126,30 +126,53 @@
         assert "knowledge persistence checkpoint" in phase3_prompt
         assert "Do not plan unavailable MCP tool calls" in phase3_prompt
 
-    async def test_planning_prompt_includes_direct_shortcut(self):
-        profile = make_profile(
-            "developer",
-            planning_phase2_enabled=False,
-        )
+    async def test_direct_shortcut_only_offered_to_subagents(self):
+        """Top-level planning is a deliberate `plan` tool call — no DIRECT offer.
+        Sub-agents plan automatically, so they keep the shortcut for trivial
+        subtasks."""
+        profile = make_profile("developer")
         llm = RecordingLLM(["DIRECT"])
         engine = PlanningEngine(FakeContextBuilder())
         context = [Message(role="user", content="hello")]
 
         async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
             pass
+        top_level_prompt = llm.calls[0][0].content
+        assert "CRITICAL DIRECT shortcut" not in top_level_prompt
 
-        phase1_prompt = llm.calls[0][0].content
-        assert "CRITICAL DIRECT shortcut" in phase1_prompt
-        assert "greeting" in phase1_prompt
-        assert "DIRECT (uppercase)" in phase1_prompt
+        llm_sub = RecordingLLM(["DIRECT"])
+        async for _event in engine.run(
+            context, profile, llm_sub, mem=None, tool_schemas=[], is_subagent=True
+        ):
+            pass
+        subagent_prompt = llm_sub.calls[0][0].content
+        assert "CRITICAL DIRECT shortcut" in subagent_prompt
+        assert "greeting" in subagent_prompt
+        assert "DIRECT (uppercase)" in subagent_prompt
+
+    async def test_subagent_direct_shortcut_skips_plan(self):
+        profile = make_profile("developer")
+        llm = RecordingLLM(["DIRECT"])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [Message(role="user", content="hello")]
+
+        events = []
+        async for event in engine.run(context, profile, llm, mem=None, tool_schemas=[], is_subagent=True):
+            events.append(event)
+
+        assert len(llm.calls) == 1
+        assert not any(isinstance(e, PlanReady) for e in events)
 
     async def test_planning_prompt_omits_mcp_when_profile_has_no_mcp_servers(self):
         profile = make_profile(
             "developer",
-            planning_phase2_enabled=False,
             mcp_servers={},
         )
-        llm = RecordingLLM(["DIRECT"])
+        llm = RecordingLLM([
+            "TASK: hi\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: simple\nSUBTASKS:\n1. Step\nCOMMITMENTS: none",
+            "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
         engine = PlanningEngine(FakeContextBuilder())
         context = [Message(role="user", content="hello")]
 
@@ -160,13 +183,16 @@
         assert "gnexus-book instructions" not in phase1_prompt
         assert "Connected MCP knowledge servers are authoritative only when the active profile exposes their tools" in phase1_prompt
 
-    async def test_planning_flags(self):
-        """Planning messages must have correct is_display / is_context flags."""
-        profile = make_profile("developer", planning_phase2_enabled=False)
+    async def test_planning_flags_and_top_level_has_no_execute_prompt(self):
+        """Planning messages must have correct is_display / is_context flags.
+        Top-level runs (the `plan` tool path) get NO injected "execute now"
+        prompt — the tool result carries the follow-up instruction instead.
+        Sub-agents still get the injected go-ahead."""
+        profile = make_profile("developer")
         llm = RecordingLLM([
             "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
             "KNOWLEDGE SOURCE ASSESSMENT: NONE\nKNOWLEDGE CAPTURE: NONE\n"
-            "COMPLEXITY: low\nSUBTASKS:\n1. Step one\nREFLECT: no\nCOMMITMENTS: none",
+            "COMPLEXITY: simple\nSUBTASKS:\n1. Step one\nCOMMITMENTS: none",
             "## Plan\n\n**Task:** test\n**Goal:** done\n\n**Steps:**\n1. Step one → SELF\n",
         ])
         engine = PlanningEngine(FakeContextBuilder())
@@ -176,7 +202,6 @@
         async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[], messages=messages):
             pass
 
-        # Messages list should contain: user, plan context (is_display=False), plan marker (is_context=False), prompt
         plan_ctx = [m for m in messages if m.role == "assistant" and not m.is_plan and m.is_display is False]
         plan_marker = [m for m in messages if m.is_plan is True]
         prompt_msgs = [m for m in messages if m.role == "user" and m.content.startswith("Plan is ready")]
@@ -186,82 +211,40 @@
         assert len(plan_marker) == 1
         assert plan_marker[0].is_context is False
         assert plan_marker[0].is_display is True
+        # Top-level: the plan tool composes the follow-up — nothing injected.
+        assert prompt_msgs == []
+
+    async def test_subagent_gets_execute_prompt(self):
+        profile = make_profile("developer")
+        llm = RecordingLLM([
+            "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: simple\nSUBTASKS:\n1. Step one\nCOMMITMENTS: none",
+            "## Plan\n\n**Steps:**\n1. Step one → SELF\n",
+        ])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [Message(role="user", content="do it")]
+        messages = []
+
+        async for _event in engine.run(
+            context, profile, llm, mem=None, tool_schemas=[], messages=messages, is_subagent=True
+        ):
+            pass
+
+        prompt_msgs = [m for m in messages if m.role == "user" and m.content.startswith("Plan is ready")]
         assert len(prompt_msgs) == 1
         assert prompt_msgs[0].is_display is False
 
 
-_OBSERVE_ANALYSIS = (
-    "TASK: look at directory X\n"
-    "GOAL: report what is in directory X\n"
-    "MODE: observe\n"
-    "UNKNOWNS: NONE\nRESOURCES: NONE\nKNOWLEDGE SOURCE ASSESSMENT: NONE\n"
-    "KNOWLEDGE CAPTURE: NONE\nCOMPLEXITY: simple\nSUBTASKS:\n1. List directory\n"
-    "REFLECT: no\nCOMMITMENTS: none"
-)
-_ACT_ANALYSIS = _OBSERVE_ANALYSIS.replace("MODE: observe", "MODE: act")
-
-
-class TestObserveSkip:
-    async def test_observe_skips_phase3(self):
-        profile = make_profile(
-            "developer",
-            planning_phase2_enabled=False,
-            observe_skips_plan_enabled=True,
-        )
-        # Only one response needed — observe must stop after Phase 1.
-        llm = RecordingLLM([_OBSERVE_ANALYSIS])
-        engine = PlanningEngine(FakeContextBuilder())
-        context = [Message(role="user", content="look at directory X")]
-
-        events = []
-        async for event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
-            events.append(event)
-
-        # Only Phase 1 ran — single LLM call, no plan produced.
-        assert len(llm.calls) == 1
-        assert not any(isinstance(e, PlanReady) for e in events)
-
-    async def test_observe_flag_off_still_plans(self):
-        profile = make_profile(
-            "developer",
-            planning_phase2_enabled=False,
-            observe_skips_plan_enabled=False,
-        )
-        llm = RecordingLLM([_OBSERVE_ANALYSIS, "## Plan\n\n**Steps:**\n1. List → TOOL: filesystem\n"])
-        engine = PlanningEngine(FakeContextBuilder())
-        context = [Message(role="user", content="look at directory X")]
-
-        events = []
-        async for event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
-            events.append(event)
-
-        # Default-off: observe classification is ignored, Phase 3 runs.
-        assert len(llm.calls) == 2
-        assert any(isinstance(e, PlanReady) for e in events)
-
-    async def test_act_mode_still_plans(self):
-        profile = make_profile(
-            "developer",
-            planning_phase2_enabled=False,
-            observe_skips_plan_enabled=True,
-        )
-        llm = RecordingLLM([_ACT_ANALYSIS, "## Plan\n\n**Steps:**\n1. Build → TOOL: filesystem\n"])
-        engine = PlanningEngine(FakeContextBuilder())
-        context = [Message(role="user", content="build a thing")]
-
-        events = []
-        async for event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
-            events.append(event)
-
-        # act mode does not skip — Phase 3 runs even with the flag on.
-        assert len(llm.calls) == 2
-        assert any(isinstance(e, PlanReady) for e in events)
-
-
 class TestPhase1Prompt:
-    async def test_phase1_prompt_includes_mode_classification(self):
-        profile = make_profile("developer", planning_phase2_enabled=False)
-        llm = RecordingLLM(["DIRECT"])
+    async def test_phase1_prompt_has_complexity_no_mode_no_reflect(self):
+        """MODE/observe and REFLECT/Phase 2 are gone; COMPLEXITY remains — the
+        `plan` tool picks the confirmation instruction from it."""
+        profile = make_profile("developer")
+        llm = RecordingLLM([
+            "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: medium\nSUBTASKS:\n1. Step\nCOMMITMENTS: none",
+            "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
         engine = PlanningEngine(FakeContextBuilder())
         context = [Message(role="user", content="hello")]
 
@@ -269,8 +252,168 @@
             pass
 
         phase1_prompt = llm.calls[0][0].content
-        assert "MODE: observe | act" in phase1_prompt
-        assert "reading files to answer is still observe" in phase1_prompt
+        assert "COMPLEXITY: simple | medium | complex" in phase1_prompt
+        assert "MODE: observe | act" not in phase1_prompt
+        assert "REFLECT" not in phase1_prompt
+        assert "PHASE 2" not in phase1_prompt
+
+
+class TestPlanningThinkFlags:
+    async def test_both_phases_call_llm_with_think_false(self):
+        """Non-streaming planner calls must not request extended reasoning:
+        cloud reasoning models leak their chain-of-thought into content with
+        think=True (gemma4 emits "thought<channel|>" headers, glm-flash returns
+        empty content). think=False keeps the structured output clean —
+        consistent with compressor / ai_helper."""
+        profile = make_profile("developer", think_enabled=True)
+        llm = RecordingLLM([
+            "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: medium\nSUBTASKS:\n1. Step\nCOMMITMENTS: none",
+            "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [Message(role="user", content="hello")]
+
+        async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
+            pass
+
+        assert len(llm.kwargs) == 2
+        assert all(kw.get("think") is False for kw in llm.kwargs)
+
+
+class TestChannelArtifactStrip:
+    """gemma4 on ollama-cloud leaks its thinking-channel header into content
+    on non-streaming calls — the planner strips the artifact so the structured
+    output parses cleanly."""
+
+    _ARTIFACT = "thought\n<channel|>"
+
+    async def test_phase1_artifact_stripped_before_parsing(self):
+        profile = make_profile("developer")
+        llm = RecordingLLM([
+            self._ARTIFACT
+            + "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: complex\nSUBTASKS:\n1. Step\nCOMMITMENTS: none",
+            "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [Message(role="user", content="risky task")]
+
+        events = []
+        async for event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
+            events.append(event)
+
+        # Without stripping the artifact the DIRECT/empty check would not fire,
+        # but COMPLEXITY sits right behind the junk — it must be parsed.
+        assert engine.last_complexity == "complex"
+        # The debug log stores the cleaned analysis.
+        dbg = [e for e in events if type(e).__name__ == "PlanningDebugData"]
+        assert dbg and not dbg[-1].log["phases"]["1"]["output"].startswith("thought")
+
+    async def test_phase3_artifact_stripped_from_plan_text(self):
+        profile = make_profile("developer")
+        llm = RecordingLLM([
+            "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: simple\nSUBTASKS:\n1. Step\nCOMMITMENTS: none",
+            self._ARTIFACT + "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [Message(role="user", content="hello")]
+
+        events = []
+        async for event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
+            events.append(event)
+
+        ready = [e for e in events if isinstance(e, PlanReady)]
+        assert len(ready) == 1
+        assert ready[0].plan.startswith("## Plan")
+
+
+class TestPhase1ContextWindow:
+    """Mid-session plan calls arrive after many tool results; Phase 1 windows
+    the conversation to a char budget, keeping the newest messages and pinning
+    the first user message (the original task statement)."""
+
+    async def test_small_context_kept_intact(self):
+        profile = make_profile("developer")
+        llm = RecordingLLM([
+            "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: simple\nSUBTASKS:\n1. Step\nCOMMITMENTS: none",
+            "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [
+            Message(role="user", content="original task"),
+            Message(role="assistant", content="doing it"),
+            Message(role="user", content="latest request"),
+        ]
+
+        async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
+            pass
+
+        phase1_prompt = "\n".join(m.content or "" for m in llm.calls[0])
+        for needle in ("original task", "doing it", "latest request"):
+            assert needle in phase1_prompt
+
+    async def test_large_context_windowed_with_first_user_message_pinned(self):
+        from navi.core.planning import _PHASE1_CONTEXT_MAX_CHARS
+
+        profile = make_profile("developer")
+        llm = RecordingLLM([
+            "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: simple\nSUBTASKS:\n1. Step\nCOMMITMENTS: none",
+            "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
+        engine = PlanningEngine(FakeContextBuilder())
+        # One huge mid-conversation tool result that alone blows the budget…
+        filler = "x" * (_PHASE1_CONTEXT_MAX_CHARS + 5000)
+        context = [
+            Message(role="user", content="original task statement"),
+            Message(role="user", content=filler),        # dropped by the window
+            Message(role="assistant", content="recent small note"),
+            Message(role="user", content="latest request"),
+        ]
+
+        async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
+            pass
+
+        phase1_prompt = "\n".join(m.content or "" for m in llm.calls[0])
+        assert len(phase1_prompt) < _PHASE1_CONTEXT_MAX_CHARS + 10_000  # system prompt + window
+        assert "original task statement" in phase1_prompt  # pinned first user message
+        assert "latest request" in phase1_prompt           # newest message kept
+        assert "recent small note" in phase1_prompt        # recent exchange kept
+        assert filler[:100] not in phase1_prompt           # huge message dropped
+
+
+class TestComplexityExport:
+    async def test_last_complexity_parsed_from_phase1(self):
+        profile = make_profile("developer")
+        llm = RecordingLLM([
+            "TASK: test\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+            "COMPLEXITY: complex\nSUBTASKS:\n1. Step\nCOMMITMENTS: none",
+            "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [Message(role="user", content="risky task")]
+
+        async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
+            pass
+
+        assert engine.last_complexity == "complex"
+
+    async def test_last_complexity_empty_when_unparseable(self):
+        profile = make_profile("developer")
+        llm = RecordingLLM([
+            "garbage without classification",
+            "## Plan\n\n**Steps:**\n1. Step → SELF\n",
+        ])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [Message(role="user", content="hello")]
+
+        async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
+            pass
+
+        assert engine.last_complexity == ""
 
 
 class _MemKv:
@@ -300,9 +443,13 @@
         kv = _MemKv()
         monkeypatch.setattr(todo_mod, "_kv_store", kv)
 
-        profile = make_profile("developer", planning_phase2_enabled=False)
+        profile = make_profile("developer")
         llm = RecordingLLM(
-            [_ACT_ANALYSIS, "## Plan\n\n**Steps:**\n1. Build → TOOL: filesystem\n"]
+            [
+                "TASK: build\nGOAL: done\nUNKNOWNS: NONE\nRESOURCES: NONE\n"
+                "COMPLEXITY: medium\nSUBTASKS:\n1. Build\nCOMMITMENTS: none",
+                "## Plan\n\n**Steps:**\n1. Build → TOOL: filesystem\n",
+            ]
         )
         engine = PlanningEngine(FakeContextBuilder())
         context = [Message(role="user", content="build a thing")]
@@ -318,4 +465,4 @@
 
         # Sub-agent's KV row received the plan step; parent row is empty.
         assert await kv.get(None, "sub_run_xyz", "todo", "tasks") is not None
-        assert await kv.get(None, "parent_sess", "todo", "tasks") is None
+        assert await kv.get(None, "parent_sess", "todo", "tasks") is None
\ No newline at end of file
diff --git a/tests/unit/tools/test_plan.py b/tests/unit/tools/test_plan.py
new file mode 100644
index 0000000..0c7d29c
--- /dev/null
+++ b/tests/unit/tools/test_plan.py
@@ -0,0 +1,411 @@
+"""Unit tests for navi.tools.plan — PlanTool + PlanRunner.
+
+Covers:
+- PlanTool schema (both params optional) and the no-runner guard.
+- Fresh planning (no reason): planner runs without [RE-PLAN] framing.
+- Revision mode (reason given): runner packs reason + goal + todo + scratchpad
+  into replan_context, re-runs PlanningEngine with is_replan=True, captures
+  PlanReady.plan, logs PlanningDebugData, and replaces the todo.
+- PlanningStatus / PlanReady are forwarded to the event sink (UI sees them
+  mid-turn); the execute prompt is NOT injected on the tool path.
+- The tool result's follow-up instruction follows COMPLEXITY: complex →
+  present and wait for confirmation; otherwise proceed.
+- The re-plan prompt framing ([RE-PLAN] block) is present only with context.
+"""
+
+import asyncio
+
+import pytest
+
+from navi.core.events import PlanReady, PlanningStatus
+from navi.core.planning import PlanningEngine
+from navi.llm.base import LLMResponse, Message
+from navi.tools._internal.base import (
+    ToolContext,
+    current_event_sink,
+    current_plan_runner,
+    current_session_id,
+    current_user_id,
+)
+from navi.tools.plan import PlanRunner, PlanTool
+from navi.tools import scratchpad as scratchpad_mod
+from navi.tools import todo as todo_mod
+from tests.conftest_factory import make_profile
+
+
+class RecordingLLM:
+    def __init__(self, responses):
+        self.responses = list(responses)
+        self.calls = []
+
+    async def complete(self, messages, **kwargs):
+        self.calls.append(messages)
+        return LLMResponse(
+            content=self.responses.pop(0),
+            tool_calls=None,
+            finish_reason="stop",
+        )
+
+
+class FakeContextBuilder:
+    def build_system_prompt(self, profile):
+        return "base system prompt"
+
+    def _mcp_context_msg(self, profile=None):
+        return None
+
+
+class FakeKvStore:
+    """In-memory KV store for todo + scratchpad tests."""
+
+    def __init__(self):
+        self._data: dict[tuple, str] = {}
+
+    async def get(self, user_id, session_id, scope, key):
+        return self._data.get((user_id or "", session_id, scope, key))
+
+    async def set(self, user_id, session_id, scope, key, value):
+        self._data[(user_id or "", session_id, scope, key)] = value
+
+    async def get_all(self, user_id, session_id, scope):
+        return {
+            k[3]: v
+            for k, v in self._data.items()
+            if k[:3] == (user_id or "", session_id, scope)
+        }
+
+    async def delete(self, user_id, session_id, scope, key):
+        self._data.pop((user_id or "", session_id, scope, key), None)
+
+    async def clear_scope(self, user_id, session_id, scope):
+        keys = [k for k in self._data if k[:3] == (user_id or "", session_id, scope)]
+        for k in keys:
+            del self._data[k]
+
+
+class FakeSession:
+    def __init__(self, sid="sess_plan", user_id="u1"):
+        self.id = sid
+        self.user_id = user_id
+        self.context: list[Message] = []
+        self.messages: list[Message] = []
+        self.planning_logs: list[dict] = []
+
+
+@pytest.fixture(autouse=True)
+def _fake_kv():
+    store = FakeKvStore()
+    todo_mod._kv_store = store
+    scratchpad_mod._kv_store = store
+    yield store
+    todo_mod._kv_store = None
+    scratchpad_mod._kv_store = None
+
+
+_FRESH_ANALYSIS = (
+    "TASK: build feature\n"
+    "GOAL: feature works\n"
+    "UNKNOWNS: NONE\nRESOURCES: NONE\nKNOWLEDGE SOURCE ASSESSMENT: NONE\n"
+    "KNOWLEDGE CAPTURE: NONE\nCOMPLEXITY: simple\nSUBTASKS:\n1. New step\n"
+    "COMMITMENTS: none"
+)
+_REPLAN_ANALYSIS = (
+    "TASK: revised task\n"
+    "GOAL: revised goal\n"
+    "UNKNOWNS: NONE\nRESOURCES: NONE\nKNOWLEDGE SOURCE ASSESSMENT: NONE\n"
+    "KNOWLEDGE CAPTURE: NONE\nCOMPLEXITY: simple\nSUBTASKS:\n1. New step\n"
+    "COMMITMENTS: none"
+)
+_COMPLEX_ANALYSIS = _REPLAN_ANALYSIS.replace("COMPLEXITY: simple", "COMPLEXITY: complex")
+_PLAN = (
+    "## Plan\n\n"
+    "**Task:** revised task\n**Goal:** revised goal\n\n"
+    "**Steps:**\n"
+    "1. New step one → TOOL: filesystem\n"
+    "2. New step two → SELF\n\n"
+    "**Parallel:** NONE\n**Risks:** NONE"
+)
+
+
+def _make_runner(responses, session=None, profile=None):
+    session = session or FakeSession()
+    profile = profile or make_profile("developer")
+    llm = RecordingLLM(responses)
+    engine = PlanningEngine(FakeContextBuilder())
+    runner = PlanRunner(engine, session, profile, llm, mem=None, tool_schemas=[])
+    return runner, session, llm, engine, profile
+
+
+def _bind_runner(runner):
+    token = current_plan_runner.set(runner)
+    return token
+
+
+# ── PlanTool schema ───────────────────────────────────────────────────────────
+
+
+class TestPlanToolSchema:
+    def test_name_and_params(self):
+        t = PlanTool()
+        assert t.name == "plan"
+        assert t.parameters["required"] == []
+        assert "reason" in t.parameters["properties"]
+        assert "updated_goal" in t.parameters["properties"]
+
+    async def test_no_runner_returns_error(self):
+        # No current_plan_runner set in this context — both fresh and revision
+        # calls must report unavailability.
+        t = PlanTool()
+        for params in ({}, {"reason": "plan is stale"}):
+            result = await t.execute(params, ToolContext())
+            assert result.success is False
+            assert "not available" in (result.error or "")
+
+    async def test_events_reach_sink_via_contextvar_when_ctx_has_none(self):
+        """The agent loop passes ToolContext(event_sink=None) — the ContextVar is
+        the real channel. The tool must fall back to current_event_sink, not
+        swallow PlanningStatus/PlanReady when ctx.event_sink is None."""
+        runner, session, llm, engine, profile = _make_runner([_REPLAN_ANALYSIS, _PLAN])
+        sink: asyncio.Queue = asyncio.Queue()
+        sink_tok = current_event_sink.set(sink)
+        runner_tok = _bind_runner(runner)
+        try:
+            result = await PlanTool().execute(
+                {"reason": "stale"}, ToolContext(event_sink=None)
+            )
+        finally:
+            current_plan_runner.reset(runner_tok)
+            current_event_sink.reset(sink_tok)
+
+        assert result.success is True
+        events = []
+        while not sink.empty():
+            events.append(sink.get_nowait())
+        assert any(isinstance(e, PlanningStatus) for e in events)
+        assert any(isinstance(e, PlanReady) and e.plan == _PLAN for e in events)
+
+
+# ── PlanRunner + PlanningEngine integration ──────────────────────────────────
+
+
+class TestPlanRunner:
+    async def test_fresh_plan_runs_planner_without_replan_framing(self):
+        runner, session, llm, engine, profile = _make_runner([_FRESH_ANALYSIS, _PLAN])
+
+        plan = await runner.plan(None, None, event_sink=None)
+
+        assert plan == _PLAN
+        # Phase 1 + Phase 3 ran.
+        assert len(llm.calls) == 2
+        phase1_prompt = llm.calls[0][0].content
+        assert "[RE-PLAN]" not in phase1_prompt
+        assert "Reason for re-plan" not in phase1_prompt
+        assert "Read the user's latest request." in phase1_prompt
+
+    async def test_revision_runs_planner_with_is_replan_and_returns_plan(self):
+        runner, session, llm, engine, profile = _make_runner([_REPLAN_ANALYSIS, _PLAN])
+
+        plan = await runner.plan("config is TOML not JSON", None, event_sink=None)
+
+        assert plan == _PLAN
+        assert len(llm.calls) == 2
+        phase1_prompt = llm.calls[0][0].content
+        assert "[RE-PLAN]" in phase1_prompt
+        assert "config is TOML not JSON" in phase1_prompt
+        # Top-level runs never see the DIRECT shortcut.
+        assert "CRITICAL DIRECT shortcut" not in phase1_prompt
+        # Re-plan opening line used instead of the fresh-request one.
+        assert "Re-plan based on the [RE-PLAN] context" in phase1_prompt
+        assert "Read the user's latest request." not in phase1_prompt
+
+    async def test_replan_context_includes_todo_findings_errors(self):
+        # Seed a todo + scratchpad sections the runner must pack into context.
+        sid_tok = current_session_id.set("sess_plan")
+        uid_tok = current_user_id.set("u1")
+        try:
+            from navi.tools.todo import set_tasks
+            await set_tasks("sess_plan", ["Old step one", "Old step two"])
+            await scratchpad_mod._kv_store.set("u1", "sess_plan", "scratchpad", "findings", "config parser expects TOML")
+            await scratchpad_mod._kv_store.set("u1", "sess_plan", "scratchpad", "errors", "JSON parse failed at line 3")
+
+            runner, session, llm, engine, profile = _make_runner(
+                [_REPLAN_ANALYSIS, _PLAN], session=FakeSession(sid="sess_plan", user_id="u1")
+            )
+
+            plan = await runner.plan("config is TOML not JSON", "parse config correctly", event_sink=None)
+
+            assert plan == _PLAN
+            phase1_prompt = llm.calls[0][0].content
+            assert "Reason for re-plan: config is TOML not JSON" in phase1_prompt
+            assert "Updated goal: parse config correctly" in phase1_prompt
+            assert "Old step one" in phase1_prompt
+            assert "config parser expects TOML" in phase1_prompt
+            assert "JSON parse failed at line 3" in phase1_prompt
+        finally:
+            current_user_id.reset(uid_tok)
+            current_session_id.reset(sid_tok)
+
+    async def test_revision_replaces_todo_with_new_steps(self):
+        from navi.tools.todo import _load_tasks, set_tasks
+
+        sid_tok = current_session_id.set("sess_plan")
+        uid_tok = current_user_id.set("u1")
+        try:
+            await set_tasks("sess_plan", ["Old step one", "Old step two"])
+            assert [t.text for t in await _load_tasks("sess_plan")] == ["Old step one", "Old step two"]
+
+            runner, session, llm, engine, profile = _make_runner(
+                [_REPLAN_ANALYSIS, _PLAN], session=FakeSession(sid="sess_plan")
+            )
+
+            plan = await runner.plan("stale plan", None, event_sink=None)
+
+            assert plan == _PLAN
+            tasks = await _load_tasks("sess_plan")
+            assert [t.text for t in tasks] == [
+                "New step one → TOOL: filesystem",
+                "New step two → SELF",
+            ]
+        finally:
+            current_user_id.reset(uid_tok)
+            current_session_id.reset(sid_tok)
+
+    async def test_logs_planning_debug_data_to_session(self):
+        runner, session, llm, engine, profile = _make_runner([_REPLAN_ANALYSIS, _PLAN])
+
+        await runner.plan("stale plan", None, event_sink=None)
+
+        assert len(session.planning_logs) == 1
+        assert session.planning_logs[0]["result"] == "plan"
+
+    async def test_appends_plan_to_session_context_without_execute_prompt(self):
+        """The plan message lands in the session, but the top-level tool path
+        must NOT inject the 'Plan is ready. Execute it now' prompt — the tool
+        result carries the follow-up instruction instead."""
+        runner, session, llm, engine, profile = _make_runner([_REPLAN_ANALYSIS, _PLAN])
+
+        await runner.plan("stale plan", None, event_sink=None)
+
+        plan_msgs = [m for m in session.context if m.role == "assistant"]
+        assert any(_PLAN in (m.content or "") for m in plan_msgs)
+        prompt_msgs = [
+            m for m in session.context
+            if m.role == "user" and "Plan is ready" in (m.content or "")
+        ]
+        assert prompt_msgs == []
+
+    async def test_planning_events_forwarded_to_event_sink(self):
+        """PlanningStatus + PlanReady must reach the sink so the UI shows
+        planning progress and the plan card mid-turn."""
+        runner, session, llm, engine, profile = _make_runner([_REPLAN_ANALYSIS, _PLAN])
+        sink: asyncio.Queue = asyncio.Queue()
+
+        await runner.plan("stale plan", None, event_sink=sink)
+
+        events = []
+        while not sink.empty():
+            events.append(sink.get_nowait())
+        assert any(isinstance(e, PlanningStatus) for e in events)
+        assert any(isinstance(e, PlanReady) and e.plan == _PLAN for e in events)
+
+    async def test_returns_none_on_planner_failure(self):
+        class _BoomLLM:
+            async def complete(self, messages, **kwargs):
+                raise RuntimeError("llm down")
+
+        session = FakeSession()
+        profile = make_profile("developer")
+        engine = PlanningEngine(FakeContextBuilder())
+        runner = PlanRunner(engine, session, profile, _BoomLLM(), mem=None, tool_schemas=[])
+
+        assert await runner.plan("stale plan", None, event_sink=None) is None
+
+
+# ── PlanTool follow-up instruction (confirmation by COMPLEXITY) ────────────────
+
+
+class TestPlanToolConfirmationInstruction:
+    async def test_complex_task_asks_to_wait_for_confirmation(self):
+        runner, session, llm, engine, profile = _make_runner([_COMPLEX_ANALYSIS, _PLAN])
+        token = _bind_runner(runner)
+        try:
+            result = await PlanTool().execute({"reason": "stale plan"}, ToolContext())
+        finally:
+            current_plan_runner.reset(token)
+
+        assert result.success is True
+        assert "# Revised plan" in result.output
+        assert _PLAN in result.output
+        assert "COMPLEX" in result.output
+        assert "confirmation" in result.output
+        assert "Do not start executing yet" in result.output
+
+    async def test_simple_task_proceeds(self):
+        runner, session, llm, engine, profile = _make_runner([_FRESH_ANALYSIS, _PLAN])
+        token = _bind_runner(runner)
+        try:
+            result = await PlanTool().execute({}, ToolContext())
+        finally:
+            current_plan_runner.reset(token)
+
+        assert result.success is True
+        assert "# Plan" in result.output
+        assert "Proceed with execution now" in result.output
+        assert "confirmation" not in result.output
+
+    async def test_fresh_failure_returns_error_without_reason(self):
+        class _BoomLLM:
+            async def complete(self, messages, **kwargs):
+                raise RuntimeError("llm down")
+
+        session = FakeSession()
+        profile = make_profile("developer")
+        engine = PlanningEngine(FakeContextBuilder())
+        runner = PlanRunner(engine, session, profile, _BoomLLM(), mem=None, tool_schemas=[])
+        token = _bind_runner(runner)
+        try:
+            result = await PlanTool().execute({}, ToolContext())
+        finally:
+            current_plan_runner.reset(token)
+
+        assert result.success is False
+        assert "produced no plan" in (result.error or "")
+
+    async def test_revision_failure_returns_error_with_reason(self):
+        class _BoomLLM:
+            async def complete(self, messages, **kwargs):
+                raise RuntimeError("llm down")
+
+        session = FakeSession()
+        profile = make_profile("developer")
+        engine = PlanningEngine(FakeContextBuilder())
+        runner = PlanRunner(engine, session, profile, _BoomLLM(), mem=None, tool_schemas=[])
+        token = _bind_runner(runner)
+        try:
+            result = await PlanTool().execute({"reason": "stale"}, ToolContext())
+        finally:
+            current_plan_runner.reset(token)
+
+        assert result.success is False
+        assert "re-planning produced no plan" in (result.error or "")
+
+
+# ── PlanningEngine is_replan prompt behavior ──────────────────────────────────
+
+
+class TestPlanningIsReplan:
+    async def test_replan_context_omitted_does_not_inject_block(self):
+        """is_replan without replan_context must not inject an empty [RE-PLAN] block."""
+        profile = make_profile("developer")
+        llm = RecordingLLM([_REPLAN_ANALYSIS])
+        engine = PlanningEngine(FakeContextBuilder())
+        context = [Message(role="user", content="hi")]
+
+        async for _ev in engine.run(
+            context, profile, llm, mem=None, tool_schemas=[],
+            is_replan=True, replan_context=None,
+        ):
+            pass
+
+        phase1_prompt = llm.calls[0][0].content
+        assert "[RE-PLAN]" not in phase1_prompt
\ No newline at end of file
diff --git a/tests/unit/tools/test_replan.py b/tests/unit/tools/test_replan.py
deleted file mode 100644
index c7f0c86..0000000
--- a/tests/unit/tools/test_replan.py
+++ /dev/null
@@ -1,320 +0,0 @@
-"""Unit tests for navi.tools.replan — ReplanTool + ReplanRunner.
-
-Covers:
-- ReplanTool schema/param requirements and the no-runner / missing-reason guards.
-- ReplanRunner packs reason + goal + todo + scratchpad into replan_context,
-  re-runs PlanningEngine with is_replan=True (DIRECT + observe-skip suppressed),
-  captures PlanReady.plan, logs PlanningDebugData, and replaces the todo.
-- The re-plan prompt framing ([RE-PLAN] block, re-plan opening line) is present
-  and the DIRECT shortcut / observe-skip are suppressed for is_replan.
-"""
-
-import pytest
-
-from navi.core.events import PlanReady
-from navi.core.planning import PlanningEngine
-from navi.llm.base import LLMResponse, Message
-from navi.tools._internal.base import (
-    ToolContext,
-    current_replan_runner,
-    current_session_id,
-    current_user_id,
-)
-from navi.tools.replan import ReplanRunner, ReplanTool
-from navi.tools import scratchpad as scratchpad_mod
-from navi.tools import todo as todo_mod
-from tests.conftest_factory import FakePool, make_profile
-
-
-class RecordingLLM:
-    def __init__(self, responses):
-        self.responses = list(responses)
-        self.calls = []
-
-    async def complete(self, messages, **kwargs):
-        self.calls.append(messages)
-        return LLMResponse(
-            content=self.responses.pop(0),
-            tool_calls=None,
-            finish_reason="stop",
-        )
-
-
-class FakeContextBuilder:
-    def build_system_prompt(self, profile):
-        return "base system prompt"
-
-    def _mcp_context_msg(self, profile=None):
-        return None
-
-
-class FakeKvStore:
-    """In-memory KV store for todo + scratchpad tests."""
-
-    def __init__(self):
-        self._data: dict[tuple, str] = {}
-
-    async def get(self, user_id, session_id, scope, key):
-        return self._data.get((user_id or "", session_id, scope, key))
-
-    async def set(self, user_id, session_id, scope, key, value):
-        self._data[(user_id or "", session_id, scope, key)] = value
-
-    async def get_all(self, user_id, session_id, scope):
-        return {
-            k[3]: v
-            for k, v in self._data.items()
-            if k[:3] == (user_id or "", session_id, scope)
-        }
-
-    async def delete(self, user_id, session_id, scope, key):
-        self._data.pop((user_id or "", session_id, scope, key), None)
-
-    async def clear_scope(self, user_id, session_id, scope):
-        keys = [k for k in self._data if k[:3] == (user_id or "", session_id, scope)]
-        for k in keys:
-            del self._data[k]
-
-
-class FakeSession:
-    def __init__(self, sid="sess_replan", user_id="u1"):
-        self.id = sid
-        self.user_id = user_id
-        self.context: list[Message] = []
-        self.messages: list[Message] = []
-        self.planning_logs: list[dict] = []
-
-
-@pytest.fixture(autouse=True)
-def _fake_kv():
-    store = FakeKvStore()
-    todo_mod._kv_store = store
-    scratchpad_mod._kv_store = store
-    yield store
-    todo_mod._kv_store = None
-    scratchpad_mod._kv_store = None
-
-
-_REPLAN_ANALYSIS = (
-    "TASK: revised task\n"
-    "GOAL: revised goal\n"
-    "MODE: act\n"
-    "UNKNOWNS: NONE\nRESOURCES: NONE\nKNOWLEDGE SOURCE ASSESSMENT: NONE\n"
-    "KNOWLEDGE CAPTURE: NONE\nCOMPLEXITY: simple\nSUBTASKS:\n1. New step\n"
-    "REFLECT: no\nCOMMITMENTS: none"
-)
-_REPLAN_PLAN = (
-    "## Plan\n\n"
-    "**Task:** revised task\n**Goal:** revised goal\n\n"
-    "**Steps:**\n"
-    "1. New step one → TOOL: filesystem\n"
-    "2. New step two → SELF\n\n"
-    "**Parallel:** NONE\n**Risks:** NONE"
-)
-
-
-def _make_runner(store, responses, session=None, profile=None):
-    session = session or FakeSession()
-    profile = profile or make_profile("developer", planning_phase2_enabled=False)
-    llm = RecordingLLM(responses)
-    engine = PlanningEngine(FakeContextBuilder())
-    runner = ReplanRunner(engine, session, profile, llm, mem=None, tool_schemas=[])
-    return runner, session, llm, engine, profile
-
-
-# ── ReplanTool schema ───────────────────────────────────────────────────────
-
-
-class TestReplanToolSchema:
-    def test_name_and_required_params(self):
-        t = ReplanTool()
-        assert t.name == "replan"
-        assert t.parameters["required"] == ["reason"]
-        assert "reason" in t.parameters["properties"]
-        assert "updated_goal" in t.parameters["properties"]
-
-    async def test_missing_reason_returns_error(self):
-        t = ReplanTool()
-        result = await t.execute({}, ToolContext())
-        assert result.success is False
-        assert "reason is required" in (result.error or "")
-
-    async def test_no_runner_returns_error(self):
-        # No current_replan_runner set in this context.
-        t = ReplanTool()
-        result = await t.execute({"reason": "plan is stale"}, ToolContext())
-        assert result.success is False
-        assert "not available" in (result.error or "")
-
-
-# ── ReplanRunner + PlanningEngine is_replan integration ──────────────────────
-
-
-class TestReplanRunner:
-    async def test_replan_runs_planner_with_is_replan_and_returns_plan(self):
-        runner, session, llm, engine, profile = _make_runner(
-            None, [_REPLAN_ANALYSIS, _REPLAN_PLAN]
-        )
-
-        plan = await runner.replan("config is TOML not JSON", None)
-
-        assert plan == _REPLAN_PLAN
-        # Phase 1 + Phase 3 ran (phase2 disabled).
-        assert len(llm.calls) == 2
-        # Phase 1 prompt carries the [RE-PLAN] framing and the reason.
-        phase1_prompt = llm.calls[0][0].content
-        assert "[RE-PLAN]" in phase1_prompt
-        assert "config is TOML not JSON" in phase1_prompt
-        # DIRECT shortcut suppressed for re-plan.
-        assert "CRITICAL DIRECT shortcut" not in phase1_prompt
-        # Re-plan opening line used instead of the fresh-request one.
-        assert "Re-plan based on the [RE-PLAN] context" in phase1_prompt
-        assert "Read the user's latest request." not in phase1_prompt
-
-    async def test_replan_context_includes_todo_findings_errors(self):
-        # Seed a todo + scratchpad sections the runner must pack into context.
-        # ContextVars (current_session_id / current_user_id) must be set for the
-        # whole flow — render_todo_lines reads user_id from current_user_id, and
-        # planning.run's set_tasks scopes by current_session_id. In production the
-        # WS handler / run_stream set these; here we set them explicitly.
-        sid_tok = current_session_id.set("sess_replan")
-        uid_tok = current_user_id.set("u1")
-        try:
-            from navi.tools.todo import set_tasks
-            await set_tasks("sess_replan", ["Old step one", "Old step two"])
-            await scratchpad_mod._kv_store.set("u1", "sess_replan", "scratchpad", "findings", "config parser expects TOML")
-            await scratchpad_mod._kv_store.set("u1", "sess_replan", "scratchpad", "errors", "JSON parse failed at line 3")
-
-            runner, session, llm, engine, profile = _make_runner(
-                None, [_REPLAN_ANALYSIS, _REPLAN_PLAN], session=FakeSession(sid="sess_replan", user_id="u1")
-            )
-
-            plan = await runner.replan("config is TOML not JSON", "parse config correctly")
-
-            assert plan == _REPLAN_PLAN
-            phase1_prompt = llm.calls[0][0].content
-            # The packed context surfaces all four pieces.
-            assert "Reason for re-plan: config is TOML not JSON" in phase1_prompt
-            assert "Updated goal: parse config correctly" in phase1_prompt
-            assert "Old step one" in phase1_prompt
-            assert "config parser expects TOML" in phase1_prompt
-            assert "JSON parse failed at line 3" in phase1_prompt
-        finally:
-            current_user_id.reset(uid_tok)
-            current_session_id.reset(sid_tok)
-
-    async def test_replan_replaces_todo_with_new_steps(self):
-        from navi.tools.todo import _load_tasks, set_tasks
-
-        sid_tok = current_session_id.set("sess_replan")
-        uid_tok = current_user_id.set("u1")
-        try:
-            await set_tasks("sess_replan", ["Old step one", "Old step two"])
-            assert [t.text for t in await _load_tasks("sess_replan")] == ["Old step one", "Old step two"]
-
-            runner, session, llm, engine, profile = _make_runner(
-                None, [_REPLAN_ANALYSIS, _REPLAN_PLAN], session=FakeSession(sid="sess_replan")
-            )
-
-            plan = await runner.replan("stale plan", None)
-
-            assert plan == _REPLAN_PLAN
-            # planning.run's set_tasks replaced the todo with the new plan steps.
-            # _parse_plan_steps keeps the full step line including the executor tag.
-            tasks = await _load_tasks("sess_replan")
-            assert [t.text for t in tasks] == [
-                "New step one → TOOL: filesystem",
-                "New step two → SELF",
-            ]
-        finally:
-            current_user_id.reset(uid_tok)
-            current_session_id.reset(sid_tok)
-
-    async def test_replan_logs_planning_debug_data_to_session(self):
-        runner, session, llm, engine, profile = _make_runner(
-            None, [_REPLAN_ANALYSIS, _REPLAN_PLAN]
-        )
-
-        await runner.replan("stale plan", None)
-
-        # PlanningDebugData log captured (is_subagent=False for replan).
-        assert len(session.planning_logs) == 1
-        assert session.planning_logs[0]["result"] == "plan"
-
-    async def test_replan_appends_plan_and_prompt_to_session_context(self):
-        runner, session, llm, engine, profile = _make_runner(
-            None, [_REPLAN_ANALYSIS, _REPLAN_PLAN]
-        )
-
-        await runner.replan("stale plan", None)
-
-        # New plan assistant message + execute prompt appended.
-        plan_msgs = [m for m in session.context if m.role == "assistant"]
-        assert any(_REPLAN_PLAN in (m.content or "") for m in plan_msgs)
-        prompt_msgs = [m for m in session.context if m.role == "user" and "Plan is ready" in (m.content or "")]
-        assert len(prompt_msgs) == 1
-
-    async def test_replan_returns_none_on_planner_failure(self):
-        # First LLM call raises → runner catches and returns None.
-        class _BoomLLM:
-            async def complete(self, messages, **kwargs):
-                raise RuntimeError("llm down")
-
-        session = FakeSession()
-        profile = make_profile("developer", planning_phase2_enabled=False)
-        engine = PlanningEngine(FakeContextBuilder())
-        runner = ReplanRunner(engine, session, profile, _BoomLLM(), mem=None, tool_schemas=[])
-
-        plan = await runner.replan("stale plan", None)
-        assert plan is None
-
-
-# ── PlanningEngine is_replan prompt/observe-skip behavior ────────────────────
-
-
-class TestPlanningIsReplan:
-    async def test_is_replan_suppresses_observe_skip(self):
-        """An observe-classified analysis must NOT skip Phase 3 when is_replan —
-        a stale plan always needs a new plan."""
-        profile = make_profile(
-            "developer",
-            planning_phase2_enabled=False,
-            observe_skips_plan_enabled=True,
-        )
-        observe_analysis = (
-            "TASK: look at X\nGOAL: report X\nMODE: observe\n"
-            "UNKNOWNS: NONE\nRESOURCES: NONE\nKNOWLEDGE SOURCE ASSESSMENT: NONE\n"
-            "KNOWLEDGE CAPTURE: NONE\nCOMPLEXITY: simple\nSUBTASKS:\n1. List\n"
-            "REFLECT: no\nCOMMITMENTS: none"
-        )
-        llm = RecordingLLM([observe_analysis, _REPLAN_PLAN])
-        engine = PlanningEngine(FakeContextBuilder())
-        context = [Message(role="user", content="look at X")]
-
-        events = []
-        async for ev in engine.run(
-            context, profile, llm, mem=None, tool_schemas=[],
-            force_plan=True, is_replan=True, replan_context="Reason for re-plan: stale",
-        ):
-            events.append(ev)
-
-        # Observe-skip suppressed → Phase 3 ran, plan produced.
-        assert len(llm.calls) == 2
-        assert any(isinstance(e, PlanReady) for e in events)
-
-    async def test_replan_context_omitted_does_not_inject_block(self):
-        """is_replan without replan_context must not inject an empty [RE-PLAN] block."""
-        profile = make_profile("developer", planning_phase2_enabled=False)
-        llm = RecordingLLM(["DIRECT"])  # force_plan+is_replan suppress DIRECT → needs full analysis, but one call is enough to inspect prompt
-        engine = PlanningEngine(FakeContextBuilder())
-        context = [Message(role="user", content="hi")]
-
-        # is_replan=True but replan_context=None
-        async for _ev in engine.run(
-            context, profile, llm, mem=None, tool_schemas=[],
-            force_plan=True, is_replan=True, replan_context=None,
-        ):
-            pass
-
-        phase1_prompt = llm.calls[0][0].content
-        assert "[RE-PLAN]" not in phase1_prompt
\ No newline at end of file