"""Planning pipeline — extracted from agent.py.
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
import re
from datetime import datetime, timezone
from typing import AsyncGenerator
import structlog
from navi.config import settings
from navi.llm.base import Message
from navi.tools._internal.base import current_stop_event
from .events import (
AIHelperTokensUsed,
PlanningDebugData,
PlanningStatus,
PlanReady,
)
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.
Returns ``(milestone, text)`` per step. The milestone is the single
uppercase letter in brackets at the start of a step line —
``1. [A] description → TOOL: x`` → ``("A", "description → TOOL: x")``.
Lines without a marker get ``""``. A bracketed tag that is NOT a single
letter (e.g. ``[TOOL]``) is not a milestone; such lines are skipped, as
before (legacy behaviour preserved for old plan shapes).
"""
m = re.search(r'\*\*Steps:\*\*\s*\n(.*?)(?=\n\s*\*\*[^*\n]+:\*\*|\Z)', plan_text, re.DOTALL)
if not m:
return []
steps_block = m.group(1)
steps: list[tuple[str, str]] = []
for sm in re.finditer(r'^\s*\d+[\.\)]\s*(?:\[([A-Z])\]\s*)?(.+)', steps_block, re.MULTILINE):
milestone = sm.group(1) # None when the marker is absent
step = sm.group(2).strip()
if not step:
continue
# Skip lines starting with a bracket that is NOT a single-letter
# milestone marker (e.g. "[TOOL] Do thing") — legacy behaviour.
if milestone is None and step.startswith("["):
continue
steps.append((milestone or "", step))
return steps
class PlanningEngine:
"""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,
context: list[Message],
profile,
llm,
mem: Message | None,
tool_schemas: list | None = None,
messages: list[Message] | None = None,
system_prompt_override: str | None = None,
is_subagent: bool = False,
is_replan: bool = False,
replan_context: str | None = None,
) -> AsyncGenerator:
"""Planning pipeline (async generator):
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:
tool_lines = []
for schema in tool_schemas:
fn = schema.function if hasattr(schema, "function") else schema.get("function", {})
name = fn.get("name", "")
desc = (fn.get("description") or "").split("\n")[0][:80]
tool_lines.append(f" - {name}: {desc}")
available_tools_block = (
"Available tools (use these exact names for TOOL: assignments):\n"
+ "\n".join(tool_lines)
+ "\n\n"
)
else:
available_tools_block = ""
_stop = current_stop_event.get()
_dbg: dict = {"timestamp": datetime.now(timezone.utc).isoformat(), "result": "plan", "phases": {}}
_base_sys = system_prompt_override if system_prompt_override is not None else self._ctx_builder.build_system_prompt(profile)
_mcp_msg = self._ctx_builder._mcp_context_msg(profile)
if _mcp_msg:
_base_sys = _base_sys + "\n\n---\n\n" + (_mcp_msg.content or "")
# 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)
if _has_replan_ctx:
_replan_block = (
"[RE-PLAN]\n\n"
"You are revising an existing plan mid-task — NOT starting fresh. The original plan became stale because of what you discovered during execution (not because a step failed — that is handled by [Adaptive re-plan]). "
"The current todo and scratchpad below reflect real progress: preserve already-completed work, replace only the remaining steps, and account for what changed. "
"Treat the reason and updated goal (if any) as the new direction.\n\n"
f"{replan_context}\n\n---\n\n"
)
# ── Phase 1: Task analysis ────────────────────────────────────────────
analysis: str = ""
if profile.planning_phase1_enabled:
yield PlanningStatus(phase=1, label="Working on it...", is_subagent=is_subagent)
phase1_system = Message(
role="system",
content=(
_base_sys
+ "\n\n---\n\n"
+ _replan_block
+ "[PLANNING — PHASE 1: ANALYSIS]\n\n"
+ (
"Read the user's latest request.\n\n"
if not is_replan else
(
"Re-plan based on the [RE-PLAN] context above plus the user's original request.\n\n"
if _has_replan_ctx else
"Re-plan the task: revise the existing plan based on the user's original request and current progress.\n\n"
)
)
+ (
""
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 "
"casual/social chat — output exactly: DIRECT\n"
"- If the user asks a simple question you can answer from your existing "
"knowledge without any tools (general facts, definitions, simple math) — "
"output exactly: DIRECT\n"
"- If the user gives a one-step instruction that needs no tool (e.g., 'stop', "
"'continue', 'ok') — output exactly: DIRECT\n"
"- Only build a full plan when tools, files, web, research, or multi-step "
"execution are actually needed.\n\n"
"Output must be literally the single word DIRECT (uppercase), nothing else. "
"No TASK, no GOAL, no STEPS, no analysis for trivial messages.\n\n"
)
+ available_tools_block
+ "Knowledge store rules (critical):\n"
"- `memory` is only for personal user facts and preferences.\n"
"- Connected MCP knowledge servers are authoritative only when the active profile exposes their tools.\n"
"- If the domain is infrastructure and gnexus-book tools are available, use gnexus-book as the primary source and persistence target.\n"
"- If no relevant MCP tools are available to this profile, do not plan to call unavailable MCP tools; use docs, files, command output, web, or ask the user after checking available sources.\n"
"- Never use memory for infrastructure inventory, service topology, network routes, proxy mappings, server roles, or service relationships.\n\n"
"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"
"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"
"- context sources: [which of connected MCP knowledge servers / memory / docs / web you will check and why]\n"
"KNOWLEDGE SOURCE ASSESSMENT:\n"
"- Domain: [user personal facts / infrastructure / project documentation / own capabilities / external web]\n"
"- Primary source: [connected knowledge servers / memory / docs / web / source files / command output]\n"
"- Fallback: [alternative source if primary is unavailable]\n"
"KNOWLEDGE CAPTURE:\n"
"- New information to save: [specific facts, conventions, or discoveries that should persist beyond this session]\n"
"- Target: [memory / connected knowledge server / docs / none — choose the best persistent store available to this profile]\n"
"- Duplication check: [which target-specific search/read/list step prevents duplicates]\n"
"- Rationale: [why this knowledge is stable and reusable]\n"
"COMPLEXITY: simple | medium | complex — choose based on ambiguity, number of files/systems, risk, and autonomy needed.\n"
"SUBTASKS:\n"
"1. [discrete unit of work]\n"
"2. [discrete unit of work]\n"
"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"
"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. "
"Hard maximum: 15 subtasks. Each must be concrete and actionable. "
"No execution yet — analysis only."
),
)
phase1_ctx: list[Message] = [phase1_system]
if mem:
phase1_ctx.append(mem)
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=False),
timeout=settings.llm_complete_timeout,
)
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"
if not is_subagent:
yield PlanningDebugData(log=_dbg)
return
except Exception:
log.warning("agent.planning_phase1_failed", exc_info=True)
_dbg["result"] = "phase1_error"
if not is_subagent:
yield PlanningDebugData(log=_dbg)
return
if r1.prompt_tokens or r1.completion_tokens:
yield AIHelperTokensUsed(
prompt_tokens=r1.prompt_tokens or 0,
completion_tokens=r1.completion_tokens or 0,
)
_dbg["phases"]["1"] = {
"output": analysis,
"prompt_tokens": r1.prompt_tokens or 0,
"completion_tokens": r1.completion_tokens or 0,
}
if not analysis or analysis.upper().startswith("DIRECT"):
log.debug("agent.planning_skipped", reason="direct")
_dbg["result"] = "direct"
if not is_subagent:
yield PlanningDebugData(log=_dbg)
return
if _stop and _stop.is_set():
log.debug("agent.planning_stopped", phase=1)
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 3: Execution plan ────────────────────────────────────────────
if not profile.planning_phase3_enabled:
log.debug("agent.planning_phase3_skipped")
_dbg["result"] = "phase1_only"
if not is_subagent:
yield PlanningDebugData(log=_dbg)
return
yield PlanningStatus(phase=3, label="Building execution plan...", is_subagent=is_subagent)
phase3_system = Message(
role="system",
content=(
_base_sys
+ "\n\n---\n\n"
"[PLANNING — PHASE 3: EXECUTION PLAN]\n\n"
"Task analysis:\n\n"
f"{analysis}\n\n"
"---\n\n"
+ 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"
"- AGENT: <profile_id> — a bounded subtask needing 3+ tool calls; one subagent handles this ONE step\n"
"- SELF — final user-facing synthesis or an internal judgment that needs no tool call\n\n"
"Executor classification rules (critical):\n"
"- If a step names or implies a tool action, mark it TOOL with that exact tool name, never SELF.\n"
"- Use TOOL for searching, reading, writing files, editing files, scratchpad notes, todo updates, "
"image inspection, rendering, compiling, publishing, sharing, terminal commands, API calls, and verification through tool output.\n"
"- Use SELF only for synthesis, choosing between already-known options, or explaining completed results.\n"
"- If a planned step cannot be completed without later calling a tool, it is not SELF.\n\n"
"Planning boundary (critical):\n"
"The plan is an execution contract, not an implementation. It may describe intent, order, executor, "
"inputs, expected outputs, and verification. It must NOT contain implementation code, source snippets, "
"function bodies, CSS/HTML/SQL/Python/JS, patches, exact file contents, or detailed command scripts. "
"Implementation belongs later in tool calls, file edits, terminal/code execution, or final artifacts. "
"A valid plan says what to change and how to verify it, not the code that performs the change.\n\n"
"Plan depth:\n"
"- simple: 1-3 steps\n"
"- medium: 5-9 steps\n"
"- complex or autonomous: 8-20 steps\n"
"- hard maximum: 20 steps\n"
"Use enough steps to make execution unambiguous. Do not compress unrelated actions into one step. "
"For complex/autonomous tasks, decompose large steps into smaller independent steps that each "
"produce a verifiable result on their own — more, finer steps beat fewer coarse ones for "
"execution tracking, because the model loses sight of progress inside a long step.\n\n"
"Knowledge source and persistence rules (critical):\n"
"- `memory` is only for personal user facts and preferences.\n"
"- Never store infrastructure inventory, service topology, network routes, proxy mappings, server roles, or service relationships in memory.\n"
"- Connected MCP knowledge servers are authoritative only when the active profile exposes their tools. Do not plan unavailable MCP tool calls.\n"
"- If the task domain is infrastructure and gnexus-book tools are available, include a gnexus-book read/search step before answering or changing anything.\n"
"- If execution may discover durable facts, include a dedicated knowledge persistence checkpoint before final synthesis.\n\n"
"For every non-trivial task, include steps for information gathering from connected knowledge servers/docs/files/tool schemas, "
"implementation or analysis, verification, knowledge persistence checkpoint, and final synthesis. "
"Choose the persistence target based on the fact's scope and the active profile's available tools: "
"memory tool for personal user facts and preferences only; connected knowledge servers for their own canonical domains "
"(for example gnexus-book for infrastructure inventory when its MCP tools are available); docs/ or manuals/ for project-wide documentation; "
"filesystem for standalone files. "
"Always search/read/list the selected target first to avoid duplicates. "
"The checkpoint can be SELF only when the task could not have discovered durable reusable facts; otherwise assign it to the exact persistence/search tool that will be needed later.\n\n"
"AGENT scoping rules (critical):\n"
"- Each AGENT step is one focused, independently verifiable unit of work.\n"
"- One AGENT step = one spawn_agent call later. Do NOT bundle multiple concerns.\n"
"- Comma test: if your step description lists things with 'and' or commas, "
"each item is a separate step.\n"
"- Good: 'Research X pricing from 3 sources' | 'Audit SSH config on host Y'\n"
"- Bad: 'Research everything and write the full report' (too broad — split it)\n\n"
"Required output format (use exactly this structure):\n\n"
"## Plan\n\n"
"**Task:** [reformulated task]\n"
"**Goal:** [success criterion]\n\n"
"**Milestones:**\n"
"A. [strategic phase]\n"
"B. [strategic phase]\n"
"C. [strategic phase]\n\n"
"**Steps:**\n"
"1. [A] [description] → TOOL: tool_name\n"
"2. [A] [description] → AGENT: profile_id\n"
"3. [B] [description] → AGENT: profile_id\n"
"4. [B] Knowledge persistence checkpoint: [search/read selected target; persist stable facts if discovered, or confirm none] → TOOL: tool_name OR SELF\n"
"5. [C] [final synthesis] → SELF\n"
"... continue to the needed depth, up to 20 steps\n\n"
"Each step line begins with its milestone letter in brackets, matching a Milestone above "
"(`N. [A] description → EXECUTOR`). The letter groups steps into strategic phases and is shown "
"as a heading in the todo tracker, so the model keeps its bearings inside a long run. If a "
"step genuinely does not fit any milestone, omit the marker. The marker is a single uppercase "
"letter only — never `[TOOL]` or multi-word tags.\n\n"
"**Parallel:** [step numbers that can run simultaneously, or NONE]\n"
"**Risks:** [unknowns to watch for, or NONE]\n\n"
"Reject vague steps such as 'research and implement everything', 'fix all issues', "
"or 'analyze project and make changes'. Split them into concrete, verifiable units. "
"Do not write prose. Do not start executing. Plan only."
),
)
phase3_ctx: list[Message] = [phase3_system]
if mem:
phase3_ctx.append(mem)
user_msgs = [m for m in context if m.role == "user"]
if user_msgs:
phase3_ctx.append(user_msgs[-1])
try:
r2 = await asyncio.wait_for(
llm.complete(phase3_ctx, tools=None, temperature=0.3, model=profile.model, think=False),
timeout=settings.llm_complete_timeout,
)
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"
if not is_subagent:
yield PlanningDebugData(log=_dbg)
return
except Exception:
log.warning("agent.planning_phase3_failed", exc_info=True)
_dbg["result"] = "phase3_error"
if not is_subagent:
yield PlanningDebugData(log=_dbg)
return
if r2.prompt_tokens or r2.completion_tokens:
yield AIHelperTokensUsed(
prompt_tokens=r2.prompt_tokens or 0,
completion_tokens=r2.completion_tokens or 0,
)
_dbg["phases"]["3"] = {
"output": plan_text,
"prompt_tokens": r2.prompt_tokens or 0,
"completion_tokens": r2.completion_tokens or 0,
}
if not plan_text:
_dbg["result"] = "empty_plan"
if not is_subagent:
yield PlanningDebugData(log=_dbg)
return
if not re.search(r"^\s*\d+[\.\)]", plan_text, re.MULTILINE):
log.warning("agent.planning_no_numbered_steps", plan_preview=plan_text[:200])
if _stop and _stop.is_set():
log.debug("agent.planning_stopped", phase=3)
return
if not re.search(r"(TOOL:|AGENT:|→\s*SELF)", plan_text):
log.warning("agent.planning_no_executors", hint="plan lacks TOOL/AGENT/SELF assignments")
plan_ctx_msg = Message(role="assistant", content=plan_text, is_display=False)
context.append(plan_ctx_msg)
if messages is not None:
messages.append(plan_ctx_msg)
messages.append(Message(role="assistant", content=plan_text, is_plan=True, is_context=False))
# 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:
try:
from navi.tools.todo import set_tasks
from navi.tools._internal.base import (
current_session_id as _sid_var,
current_todo_session_id as _todo_sid_var,
)
# Sub-agents set current_todo_session_id to their run id; use it so the
# sub-agent's plan populates its own todo row, not the parent session's.
_sid = _todo_sid_var.get() or _sid_var.get() or "__default__"
await set_tasks(_sid, _todo_steps)
log.debug("agent.todo_auto_populated", steps=len(_todo_steps), session=_sid)
except Exception:
log.warning("agent.todo_auto_populate_failed", exc_info=True)
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)
