diff --git a/navi/core/agent.py b/navi/core/agent.py index c69a575..6e52620 100644 --- a/navi/core/agent.py +++ b/navi/core/agent.py @@ -499,7 +499,11 @@ ctx_task = asyncio.create_task(self._ctx_builder._collect_context_injections(profile)) mem_facts_task = asyncio.create_task( self._ctx_builder._memory_facts_msg( - user_message, user_id=session.user_id, injected_ids=turn_ctx.injected_fact_ids + user_message, + user_id=session.user_id, + injected_ids=turn_ctx.injected_fact_ids, + scope_boundary_enabled=profile.scope_boundary_enabled, + session_cwd=(session.session_metadata or {}).get("cwd"), ) ) ctx_injections = await ctx_task diff --git a/navi/core/context_builder.py b/navi/core/context_builder.py index 6b99de2..0f549fe 100644 --- a/navi/core/context_builder.py +++ b/navi/core/context_builder.py @@ -3,6 +3,7 @@ Extracted from agent.py to reduce the Agent class surface area. """ +from pathlib import Path from typing import TYPE_CHECKING import structlog @@ -13,6 +14,29 @@ log = structlog.get_logger() + +def _fact_value_is_abs_path_outside(value: str, cwd: str) -> bool: + """True when a fact's value is an absolute filesystem path outside `cwd`. + + Used to keep context-dependent path facts (e.g. "project_root → /home/.../navi-1") + from being injected into sessions running in a different project, where they + would misdirect the agent. Relative values are left through (we cannot know + what they were recorded relative to). Non-path values always pass. + """ + v = str(value).strip() + if not v or not (v.startswith("/") or v.startswith("~")): + return False + try: + p = Path(v).expanduser().resolve() + base = Path(cwd).expanduser().resolve() + except Exception: + return True # unresolvable path — treat as out-of-scope, drop it + try: + p.relative_to(base) + return False # within the session cwd — keep + except ValueError: + return True # absolute path outside the cwd — drop + if TYPE_CHECKING: from navi.context_providers._loader import ContextProviderRegistry from navi.memory.store import MemoryStore @@ -100,6 +124,8 @@ user_message: str, user_id: str | None = None, injected_ids: set[str] | None = None, + scope_boundary_enabled: bool = False, + session_cwd: str | None = None, ) -> "Message | None": if self._memory is None: log.info("memory_facts_msg.skip", reason="no_memory_store") @@ -133,6 +159,21 @@ injected_ids.add(f["id"]) else: new_facts = facts + + # Bounded autonomy: drop facts whose value is an absolute path outside the + # session cwd (e.g. "project_root → /home/.../navi-1" while the user works + # in another project) so they cannot misdirect the agent. Facts are kept + # when working inside that path (then they are correct). Gated by the + # profile flag so free-flight stays reproducible by toggling it off. + if scope_boundary_enabled and session_cwd: + kept = [f for f in new_facts if not _fact_value_is_abs_path_outside(str(f.get("value", "")), session_cwd)] + dropped = len(new_facts) - len(kept) + if dropped: + log.info("memory_facts_msg.scope_filtered", dropped=dropped, session_cwd=session_cwd) + new_facts = kept + if not new_facts: + return None + lines = ["Known facts about the user:"] for f in new_facts: conf = f.get("confidence", 70) diff --git a/tests/unit/core/test_context_builder.py b/tests/unit/core/test_context_builder.py index 8eb406c..f583610 100644 --- a/tests/unit/core/test_context_builder.py +++ b/tests/unit/core/test_context_builder.py @@ -1,6 +1,8 @@ """Unit tests for ContextBuilder.""" -from navi.core.context_builder import ContextBuilder +import asyncio + +from navi.core.context_builder import ContextBuilder, _fact_value_is_abs_path_outside from navi.llm.base import Message from tests.conftest_factory import make_profile, make_profile_registry @@ -148,3 +150,108 @@ assert any("gnexus-book instructions" in (m.content or "") for m in result) assert mcp.calls == [{"gnexus-book"}] + + +class FakeMemoryStore: + """Minimal MemoryStore stub returning canned facts from search_facts.""" + + def __init__(self, facts): + self._facts = facts + + async def search_facts(self, query, user_id=None, limit=15): + return self._facts[:limit] + + +class TestFactPathScope: + def test_abs_path_outside_cwd(self): + assert _fact_value_is_abs_path_outside("/home/u/navi-1", "/home/u/other-proj") is True + + def test_abs_path_inside_cwd(self): + assert _fact_value_is_abs_path_outside("/home/u/other-proj/sub", "/home/u/other-proj") is False + + def test_cwd_itself(self): + assert _fact_value_is_abs_path_outside("/home/u/other-proj", "/home/u/other-proj") is False + + def test_non_path_value_kept(self): + assert _fact_value_is_abs_path_outside("Linux", "/home/u/other-proj") is False + + def test_relative_path_kept(self): + assert _fact_value_is_abs_path_outside("workspace/secrets/x.yaml", "/home/u/other-proj") is False + + def test_empty_value_kept(self): + assert _fact_value_is_abs_path_outside("", "/home/u/other-proj") is False + + +def _fact(key, value, fid="1"): + return {"id": fid, "category": "technical", "key": key, "value": value, "confidence": 80} + + +class TestMemoryFactsScopeFilter: + def _builder(self, facts): + return ContextBuilder(profile_registry=make_profile_registry(), memory_store=FakeMemoryStore(facts)) + + def test_filter_drops_abs_path_outside_cwd(self): + facts = [_fact("project_root", "/home/u/navi-1")] + builder = self._builder(facts) + msg = asyncio.run( + builder._memory_facts_msg( + "please look at the contents of this directory and report back what you find there", + scope_boundary_enabled=True, + session_cwd="/home/u/other-proj", + ) + ) + assert msg is None # the only fact was filtered out → nothing to inject + + def test_filter_keeps_abs_path_inside_cwd(self): + facts = [_fact("project_root", "/home/u/other-proj")] + builder = self._builder(facts) + msg = asyncio.run( + builder._memory_facts_msg( + "please look at the contents of this directory and report back what you find there", + scope_boundary_enabled=True, + session_cwd="/home/u/other-proj", + ) + ) + assert msg is not None + assert "project_root" in msg.content + assert "/home/u/other-proj" in msg.content + + def test_filter_keeps_non_path_facts(self): + facts = [_fact("os", "Linux"), _fact("project_root", "/home/u/navi-1")] + builder = self._builder(facts) + msg = asyncio.run( + builder._memory_facts_msg( + "please look at the contents of this directory and report back what you find there", + scope_boundary_enabled=True, + session_cwd="/home/u/other-proj", + ) + ) + assert msg is not None + assert "os" in msg.content and "Linux" in msg.content + assert "navi-1" not in msg.content # path fact dropped + + def test_filter_disabled_injects_path_outside_cwd(self): + facts = [_fact("project_root", "/home/u/navi-1")] + builder = self._builder(facts) + msg = asyncio.run( + builder._memory_facts_msg( + "please look at the contents of this directory and report back what you find there", + scope_boundary_enabled=False, + session_cwd="/home/u/other-proj", + ) + ) + assert msg is not None + assert "navi-1" in msg.content # free-flight: nothing filtered + + def test_no_cwd_no_filtering(self): + facts = [_fact("project_root", "/home/u/navi-1")] + builder = self._builder(facts) + msg = asyncio.run( + builder._memory_facts_msg( + "please look at the contents of this directory and report back what you find there", + scope_boundary_enabled=True, + session_cwd=None, + ) + ) + assert msg is not None + assert "navi-1" in msg.content # without cwd we cannot judge scope