"""Unit tests for navi.core.planning."""
from navi.core.planning import PlanningEngine, _parse_plan_steps
from navi.core.events import PlanReady
from navi.llm.base import LLMResponse, Message
from tests.conftest_factory import make_profile
class RecordingLLM:
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,
finish_reason="stop",
)
class FakeContextBuilder:
def build_system_prompt(self, profile):
return "base system prompt"
def _mcp_context_msg(self, profile=None):
if profile and profile.mcp_servers:
return Message(
role="system",
content="gnexus-book instructions with mcp_gnexus-book_search_docs",
)
return None
class TestParsePlanSteps:
def test_basic_numbered_list(self):
text = "**Steps:**\n1. First step\n2. Second step\n3. Third step"
assert _parse_plan_steps(text) == [("", "First step"), ("", "Second step"), ("", "Third step")]
def test_parenthesised_numbers(self):
text = "**Steps:**\n1) Step one\n2) Step two"
assert _parse_plan_steps(text) == [("", "Step one"), ("", "Step two")]
def test_ignores_bracket_prefixes(self):
text = "**Steps:**\n1. [TOOL] Do thing\n2. Normal step"
assert _parse_plan_steps(text) == [("", "Normal step")]
def test_steps_carry_milestone_marker(self):
"""A single uppercase letter in brackets at the start of a step line is
parsed as the milestone label; multi-word bracketed tags are not."""
text = (
"**Steps:**\n"
"1. [A] read config.py → TOOL: filesystem\n"
"2. [A] implement parser → AGENT: developer\n"
"3. [B] run tests → TOOL: terminal\n"
"4. [TOOL] skipped line → ignored\n"
)
assert _parse_plan_steps(text) == [
("A", "read config.py → TOOL: filesystem"),
("A", "implement parser → AGENT: developer"),
("B", "run tests → TOOL: terminal"),
]
def test_empty_steps_section(self):
text = "**Steps:**\n\n**Notes:** nothing"
assert _parse_plan_steps(text) == []
def test_no_steps_section(self):
text = "Some random text without steps"
assert _parse_plan_steps(text) == []
class TestPlanningPrompt:
async def test_planning_prompt_includes_profile_mcp_and_persistence_rules(self):
profile = make_profile(
"server_admin",
mcp_servers={"gnexus-book": ["read", "write"]},
)
llm = RecordingLLM([
"TASK: document infra\n"
"GOAL: docs updated\n"
"UNKNOWNS: NONE\n"
"RESOURCES:\n"
"- mcp_gnexus-book_search_docs: search docs\n"
"- context sources: gnexus-book\n"
"KNOWLEDGE SOURCE ASSESSMENT:\n"
"- Domain: infrastructure\n"
"- Primary source: connected knowledge servers\n"
"- Fallback: docs\n"
"KNOWLEDGE CAPTURE:\n"
"- New information to save: stable infra facts\n"
"- Target: connected knowledge server\n"
"- Duplication check: search target\n"
"- Rationale: reusable\n"
"COMPLEXITY: medium\n"
"SUBTASKS:\n"
"1. Search docs\n"
"2. Persist facts\n"
"COMMITMENTS: checkpoint",
"## Plan\n\n"
"**Task:** document infra\n"
"**Goal:** docs updated\n\n"
"**Milestones:**\nA. Inspect\nB. Persist\nC. Report\n\n"
"**Steps:**\n"
"1. Search gnexus-book → TOOL: mcp_gnexus-book_search_docs\n"
"2. Knowledge persistence checkpoint → TOOL: mcp_gnexus-book_propose_doc_change\n"
"3. Final synthesis → SELF\n\n"
"**Parallel:** NONE\n"
"**Risks:** NONE",
])
engine = PlanningEngine(FakeContextBuilder())
context = [Message(role="user", content="update infra docs")]
events = []
async for event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
events.append(event)
phase1_prompt = llm.calls[0][0].content
phase3_prompt = llm.calls[1][0].content
assert "gnexus-book instructions" in phase1_prompt
assert "memory` is only for personal user facts and preferences" in phase1_prompt
assert "Never use memory for infrastructure inventory" in phase1_prompt
assert "knowledge persistence checkpoint" in phase3_prompt
assert "Do not plan unavailable MCP tool calls" in phase3_prompt
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
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",
mcp_servers={},
)
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")]
async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
pass
phase1_prompt = llm.calls[0][0].content
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_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: 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())
context = [Message(role="user", content="hello")]
messages = []
async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[], messages=messages):
pass
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")]
assert len(plan_ctx) == 1
assert plan_ctx[0].is_context is True
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
class TestPhase1Prompt:
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")]
async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
pass
phase1_prompt = llm.calls[0][0].content
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:
"""Minimal in-memory KV store for the todo auto-populate isolation test."""
def __init__(self):
self._d: dict[tuple, str] = {}
async def get(self, user_id, session_id, scope, key):
return self._d.get((user_id or "", session_id, scope, key))
async def set(self, user_id, session_id, scope, key, value):
self._d[(user_id or "", session_id, scope, key)] = value
class TestTodoIsolation:
async def test_auto_todo_lands_in_todo_session_row(self, monkeypatch):
"""Planning auto-populates the todo into the row scoped by
current_todo_session_id (the sub-agent's run id), not the parent session id.
The parent's todo must stay empty so its goal-anchoring stays accurate."""
from navi.tools._internal.base import (
current_session_id,
current_todo_session_id,
)
from navi.tools import todo as todo_mod
kv = _MemKv()
monkeypatch.setattr(todo_mod, "_kv_store", kv)
profile = make_profile("developer")
llm = RecordingLLM(
[
"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")]
t_tok = current_todo_session_id.set("sub_run_xyz")
s_tok = current_session_id.set("parent_sess")
try:
async for _event in engine.run(context, profile, llm, mem=None, tool_schemas=[]):
pass
finally:
current_todo_session_id.reset(t_tok)
current_session_id.reset(s_tok)
# 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
