"""
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}")
