diff --git a/navi/api/routes/admin.py b/navi/api/routes/admin.py index fcd034a..55818a6 100644 --- a/navi/api/routes/admin.py +++ b/navi/api/routes/admin.py @@ -342,6 +342,8 @@ "iteration_budget_enabled": profile.iteration_budget_enabled, "goal_anchoring_enabled": profile.goal_anchoring_enabled, "goal_anchoring_interval": profile.goal_anchoring_interval, + "scope_boundary_enabled": profile.scope_boundary_enabled, + "observe_skips_plan_enabled": profile.observe_skips_plan_enabled, "anti_stall_enabled": profile.anti_stall_enabled, "anti_stall_threshold": profile.anti_stall_threshold, "step_validation_enabled": profile.step_validation_enabled, diff --git a/navi/core/context_builder.py b/navi/core/context_builder.py index 3b0b8d8..6b99de2 100644 --- a/navi/core/context_builder.py +++ b/navi/core/context_builder.py @@ -174,6 +174,26 @@ ) return Message(role="system", content="\n".join(lines)) + def _scope_boundary_msg(self, profile: "AgentProfile") -> Message | None: + """Bounded-autonomy rule: keep the agent within the requested scope. + + Returns None when the profile has not opted in (legacy free-flight behavior). + """ + if not profile.scope_boundary_enabled: + return None + return Message( + role="system", + content=( + "[Scope boundary]\n" + "Act strictly within the user's literally requested scope. Do NOT expand to " + "sibling or parent directories/projects beyond what was asked, and do NOT start " + "executing discovered TODO lists, roadmaps, milestones, or backlog docs unless " + "the user explicitly asks you to. If you encounter such docs while exploring, " + "report what you found — do not act on it. Autonomy means finishing the requested " + "task well, not enlarging it." + ), + ) + def _security_policy_msg(self) -> Message | None: """Build a dynamic security policy system message based on user role.""" from navi.tools._internal.base import ( @@ -323,6 +343,11 @@ if policy: result.append(policy) + # Inject bounded-autonomy scope boundary when the profile opts in + boundary = self._scope_boundary_msg(profile) + if boundary: + result.append(boundary) + # Inject MCP server instructions into context mcp_msg = self._mcp_context_msg(profile) if mcp_msg: diff --git a/navi/core/planning.py b/navi/core/planning.py index 784d518..b37aa6b 100644 --- a/navi/core/planning.py +++ b/navi/core/planning.py @@ -126,6 +126,11 @@ "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" @@ -202,6 +207,19 @@ if _stop and _stop.is_set(): 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: + log.debug("agent.planning_observe_skip") + _dbg["result"] = "observe_skip" + if not is_subagent: + yield PlanningDebugData(log=_dbg) + return else: log.debug("agent.planning_phase1_skipped") diff --git a/navi/profiles/base.py b/navi/profiles/base.py index dfead92..742b338 100644 --- a/navi/profiles/base.py +++ b/navi/profiles/base.py @@ -99,6 +99,19 @@ goal_anchoring_enabled: bool = True goal_anchoring_interval: int = 5 + # Bounded autonomy: inject a standing scope-boundary system message that keeps + # the agent within the user's literally requested scope and forbids acting on + # discovered TODO/roadmap/milestone docs without an explicit request. + # OFF = legacy "free flight" behavior (explore broadly, finish discovered work). + scope_boundary_enabled: bool = False + + # Observe vs act: when True, a Phase 1 analysis that classifies the request as + # MODE: observe (look/read/explain/inspect/list — no changes requested) skips + # Phase 2/3 — no multi-step execution plan, no auto-todo, no "execute step by + # step" prompt. The agent just gathers info with tools and answers. + # OFF = every non-trivial request gets a full execution plan. + observe_skips_plan_enabled: bool = False + # Detect when the model is looping without todo progress for # anti_stall_threshold iterations and inject a hard warning. anti_stall_enabled: bool = True diff --git a/navi/profiles/loader.py b/navi/profiles/loader.py index 4857438..b956ef8 100644 --- a/navi/profiles/loader.py +++ b/navi/profiles/loader.py @@ -100,6 +100,8 @@ iteration_budget_enabled=config.get("iteration_budget_enabled", True), goal_anchoring_enabled=config.get("goal_anchoring_enabled", True), goal_anchoring_interval=config.get("goal_anchoring_interval", 5), + scope_boundary_enabled=config.get("scope_boundary_enabled", False), + observe_skips_plan_enabled=config.get("observe_skips_plan_enabled", False), anti_stall_enabled=config.get("anti_stall_enabled", True), anti_stall_threshold=config.get("anti_stall_threshold", 8), step_validation_enabled=config.get("step_validation_enabled", False), @@ -152,6 +154,8 @@ "iteration_budget_enabled": profile.iteration_budget_enabled, "goal_anchoring_enabled": profile.goal_anchoring_enabled, "goal_anchoring_interval": profile.goal_anchoring_interval, + "scope_boundary_enabled": profile.scope_boundary_enabled, + "observe_skips_plan_enabled": profile.observe_skips_plan_enabled, "anti_stall_enabled": profile.anti_stall_enabled, "anti_stall_threshold": profile.anti_stall_threshold, "step_validation_enabled": profile.step_validation_enabled, diff --git a/navi/profiles/navi_code/config.json b/navi/profiles/navi_code/config.json index aaa797b..22896b6 100644 --- a/navi/profiles/navi_code/config.json +++ b/navi/profiles/navi_code/config.json @@ -22,6 +22,8 @@ "iteration_budget_enabled": true, "goal_anchoring_enabled": true, "goal_anchoring_interval": 5, + "scope_boundary_enabled": true, + "observe_skips_plan_enabled": true, "anti_stall_enabled": true, "anti_stall_threshold": 8, "step_validation_enabled": false, diff --git a/tests/unit/core/test_context_builder.py b/tests/unit/core/test_context_builder.py index 800491d..8eb406c 100644 --- a/tests/unit/core/test_context_builder.py +++ b/tests/unit/core/test_context_builder.py @@ -1,7 +1,5 @@ """Unit tests for ContextBuilder.""" -import pytest - from navi.core.context_builder import ContextBuilder from navi.llm.base import Message from tests.conftest_factory import make_profile, make_profile_registry @@ -77,6 +75,22 @@ result = builder.build(context, profile, mem=None) assert result[0].role == "system" + def test_scope_boundary_message_injected_when_enabled(self): + builder = ContextBuilder(profile_registry=make_profile_registry()) + profile = make_profile("test", scope_boundary_enabled=True) + context = [Message(role="user", content="hi")] + result = builder.build(context, profile, mem=None) + boundary = [m for m in result if m.role == "system" and "[Scope boundary]" in (m.content or "")] + assert len(boundary) == 1 + assert "Do NOT expand to sibling or parent" in boundary[0].content + + def test_scope_boundary_absent_when_disabled(self): + builder = ContextBuilder(profile_registry=make_profile_registry()) + profile = make_profile("test") # default scope_boundary_enabled=False + context = [Message(role="user", content="hi")] + result = builder.build(context, profile, mem=None) + assert not any("[Scope boundary]" in (m.content or "") for m in result) + def test_injects_memory(self): builder = ContextBuilder(profile_registry=make_profile_registry()) profile = make_profile("test") diff --git a/tests/unit/core/test_planning.py b/tests/unit/core/test_planning.py index a3ab60b..90f5165 100644 --- a/tests/unit/core/test_planning.py +++ b/tests/unit/core/test_planning.py @@ -1,8 +1,7 @@ """Unit tests for navi.core.planning.""" -import pytest - 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 @@ -173,3 +172,86 @@ assert plan_marker[0].is_display is True 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"]) + 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 "MODE: observe | act" in phase1_prompt + assert "reading files to answer is still observe" in phase1_prompt diff --git a/tests/unit/profiles/test_base.py b/tests/unit/profiles/test_base.py index a663999..c0de92d 100644 --- a/tests/unit/profiles/test_base.py +++ b/tests/unit/profiles/test_base.py @@ -1,8 +1,5 @@ """Unit tests for AgentProfile Pydantic model.""" -import pytest -from pydantic import ValidationError - from navi.profiles.base import AgentProfile @@ -80,3 +77,16 @@ custom_field="whatever", ) assert p.model_dump()["custom_field"] == "whatever" + + +class TestAutonomyFlags: + def test_new_flags_default_off(self): + p = AgentProfile( + id="test", + name="Test", + description="desc", + system_prompt="sys", + enabled_tools=[], + ) + assert p.scope_boundary_enabled is False + assert p.observe_skips_plan_enabled is False