diff --git a/docs/architecture.md b/docs/architecture.md index ddd6deeb2..4d056e3c6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -146,6 +146,7 @@ Defaults: | Memory | `/memory/` | | Cron store | `/cron/jobs.json` | | WebUI/media/log runtime data | config directory subdirectories such as `webui/`, `media/`, and `logs/` | +| Resource path aliases | `/resources//` (best-effort, derived state) | The schema accepts both camelCase and snake_case keys, but saves config with camelCase aliases. @@ -167,6 +168,10 @@ and receive only capability-specific read access to built-in/agent skills and the exact agent history file. Keep those cross-root capabilities read-only and explicit; do not treat the entire agent workspace as an allowed root. +Resource path aliases are created outside the workspace and resolve to these +same canonical targets. Authorization must continue to follow the resolved +target; the alias root itself must never be treated as a blanket capability. + ## Memory and Sessions Session history is the near-term conversation replay. Memory is the longer-term workspace state. diff --git a/docs/concepts.md b/docs/concepts.md index 0e4cdcbfb..0a0111728 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -55,6 +55,35 @@ When no separate project is selected, one directory normally serves both roles. Selecting a project changes the working context for that chat; it does not create a second agent or relocate the configured agent workspace. +### Resource Path Aliases + +When an agent runtime starts, nanobot makes a best-effort filesystem view under +the active config directory: + +```text +/resources// +├── agent -> +├── media -> /media +└── package -> +``` + +`` is deterministic for the config, agent workspace, and installed +package paths. Separate workspaces or Python environments therefore receive +separate views instead of competing for a mutable `current` link. Project files +are not linked into this view; relative paths continue to resolve from the +effective project workspace. + +These links are convenient names, not a new permission boundary. Restricted +file access still checks the resolved target, and a shell sandbox may not expose +the aliases at all. Full-access prompts use the agent alias for profile, memory, +history, and custom-skill paths; restricted prompts expose only alias subtrees +that are already readable and retain canonical exact-file paths where required. +Nanobot keeps canonical paths in config and runtime state, continues to accept +real paths, and falls back to them when links are unavailable. Creating the view +never blocks startup and never replaces an existing unowned file or directory. +The `resources/` tree is derived state, so backup and indexing tools should skip +it or preserve its links instead of following them into their targets. + ## Config Format `config.json` accepts both camelCase and snake_case keys. The docs use camelCase because nanobot writes config back to disk with camelCase aliases, for example `apiKey`, `modelPresets`, `intervalS`, and `maxToolResultChars`. diff --git a/nanobot/agent/context.py b/nanobot/agent/context.py index 61abf32cc..f30875e2c 100644 --- a/nanobot/agent/context.py +++ b/nanobot/agent/context.py @@ -1,5 +1,7 @@ """Context builder for assembling agent prompts.""" +from __future__ import annotations + import base64 import mimetypes import platform @@ -7,12 +9,17 @@ from pathlib import Path from typing import Any, Mapping, Sequence from nanobot.agent.memory import MemoryStore -from nanobot.agent.skills import SkillsLoader +from nanobot.agent.skills import ( + ResourceViewMode, + SkillsLoader, + build_resource_aliases_section, +) from nanobot.agent.tools import image_generation as image_generation_tools from nanobot.agent.tools import mcp as mcp_tools from nanobot.agent.tools.registry import ToolRegistry from nanobot.apps.cli import utils as cli_app_utils from nanobot.bus.events import InboundMessage +from nanobot.resource_links import ResourceView from nanobot.runtime_context import ( RUNTIME_CONTEXT_END, RUNTIME_CONTEXT_MESSAGE_META, @@ -61,11 +68,23 @@ class ContextBuilder: _MAX_HISTORY_TOKENS = 8_000 # hard cap on recent history section size (tokens) _RUNTIME_CONTEXT_END = RUNTIME_CONTEXT_END - def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None): + def __init__( + self, + workspace: Path, + timezone: str | None = None, + disabled_skills: list[str] | None = None, + *, + resource_view: ResourceView | None = None, + ): self.workspace = workspace self.timezone = timezone - self.memory = MemoryStore(workspace) - self.skills = SkillsLoader(workspace, disabled_skills=set(disabled_skills) if disabled_skills else None) + self.resource_view = resource_view + self.memory = MemoryStore(workspace, resource_view=resource_view) + self.skills = SkillsLoader( + workspace, + disabled_skills=set(disabled_skills) if disabled_skills else None, + resource_view=resource_view, + ) def build_system_prompt( self, @@ -76,10 +95,24 @@ class ContextBuilder: include_memory_recent_history: bool = True, session_key: str | None = None, unified_session: bool = False, + resource_view_mode: ResourceViewMode | None = None, ) -> str: """Build the system prompt from identity, bootstrap files, memory, and skills.""" root = workspace or self.workspace - parts = [self._get_identity(channel=channel, workspace=root)] + parts = [ + self._get_identity( + channel=channel, + workspace=root, + resource_view_mode=resource_view_mode, + ) + ] + + resource_aliases = build_resource_aliases_section( + self.resource_view, + resource_view_mode, + ) + if resource_aliases: + parts.append(resource_aliases) bootstrap = self._load_bootstrap_files(root) if bootstrap: @@ -120,11 +153,24 @@ class ContextBuilder: return "\n\n---\n\n".join(parts) - def _get_identity(self, channel: str | None = None, workspace: Path | None = None) -> str: + def _get_identity( + self, + channel: str | None = None, + workspace: Path | None = None, + *, + resource_view_mode: ResourceViewMode | None = None, + ) -> str: """Get the core identity section.""" root = workspace or self.workspace workspace_path = str(root.expanduser().resolve()) agent_workspace_path = str(self.workspace.expanduser().resolve()) + agent_resource_path = agent_workspace_path + if ( + resource_view_mode == "full" + and self.resource_view is not None + and self.resource_view.agent is not None + ): + agent_resource_path = str(self.resource_view.agent) system = platform.system() runtime = f"{'macOS' if system == 'Darwin' else system} {platform.machine()}, Python {platform.python_version()}" @@ -132,6 +178,7 @@ class ContextBuilder: "agent/identity.md", workspace_path=workspace_path, agent_workspace_path=agent_workspace_path, + agent_resource_path=agent_resource_path, runtime=runtime, platform_policy=render_template("agent/platform_policy.md", system=system), channel=channel or "", @@ -206,6 +253,7 @@ class ContextBuilder: include_memory_recent_history: bool = True, session_key: str | None = None, unified_session: bool = False, + resource_view_mode: ResourceViewMode | None = None, ) -> list[dict[str, Any]]: """Build the complete message list for an LLM call.""" root = workspace or self.workspace @@ -222,6 +270,7 @@ class ContextBuilder: include_memory_recent_history=include_memory_recent_history, session_key=session_key, unified_session=unified_session, + resource_view_mode=resource_view_mode, ), }, *history, diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 7e8e4a9ad..ec9915ae9 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -91,6 +91,7 @@ from nanobot.utils.runtime import ( ) if TYPE_CHECKING: + from nanobot.agent.skills import ResourceViewMode from nanobot.agent.tools.mcp import MCPConnection from nanobot.config.schema import ( ChannelsConfig, @@ -98,6 +99,8 @@ if TYPE_CHECKING: ToolsConfig, ) from nanobot.cron.service import CronService + from nanobot.resource_links import ResourceView + from nanobot.security.workspace_access import WorkspaceScope _T = TypeVar("_T") @@ -267,6 +270,7 @@ class AgentLoop: restart_mode: str = "auto", local_trigger_store: Any | None = None, idle_compact_check_interval_seconds: int = 0, + resource_view: ResourceView | None = None, ): from nanobot.config.schema import ToolsConfig @@ -338,6 +342,7 @@ class AgentLoop: self.cron_service = cron_service self.local_trigger_store = local_trigger_store self.restrict_to_workspace = restrict_to_workspace + self.resource_view = resource_view self.workspace_scopes = WorkspaceScopeResolver( default_workspace=workspace, default_restrict_to_workspace=restrict_to_workspace, @@ -347,7 +352,12 @@ class AgentLoop: self._extra_hooks: list[AgentHook] = hooks or [] self._hook_factories: list[AgentTurnHookFactory] = hook_factories or [] - self.context = ContextBuilder(workspace, timezone=timezone, disabled_skills=disabled_skills) + self.context = ContextBuilder( + workspace, + timezone=timezone, + disabled_skills=disabled_skills, + resource_view=resource_view, + ) self.sessions = session_manager or SessionManager(workspace) self.sessions.set_file_cap_archiver(self.context.memory.raw_archive) self.tools = ToolRegistry() @@ -367,6 +377,7 @@ class AgentLoop: max_concurrent_subagents=max_concurrent_subagents, fail_on_tool_error=fail_on_tool_error, llm_wall_timeout_for_session=lambda sk: runner_wall_llm_timeout_s(self.sessions, sk), + resource_view=resource_view, ) self._unified_session = unified_session self._running = False @@ -686,8 +697,20 @@ class AgentLoop: include_memory_recent_history=not ctx.ephemeral, session_key=ctx.session.key, unified_session=self._unified_session, + resource_view_mode=self._resource_view_mode_for_scope(scope), ) + def _resource_view_mode_for_scope( + self, + scope: WorkspaceScope, + ) -> ResourceViewMode | None: + """Return the alias visibility supported by this turn's tool boundary.""" + if self.resource_view is None: + return None + if scope.restrict_to_workspace or bool(self.exec_config.sandbox): + return "restricted" + return "full" + def _request_context_for_turn(self, ctx: TurnContext) -> RequestContext: assert ctx.session is not None scope = self.workspace_scopes.for_turn( diff --git a/nanobot/agent/memory.py b/nanobot/agent/memory.py index 3e92ebb7e..54271652f 100644 --- a/nanobot/agent/memory.py +++ b/nanobot/agent/memory.py @@ -15,6 +15,7 @@ from typing import TYPE_CHECKING, Any, Callable, Iterator from loguru import logger +from nanobot.resource_links import ResourceView from nanobot.runtime_context import public_history_messages from nanobot.session.manager import Session, SessionManager from nanobot.utils.gitstore import GitStore @@ -83,9 +84,16 @@ class MemoryStore: r"^\[\d{4}-\d{2}-\d{2}[^\]]*\]\s+[A-Z][A-Z0-9_]*(?:\s+\[tools:\s*[^\]]+\])?:" ) - def __init__(self, workspace: Path, max_history_entries: int = _DEFAULT_MAX_HISTORY): + def __init__( + self, + workspace: Path, + max_history_entries: int = _DEFAULT_MAX_HISTORY, + *, + resource_view: ResourceView | None = None, + ): self.workspace = workspace self.max_history_entries = max_history_entries + self.resource_view = resource_view self.memory_dir = ensure_dir(workspace / "memory") self.memory_file = self.memory_dir / "MEMORY.md" self.history_file = self.memory_dir / "history.jsonl" @@ -547,13 +555,18 @@ class MemoryStore: return has_workspace_prompt_override(self.dream_prompt_file) @staticmethod - def default_dream_prompt() -> str: + def default_dream_prompt(resource_view: ResourceView | None = None) -> str: from nanobot.agent.skills import BUILTIN_SKILLS_DIR + skill_creator_path = BUILTIN_SKILLS_DIR / "skill-creator" / "SKILL.md" + if resource_view is not None and resource_view.package is not None: + skill_creator_path = ( + resource_view.package / "skills" / "skill-creator" / "SKILL.md" + ) return render_template( "agent/dream.md", strip=True, - skill_creator_path=str(BUILTIN_SKILLS_DIR / "skill-creator" / "SKILL.md"), + skill_creator_path=str(skill_creator_path), ) def _dream_template(self) -> str: @@ -570,7 +583,7 @@ class MemoryStore: WORKSPACE_PROMPT_MAX_CHARS, original_chars, ) return text - return self.default_dream_prompt() + return self.default_dream_prompt(self.resource_view) def build_dream_prompt(self, *, max_entries: int = 20) -> tuple[str, int] | None: """Build the Dream prompt with unprocessed history context. diff --git a/nanobot/agent/skills.py b/nanobot/agent/skills.py index 1e748f04b..89d8c161b 100644 --- a/nanobot/agent/skills.py +++ b/nanobot/agent/skills.py @@ -1,16 +1,24 @@ """Skills loader for agent capabilities.""" +from __future__ import annotations + import json import os import re import shutil from pathlib import Path +from typing import Literal, TypeAlias import yaml +from nanobot.resource_links import ResourceView +from nanobot.utils.prompt_templates import render_template + # Default builtin skills directory (relative to this file) BUILTIN_SKILLS_DIR = Path(__file__).parent.parent / "skills" +ResourceViewMode: TypeAlias = Literal["full", "restricted"] + # Opening ---, YAML body (group 1), closing --- on its own line; supports CRLF. _STRIP_SKILL_FRONTMATTER = re.compile( r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", @@ -18,6 +26,39 @@ _STRIP_SKILL_FRONTMATTER = re.compile( ) +def build_resource_aliases_section( + resource_view: ResourceView | None, + mode: ResourceViewMode | None, +) -> str: + """Render healthy resource aliases without changing their access policy.""" + if resource_view is None or mode is None: + return "" + + aliases: list[tuple[str, str]] = [] + if mode == "full": + if resource_view.agent is not None: + aliases.append(("Agent workspace", str(resource_view.agent))) + if resource_view.media is not None: + aliases.append(("Media", str(resource_view.media))) + if resource_view.package is not None: + aliases.append(("Nanobot package", str(resource_view.package))) + else: + if resource_view.agent is not None: + aliases.append(("Custom skills", str(resource_view.agent / "skills"))) + if resource_view.media is not None: + aliases.append(("Media", str(resource_view.media))) + if resource_view.package is not None: + aliases.append(("Built-in skills", str(resource_view.package / "skills"))) + + if not aliases: + return "" + return render_template( + "agent/resource_aliases.md", + strip=True, + aliases=aliases, + ) + + class SkillsLoader: """ Loader for agent skills. @@ -26,11 +67,19 @@ class SkillsLoader: specific tools or perform certain tasks. """ - def __init__(self, workspace: Path, builtin_skills_dir: Path | None = None, disabled_skills: set[str] | None = None): + def __init__( + self, + workspace: Path, + builtin_skills_dir: Path | None = None, + disabled_skills: set[str] | None = None, + *, + resource_view: ResourceView | None = None, + ): self.workspace = workspace self.workspace_skills = workspace / "skills" self.builtin_skills = builtin_skills_dir or BUILTIN_SKILLS_DIR self.disabled_skills = disabled_skills or set() + self.resource_view = resource_view def _skill_entries_from_dir(self, base: Path, source: str, *, skip_names: set[str] | None = None) -> list[dict[str, str]]: if not base.exists(): @@ -125,12 +174,32 @@ class SkillsLoader: if not all_skills: return "" + workspace_alias_root = ( + self.resource_view.agent / "skills" + if self.resource_view is not None and self.resource_view.agent is not None + else None + ) + builtin_alias_root = ( + self.resource_view.package / "skills" + if self.resource_view is not None and self.resource_view.package is not None + else None + ) sections: list[str] = [] groups = ( - ("Workspace skills", "workspace", self.workspace_skills), - ("Built-in skills", "builtin", self.builtin_skills), + ( + "Workspace skills", + "workspace", + self.workspace_skills, + workspace_alias_root, + ), + ( + "Built-in skills", + "builtin", + self.builtin_skills, + builtin_alias_root, + ), ) - for label, source, root in groups: + for label, source, root, alias_root in groups: entries = [ entry for entry in all_skills @@ -139,7 +208,8 @@ class SkillsLoader: if not entries: continue - lines = [f"### {label} (`{root.expanduser().resolve()}`)"] + display_root = alias_root or root.expanduser().resolve() + lines = [f"### {label} (`{display_root}`)"] for entry in entries: skill_name = entry["name"] meta = self._get_skill_meta(skill_name) diff --git a/nanobot/agent/subagent.py b/nanobot/agent/subagent.py index b4f4cece8..b8dede68d 100644 --- a/nanobot/agent/subagent.py +++ b/nanobot/agent/subagent.py @@ -1,5 +1,7 @@ """Subagent manager for background task execution.""" +from __future__ import annotations + import asyncio import json import time @@ -13,6 +15,11 @@ from loguru import logger from nanobot.agent.hook import AgentHook, AgentHookContext from nanobot.agent.runner import AgentRunner, AgentRunSpec +from nanobot.agent.skills import ( + ResourceViewMode, + SkillsLoader, + build_resource_aliases_section, +) from nanobot.agent.tools.base import ToolResult from nanobot.agent.tools.context import ( RequestContext, @@ -28,6 +35,7 @@ from nanobot.bus.events import InboundMessage from nanobot.bus.queue import MessageBus from nanobot.config.schema import AgentDefaults, ToolsConfig from nanobot.providers.base import LLMProvider +from nanobot.resource_links import ResourceView from nanobot.security.workspace_access import ( WorkspaceScope, bind_workspace_scope, @@ -97,6 +105,7 @@ class SubagentManager: max_concurrent_subagents: int | None = None, fail_on_tool_error: bool | None = None, llm_wall_timeout_for_session: Callable[[str | None], float | None] | None = None, + resource_view: ResourceView | None = None, ): if workspace is None: raise TypeError("SubagentManager.__init__() missing required argument: 'workspace'") @@ -147,6 +156,7 @@ class SubagentManager: self.runner = AgentRunner() self._exec_session_manager = ExecSessionManager() self._llm_wall_timeout_for_session = llm_wall_timeout_for_session + self.resource_view = resource_view self._running_tasks: dict[str, asyncio.Task[str]] = {} self._task_statuses: dict[str, SubagentStatus] = {} self._session_tasks: dict[str, set[str]] = {} # session_key -> {task_id, ...} @@ -366,7 +376,20 @@ class SubagentManager: cfg.restrict_to_workspace = workspace_scope.restrict_to_workspace # Construct from the agent workspace; the bound scope below supplies the project cwd. tools = self._build_tools(tools_config=cfg) - system_prompt = self._build_subagent_prompt(workspace=root) + scope_restricted = ( + workspace_scope.restrict_to_workspace + if workspace_scope is not None + else self.restrict_to_workspace + ) + resource_view_mode: ResourceViewMode = ( + "restricted" + if scope_restricted or bool(self.tools_config.exec.sandbox) + else "full" + ) + system_prompt = self._build_subagent_prompt( + workspace=root, + resource_view_mode=resource_view_mode, + ) messages: list[dict[str, Any]] = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": task}, @@ -516,22 +539,37 @@ class SubagentManager: lines.append(f"- {result.error}") return "\n".join(lines) or (result.error or "Error: subagent execution failed.") - def _build_subagent_prompt(self, workspace: Path | None = None) -> str: + def _build_subagent_prompt( + self, + workspace: Path | None = None, + *, + resource_view_mode: ResourceViewMode | None = None, + ) -> str: """Build a focused system prompt for the subagent.""" - from nanobot.agent.skills import SkillsLoader - agent_workspace = self.workspace.expanduser().resolve() project_workspace = workspace.expanduser().resolve() if workspace else agent_workspace + history_root = agent_workspace + if ( + resource_view_mode == "full" + and self.resource_view is not None + and self.resource_view.agent is not None + ): + history_root = self.resource_view.agent skills_summary = SkillsLoader( self.workspace, disabled_skills=self.disabled_skills, + resource_view=self.resource_view, ).build_skills_summary() return render_template( "agent/subagent_system.md", workspace=str(project_workspace), agent_workspace=str(agent_workspace), - history_log=str(agent_workspace / "memory" / "history.jsonl"), + history_log=str(history_root / "memory" / "history.jsonl"), skills_summary=skills_summary or "", + resource_aliases=build_resource_aliases_section( + self.resource_view, + resource_view_mode, + ), ) async def cancel_by_session(self, session_key: str) -> int: diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 3cc71c8f3..0ac37aa0f 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -9,7 +9,10 @@ import time from collections.abc import Callable, Iterable from contextlib import nullcontext, suppress from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from nanobot.resource_links import ResourceView # Force UTF-8 encoding for Windows console if sys.platform == "win32": @@ -820,6 +823,26 @@ def _load_runtime_config(config: str | None = None, workspace: str | None = None return loaded +def _prepare_resource_view(config: Config) -> "ResourceView | None": + """Best-effort creation of the path aliases used by an actual agent runtime.""" + from nanobot.config.loader import get_config_path + from nanobot.resource_links import ensure_resource_view + + config_path = get_config_path().expanduser().resolve(strict=False) + try: + view = ensure_resource_view( + data_dir=config_path.parent, + config_path=config_path, + agent_workspace=config.workspace_path, + ) + except Exception as exc: + logger.warning("Could not prepare the nanobot resource view: {}", exc) + return None + for warning in view.warnings: + logger.warning("Resource view: {}", warning) + return view + + def _read_trigger_cli_message(message: str | None) -> str: """Read a trigger message from an argument or stdin.""" if message and message.strip(): @@ -1346,6 +1369,7 @@ def serve( ) raise typer.Exit(1) sync_workspace_templates(runtime_config.workspace_path) + resource_view = _prepare_resource_view(runtime_config) bus = MessageBus() session_manager = SessionManager(runtime_config.workspace_path) try: @@ -1354,6 +1378,7 @@ def serve( session_manager=session_manager, image_generation_provider_configs=image_gen_provider_configs(runtime_config), hook_factories=[create_file_edit_activity_hook], + resource_view=resource_view, ) except ValueError as exc: console.print(f"[red]Error: {exc}[/red]") @@ -1695,6 +1720,7 @@ def _run_gateway( except ValueError as exc: console.print(f"[red]Error: {exc}[/red]") raise typer.Exit(1) from exc + resource_view = _prepare_resource_view(config) session_manager = SessionManager(config.workspace_path) # Self-heal the gateway state file with the current PID after any restart. @@ -1743,6 +1769,7 @@ def _run_gateway( hooks=[TokenUsageHook(timezone_name=config.agents.defaults.timezone)], local_trigger_store=trigger_store, hook_factories=[create_file_edit_activity_hook], + resource_view=resource_view, ) webui_turn_coordinator = WebuiTurnCoordinator( bus=bus, @@ -2270,6 +2297,7 @@ def agent( cron = CronService(cron_store_path) _set_nanobot_logs(logs) + resource_view = _prepare_resource_view(config) try: agent_loop = AgentLoop.from_config( @@ -2277,6 +2305,7 @@ def agent( cron_service=cron, image_generation_provider_configs=image_gen_provider_configs(config), hook_factories=[create_file_edit_activity_hook], + resource_view=resource_view, ) except ValueError as exc: console.print(f"[red]Error: {exc}[/red]") diff --git a/nanobot/nanobot.py b/nanobot/nanobot.py index 9db9ab53d..2b064b4a3 100644 --- a/nanobot/nanobot.py +++ b/nanobot/nanobot.py @@ -5,7 +5,9 @@ from __future__ import annotations import asyncio from collections.abc import AsyncIterator from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any + +from loguru import logger from nanobot.agent.hook import AgentHook, SDKCaptureHook from nanobot.agent.hooks import create_file_edit_activity_hook @@ -39,6 +41,9 @@ from nanobot.sdk.types import ( ) from nanobot.utils.llm_runtime import LLMRuntime +if TYPE_CHECKING: + from nanobot.resource_links import ResourceView + __all__ = [ "Nanobot", "RunResult", @@ -61,6 +66,28 @@ __all__ = [ ] +def _prepare_resource_view(config: Config, config_path: Path) -> ResourceView | None: + """Best-effort resource aliases scoped to this SDK instance's config.""" + from nanobot.resource_links import ensure_resource_view + + try: + # CLI entry points synchronize workspace templates before this step. + # The SDK has no equivalent bootstrap phase, so ensure the link target + # exists before preparing its alias. + config.workspace_path.mkdir(parents=True, exist_ok=True) + view = ensure_resource_view( + data_dir=config_path.parent, + config_path=config_path, + agent_workspace=config.workspace_path, + ) + except Exception as exc: + logger.warning("Could not prepare the nanobot resource view: {}", exc) + return None + for warning in view.warnings: + logger.warning("Resource view: {}", warning) + return view + + class Nanobot: """Programmatic facade for running the nanobot agent. @@ -96,7 +123,7 @@ class Nanobot: model: Override the instance default model. model_preset: Override the instance default model preset. """ - from nanobot.config.loader import load_config, resolve_config_env_vars + from nanobot.config.loader import get_config_path, load_config, resolve_config_env_vars ensure_single_model_selector(model=model, model_preset=model_preset) resolved: Path | None = None @@ -105,6 +132,11 @@ class Nanobot: if not resolved.exists(): raise FileNotFoundError(f"Config not found: {resolved}") + effective_config_path = ( + resolved + if resolved is not None + else get_config_path().expanduser().resolve(strict=False) + ) config: Config = resolve_config_env_vars(load_config(resolved)) if workspace is not None: config.agents.defaults.workspace = str( @@ -117,10 +149,12 @@ class Nanobot: elif model_preset is not None: config.agents.defaults.model_preset = model_preset + resource_view = _prepare_resource_view(config, effective_config_path) loop = AgentLoop.from_config( config, image_generation_provider_configs=image_gen_provider_configs(config), hook_factories=[create_file_edit_activity_hook], + resource_view=resource_view, ) return cls(loop, config=config) diff --git a/nanobot/resource_links.py b/nanobot/resource_links.py new file mode 100644 index 000000000..c039ace84 --- /dev/null +++ b/nanobot/resource_links.py @@ -0,0 +1,443 @@ +"""Stable filesystem aliases for resources exposed to the agent. + +The aliases in this module are a compatibility view, not a new source of +filesystem permissions. Callers should keep canonical paths for persistence +and authorization, and use a non-None alias only when presenting a shorter +path to the model. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import stat +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from filelock import FileLock, Timeout + +_LOCK_TIMEOUT_SECONDS = 2 +_JUNCTION_TIMEOUT_SECONDS = 2 +_NAMESPACE_MARKER = ".nanobot-resource-views.json" +_VIEW_MARKER = ".nanobot-resource-view.json" +_MARKER_VERSION = 1 + + +@dataclass(frozen=True, slots=True) +class ResourceView: + """The healthy aliases in one immutable resource view.""" + + root: Path | None = None + agent: Path | None = None + media: Path | None = None + package: Path | None = None + warnings: tuple[str, ...] = () + + +def ensure_resource_view( + *, + data_dir: Path, + config_path: Path, + agent_workspace: Path, + package_root: Path | None = None, +) -> ResourceView: + """Create, or validate, a stable resource view. + + Expected filesystem failures are deliberately non-fatal. A caller can + use each non-None alias and fall back to its canonical path for any alias + that could not be prepared. + """ + + warnings: list[str] = [] + try: + canonical_data_dir = _canonical(data_dir) + canonical_config_path = _canonical(config_path) + canonical_agent_workspace = _canonical(agent_workspace) + canonical_package_root = _canonical( + package_root if package_root is not None else Path(__file__).parent + ) + except (OSError, RuntimeError) as exc: + return ResourceView(warnings=(f"Could not resolve resource paths: {_error_text(exc)}",)) + + view_id = _resource_view_id( + config_path=canonical_config_path, + agent_workspace=canonical_agent_workspace, + package_root=canonical_package_root, + ) + namespace_root = canonical_data_dir / "resources" + view_root = namespace_root / view_id + media_root = canonical_data_dir / "media" + + for label, target in ( + ("agent", canonical_agent_workspace), + ("package", canonical_package_root), + ): + if _paths_overlap(target, view_root): + warnings.append( + f"Resource view overlaps the {label} target and would make recursive " + f"traversal unsafe: {view_root}" + ) + return ResourceView(warnings=tuple(warnings)) + + try: + canonical_data_dir.mkdir(parents=True, exist_ok=True) + if not canonical_data_dir.is_dir(): + warnings.append(f"Resource data directory is not a directory: {canonical_data_dir}") + return ResourceView(warnings=tuple(warnings)) + except OSError as exc: + warnings.append( + f"Could not prepare resource data directory {canonical_data_dir}: {_error_text(exc)}" + ) + return ResourceView(warnings=tuple(warnings)) + + lock_path = canonical_data_dir / ".nanobot-resource-links.lock" + try: + with FileLock(str(lock_path), timeout=_LOCK_TIMEOUT_SECONDS): + return _ensure_resource_view_locked( + namespace_root=namespace_root, + view_root=view_root, + view_id=view_id, + config_path=canonical_config_path, + agent_workspace=canonical_agent_workspace, + media_root=media_root, + package_root=canonical_package_root, + warnings=warnings, + ) + except Timeout: + warnings.append(f"Timed out waiting for resource view lock: {lock_path}") + except OSError as exc: + warnings.append(f"Could not lock resource view {lock_path}: {_error_text(exc)}") + + return ResourceView(warnings=tuple(warnings)) + + +def _ensure_resource_view_locked( + *, + namespace_root: Path, + view_root: Path, + view_id: str, + config_path: Path, + agent_workspace: Path, + media_root: Path, + package_root: Path, + warnings: list[str], +) -> ResourceView: + namespace_marker = { + "kind": "nanobot-resource-views", + "version": _MARKER_VERSION, + } + if not _ensure_owned_directory( + namespace_root, + marker_name=_NAMESPACE_MARKER, + marker_payload=namespace_marker, + label="resource namespace", + warnings=warnings, + ): + return ResourceView(warnings=tuple(warnings)) + + view_marker = { + "kind": "nanobot-resource-view", + "version": _MARKER_VERSION, + "view_id": view_id, + "config_path": _path_identity(config_path), + "targets": { + "agent": _path_identity(agent_workspace), + "media": _path_identity(media_root), + "package": _path_identity(package_root), + }, + } + if not _ensure_owned_directory( + view_root, + marker_name=_VIEW_MARKER, + marker_payload=view_marker, + label="resource view", + warnings=warnings, + ): + return ResourceView(warnings=tuple(warnings)) + + try: + media_root.mkdir(parents=True, exist_ok=True) + except OSError as exc: + warnings.append(f"Could not prepare media target {media_root}: {_error_text(exc)}") + + agent_alias = _ensure_alias( + view_root / "agent", + target=agent_workspace, + view_root=view_root, + label="agent", + warnings=warnings, + ) + media_alias = _ensure_alias( + view_root / "media", + target=media_root, + view_root=view_root, + label="media", + warnings=warnings, + ) + package_alias = _ensure_alias( + view_root / "package", + target=package_root, + view_root=view_root, + label="package", + warnings=warnings, + ) + return ResourceView( + root=view_root, + agent=agent_alias, + media=media_alias, + package=package_alias, + warnings=tuple(warnings), + ) + + +def _resource_view_id( + *, + config_path: Path, + agent_workspace: Path, + package_root: Path, +) -> str: + identities = ( + _path_identity(config_path), + _path_identity(agent_workspace), + _path_identity(package_root), + ) + digest = hashlib.sha256( + "\0".join(identities).encode("utf-8", errors="surrogatepass") + ).hexdigest() + return digest[:16] + + +def _canonical(path: Path) -> Path: + return Path(path).expanduser().resolve(strict=False) + + +def _path_identity(path: Path) -> str: + return os.path.normcase(os.path.normpath(str(path))) + + +def _ensure_owned_directory( + directory: Path, + *, + marker_name: str, + marker_payload: dict[str, Any], + label: str, + warnings: list[str], +) -> bool: + created = False + try: + if os.path.lexists(directory): + if _is_link_like(directory) or not directory.is_dir(): + warnings.append(f"Unmanaged {label} collision at {directory}") + return False + else: + directory.mkdir() + created = True + except OSError as exc: + warnings.append(f"Could not prepare {label} {directory}: {_error_text(exc)}") + return False + + marker_path = directory / marker_name + if not created: + actual = _read_marker(marker_path, label=label, warnings=warnings) + if actual is None: + return False + if actual != marker_payload: + warnings.append(f"Ownership marker does not match expected {label}: {marker_path}") + return False + return True + + try: + _write_marker(marker_path, marker_payload) + except OSError as exc: + warnings.append(f"Could not write {label} marker {marker_path}: {_error_text(exc)}") + # Only an empty directory can be removed here. Never recursively + # clean a path that another process may have populated. + try: + directory.rmdir() + except OSError: + pass + return False + return True + + +def _read_marker( + marker_path: Path, + *, + label: str, + warnings: list[str], +) -> dict[str, Any] | None: + try: + if not os.path.lexists(marker_path): + warnings.append(f"Unmanaged {label} at {marker_path.parent}: ownership marker missing") + return None + if _is_link_like(marker_path) or not stat.S_ISREG(marker_path.lstat().st_mode): + warnings.append(f"Invalid {label} ownership marker: {marker_path}") + return None + payload = json.loads(marker_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + warnings.append(f"Could not read {label} marker {marker_path}: {_error_text(exc)}") + return None + + if not isinstance(payload, dict): + warnings.append(f"Invalid {label} ownership marker: {marker_path}") + return None + return payload + + +def _write_marker(marker_path: Path, payload: dict[str, Any]) -> None: + serialized = json.dumps(payload, indent=2, sort_keys=True) + "\n" + with marker_path.open("x", encoding="utf-8", newline="\n") as marker_file: + marker_file.write(serialized) + marker_file.flush() + os.fsync(marker_file.fileno()) + + +def _ensure_alias( + alias: Path, + *, + target: Path, + view_root: Path, + label: str, + warnings: list[str], +) -> Path | None: + try: + if not target.is_dir(): + warnings.append(f"Resource target for {label} is not a directory: {target}") + return None + except OSError as exc: + warnings.append(f"Could not inspect resource target for {label} {target}: {_error_text(exc)}") + return None + + if _paths_overlap(target, view_root): + warnings.append( + f"Resource target for {label} overlaps its view and would create a cycle: {target}" + ) + return None + + try: + if os.path.lexists(alias): + if _is_directory_link(alias) and _link_points_to(alias, target): + return alias + warnings.append(f"Resource alias collision for {label} at {alias}") + return None + + _create_directory_link(alias, target) + if not _is_directory_link(alias) or not _link_points_to(alias, target): + warnings.append(f"Created resource alias for {label} could not be verified: {alias}") + _remove_created_link(alias, label=label, warnings=warnings) + return None + except OSError as exc: + warnings.append(f"Could not create resource alias for {label} at {alias}: {_error_text(exc)}") + return None + + return alias + + +def _paths_overlap(first: Path, second: Path) -> bool: + return first.is_relative_to(second) or second.is_relative_to(first) + + +def _is_link_like(path: Path) -> bool: + try: + if path.is_symlink(): + return True + attributes = getattr(path.lstat(), "st_file_attributes", 0) + reparse_point = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400) + return bool(attributes & reparse_point) + except OSError: + return False + + +def _is_directory_link(path: Path) -> bool: + if not _is_link_like(path): + return False + try: + return path.is_dir() + except OSError: + return False + + +def _link_points_to(alias: Path, target: Path) -> bool: + try: + resolved_alias = alias.resolve(strict=True) + resolved_target = target.resolve(strict=True) + except (OSError, RuntimeError): + return False + return _path_identity(resolved_alias) == _path_identity(resolved_target) + + +def _remove_created_link(alias: Path, *, label: str, warnings: list[str]) -> None: + """Remove only a link-like entry created during the current call.""" + + if not os.path.lexists(alias) or not _is_link_like(alias): + return + try: + alias.unlink() + return + except OSError: + # Directory junctions on Python 3.11 may require rmdir. os.rmdir on a + # reparse point removes the junction itself and does not traverse it. + try: + os.rmdir(alias) + return + except OSError as exc: + warnings.append( + f"Could not remove unverified resource alias for {label} at " + f"{alias}: {_error_text(exc)}" + ) + + +def _create_directory_link(alias: Path, target: Path) -> None: + try: + alias.symlink_to(target, target_is_directory=True) + return + except OSError: + if not _is_windows(): + raise + _create_windows_junction(alias, target) + + +def _is_windows() -> bool: + return os.name == "nt" + + +def _create_windows_junction(alias: Path, target: Path) -> None: + alias_text = str(alias) + target_text = str(target) + if any(character in alias_text + target_text for character in ('"', "\r", "\n")): + raise OSError("Path cannot be safely passed to the Windows junction command") + + # Keep user-controlled paths out of the command string. Expanding fixed, + # quoted environment variables also protects cmd metacharacters in paths. + command_env = os.environ.copy() + command_env["NANOBOT_RESOURCE_ALIAS"] = alias_text + command_env["NANOBOT_RESOURCE_TARGET"] = target_text + command = 'mklink /J "%NANOBOT_RESOURCE_ALIAS%" "%NANOBOT_RESOURCE_TARGET%"' + try: + completed = subprocess.run( + f"cmd.exe /d /v:off /c {command}", + capture_output=True, + text=True, + errors="replace", + env=command_env, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + timeout=_JUNCTION_TIMEOUT_SECONDS, + check=False, + ) + except subprocess.TimeoutExpired as exc: + raise OSError( + f"Timed out creating Windows junction after {_JUNCTION_TIMEOUT_SECONDS}s" + ) from exc + if completed.returncode == 0: + return + + details = (completed.stderr or completed.stdout or "").strip() + suffix = f": {details}" if details else "" + raise OSError(f"mklink /J failed with exit code {completed.returncode}{suffix}") + + +def _error_text(exc: BaseException) -> str: + return str(exc) or exc.__class__.__name__ diff --git a/nanobot/templates/agent/identity.md b/nanobot/templates/agent/identity.md index dfcfb1b60..e286008a0 100644 --- a/nanobot/templates/agent/identity.md +++ b/nanobot/templates/agent/identity.md @@ -1,15 +1,16 @@ ## Runtime {{ runtime }} +{% set resource_path = agent_resource_path | default(agent_workspace_path) %} ## Workspace Your current project workspace is at: {{ workspace_path }} {% if agent_workspace_path != workspace_path %} Nanobot's agent workspace is at: {{ agent_workspace_path }} {% endif %} -- Agent profile: {{ agent_workspace_path }}/SOUL.md and {{ agent_workspace_path }}/USER.md (automatically managed by Dream — do not edit directly) -- Long-term memory: {{ agent_workspace_path }}/memory/MEMORY.md (automatically managed by Dream — do not edit directly) -- History log: {{ agent_workspace_path }}/memory/history.jsonl (append-only JSONL; prefer built-in `grep` for search). -- Custom skills: {{ agent_workspace_path }}/skills/{% raw %}{skill-name}{% endraw %}/SKILL.md +- Agent profile: {{ resource_path }}/SOUL.md and {{ resource_path }}/USER.md (automatically managed by Dream — do not edit directly) +- Long-term memory: {{ resource_path }}/memory/MEMORY.md (automatically managed by Dream — do not edit directly) +- History log: {{ resource_path }}/memory/history.jsonl (append-only JSONL; prefer built-in `grep` for search). +- Custom skills: {{ resource_path }}/skills/{% raw %}{skill-name}{% endraw %}/SKILL.md {{ platform_policy }} {% if channel == 'telegram' or channel == 'qq' or channel == 'discord' %} diff --git a/nanobot/templates/agent/resource_aliases.md b/nanobot/templates/agent/resource_aliases.md new file mode 100644 index 000000000..21a7270ae --- /dev/null +++ b/nanobot/templates/agent/resource_aliases.md @@ -0,0 +1,8 @@ +## Resource Aliases + +These stable filesystem aliases are available: +{% for label, path in aliases %} +- {{ label }}: `{{ path }}` +{% endfor %} + +Aliases are alternative path names only; they do not grant additional file or shell permissions. A sandboxed shell may not expose an alias even when a file tool can use it. Continue to use paths relative to the current project workspace for project files. diff --git a/nanobot/templates/agent/subagent_system.md b/nanobot/templates/agent/subagent_system.md index b07b01cc5..fef6e75eb 100644 --- a/nanobot/templates/agent/subagent_system.md +++ b/nanobot/templates/agent/subagent_system.md @@ -11,6 +11,10 @@ Current project workspace: {{ workspace }} Nanobot's agent workspace: {{ agent_workspace }} {% endif %} History log: {{ history_log }} +{% if resource_aliases %} + +{{ resource_aliases }} +{% endif %} {% if skills_summary %} ## Skills diff --git a/tests/agent/test_context_builder.py b/tests/agent/test_context_builder.py index b40f6f9ea..b422de732 100644 --- a/tests/agent/test_context_builder.py +++ b/tests/agent/test_context_builder.py @@ -5,6 +5,7 @@ from pathlib import Path import pytest from nanobot.agent.context import ContextBuilder +from nanobot.resource_links import ResourceView from nanobot.runtime_context import RuntimeContextBlock # --------------------------------------------------------------------------- @@ -346,6 +347,65 @@ class TestBuildSystemPrompt: assert "## AGENTS.md" not in result assert "[Archived Context Summary]" not in result + def test_resource_aliases_are_absent_without_explicit_mode(self, tmp_path): + aliases = tmp_path / "resources" / "view" + resource_view = ResourceView( + root=aliases, + agent=aliases / "agent", + media=aliases / "media", + package=aliases / "package", + ) + + result = _builder(tmp_path, resource_view=resource_view).build_system_prompt() + + assert "## Resource Aliases" not in result + + def test_full_resource_aliases_show_roots_and_policy(self, tmp_path): + aliases = tmp_path / "resources" / "view" + resource_view = ResourceView( + root=aliases, + agent=aliases / "agent", + media=aliases / "media", + package=aliases / "package", + ) + + result = _builder(tmp_path, resource_view=resource_view).build_system_prompt( + resource_view_mode="full", + ) + + assert "## Resource Aliases" in result + assert f"Agent workspace: `{resource_view.agent}`" in result + assert f"Media: `{resource_view.media}`" in result + assert f"Nanobot package: `{resource_view.package}`" in result + assert f"Long-term memory: {resource_view.agent}/memory/MEMORY.md" in result + assert f"History log: {resource_view.agent}/memory/history.jsonl" in result + assert f"Custom skills: {resource_view.agent}/skills/" in result + assert "do not grant additional file or shell permissions" in result + assert "sandboxed shell may not expose an alias" in result + assert "paths relative to the current project workspace" in result + + def test_restricted_resource_aliases_only_show_allowed_subtrees(self, tmp_path): + aliases = tmp_path / "resources" / "view" + resource_view = ResourceView( + root=aliases, + agent=aliases / "agent", + media=aliases / "media", + package=aliases / "package", + ) + + result = _builder(tmp_path, resource_view=resource_view).build_system_prompt( + resource_view_mode="restricted", + ) + + assert f"Custom skills: `{resource_view.agent / 'skills'}`" in result + assert f"Media: `{resource_view.media}`" in result + assert f"Built-in skills: `{resource_view.package / 'skills'}`" in result + assert f"Agent workspace: `{resource_view.agent}`" not in result + assert f"Nanobot package: `{resource_view.package}`" not in result + canonical_workspace = tmp_path.resolve() + assert f"History log: {canonical_workspace}/memory/history.jsonl" in result + assert f"History log: {resource_view.agent}/memory/history.jsonl" not in result + # --------------------------------------------------------------------------- # build_messages @@ -369,6 +429,25 @@ class TestBuildMessages: assert messages[1]["role"] == "user" assert "hello" in str(messages[1]["content"]) + def test_resource_view_mode_is_forwarded_to_system_prompt(self, tmp_path): + aliases = tmp_path / "resources" / "view" + resource_view = ResourceView( + root=aliases, + agent=aliases / "agent", + media=aliases / "media", + package=aliases / "package", + ) + builder = _builder(tmp_path, resource_view=resource_view) + + messages = builder.build_messages( + [], + "hello", + resource_view_mode="restricted", + ) + + assert "## Resource Aliases" in messages[0]["content"] + assert f"Custom skills: `{resource_view.agent / 'skills'}`" in messages[0]["content"] + def test_public_builder_preserves_assistant_role_compatibility(self, tmp_path): from nanobot.agent import ContextBuilder as PublicContextBuilder diff --git a/tests/agent/test_dream.py b/tests/agent/test_dream.py index 5a04ac683..7724a5dcf 100644 --- a/tests/agent/test_dream.py +++ b/tests/agent/test_dream.py @@ -5,6 +5,7 @@ import pytest from nanobot.agent.memory import MemoryStore from nanobot.config.schema import ModelPresetConfig from nanobot.providers.base import LLMResponse +from nanobot.resource_links import ResourceView from nanobot.security.workspace_access import ( bind_workspace_scope, default_workspace_scope, @@ -62,6 +63,27 @@ class TestBuildDreamPrompt: prompt, _ = result assert "skill-creator" in prompt + def test_prompt_uses_package_alias_for_skill_creator(self, tmp_path): + aliases = tmp_path / "resources" / "view" + resource_view = ResourceView( + root=aliases, + agent=aliases / "agent", + media=aliases / "media", + package=aliases / "package", + ) + store = MemoryStore(tmp_path / "workspace", resource_view=resource_view) + store.append_history("test") + + result = store.build_dream_prompt() + + assert result is not None + prompt, _ = result + expected = resource_view.package / "skills" / "skill-creator" / "SKILL.md" + assert str(expected) in prompt + + def test_default_dream_prompt_class_call_remains_compatible(self): + assert "skill-creator" in MemoryStore.default_dream_prompt() + def test_prompt_embeds_current_memory_file_contents(self, store): """Dream must see the real current file contents (Tier 4) so it edits the files, not a stale mental model.""" diff --git a/tests/agent/test_loop_resource_view.py b/tests/agent/test_loop_resource_view.py new file mode 100644 index 000000000..da1537535 --- /dev/null +++ b/tests/agent/test_loop_resource_view.py @@ -0,0 +1,107 @@ +"""AgentLoop integration tests for the runtime resource view.""" + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from nanobot.agent.loop import AgentLoop, TurnKind +from nanobot.bus.queue import MessageBus +from nanobot.config.schema import ToolsConfig +from nanobot.resource_links import ResourceView +from nanobot.security.workspace_access import build_workspace_scope + + +def _provider() -> MagicMock: + provider = MagicMock() + provider.get_default_model.return_value = "test-model" + provider.generation = SimpleNamespace( + max_tokens=4096, + temperature=0.1, + reasoning_effort=None, + ) + return provider + + +def _loop( + tmp_path: Path, + *, + resource_view: ResourceView | None, + tools_config: ToolsConfig | None = None, +) -> tuple[AgentLoop, MagicMock, MagicMock]: + with ( + patch("nanobot.agent.loop.ContextBuilder") as context_builder, + patch("nanobot.agent.loop.SessionManager"), + patch("nanobot.agent.loop.SubagentManager") as subagent_manager, + patch.object(AgentLoop, "_register_default_tools"), + ): + loop = AgentLoop( + bus=MessageBus(), + provider=_provider(), + workspace=tmp_path, + tools_config=tools_config, + resource_view=resource_view, + ) + return loop, context_builder, subagent_manager + + +def test_loop_injects_resource_view_without_creating_one(tmp_path: Path) -> None: + view = ResourceView(root=tmp_path / "resources" / "view") + + loop, context_builder, subagent_manager = _loop( + tmp_path, + resource_view=view, + ) + + assert loop.resource_view is view + assert context_builder.call_args.kwargs["resource_view"] is view + assert subagent_manager.call_args.kwargs["resource_view"] is view + + +@pytest.mark.parametrize( + ("access_mode", "sandbox", "expected"), + [ + ("full", "", "full"), + ("restricted", "", "restricted"), + ("full", "bwrap", "restricted"), + ], +) +def test_initial_prompt_uses_effective_resource_view_mode( + tmp_path: Path, + access_mode: str, + sandbox: str, + expected: str, +) -> None: + tools_config = ToolsConfig() + tools_config.exec.sandbox = sandbox + view = ResourceView(root=tmp_path / "resources" / "view") + loop, _, _ = _loop( + tmp_path, + resource_view=view, + tools_config=tools_config, + ) + scope = build_workspace_scope(tmp_path, access_mode) + loop.workspace_scopes = SimpleNamespace(for_message=MagicMock(return_value=scope)) + loop.context.build_messages.return_value = [] + turn = SimpleNamespace( + session=SimpleNamespace(key="cli:test", metadata={}), + msg=SimpleNamespace(content="hello", media=None), + history=[], + kind=TurnKind.USER, + delivery=SimpleNamespace(route=SimpleNamespace(channel="cli")), + pending_summary=None, + runtime_context_blocks=[], + ephemeral=False, + ) + + loop._build_initial_messages(turn) + + assert loop.context.build_messages.call_args.kwargs["resource_view_mode"] == expected + + +def test_initial_prompt_keeps_legacy_mode_without_resource_view(tmp_path: Path) -> None: + loop, _, _ = _loop(tmp_path, resource_view=None) + scope = build_workspace_scope(tmp_path, "full") + + assert loop._resource_view_mode_for_scope(scope) is None diff --git a/tests/agent/test_skills_loader.py b/tests/agent/test_skills_loader.py index 5229efacf..49cd82388 100644 --- a/tests/agent/test_skills_loader.py +++ b/tests/agent/test_skills_loader.py @@ -8,6 +8,7 @@ from pathlib import Path import pytest from nanobot.agent.skills import SkillsLoader +from nanobot.resource_links import ResourceView def _write_skill( @@ -315,6 +316,41 @@ def test_build_skills_summary_groups_paths_by_root(tmp_path: Path) -> None: assert "`beta/SKILL.md`" in summary +def test_build_skills_summary_uses_alias_roots_but_keeps_canonical_entries( + tmp_path: Path, +) -> None: + workspace = tmp_path / "ws" + workspace_skills = workspace / "skills" + workspace_skills.mkdir(parents=True) + workspace_path = _write_skill(workspace_skills, "alpha", body="# Alpha") + builtin = tmp_path / "builtin" + builtin_path = _write_skill(builtin, "beta", body="# Beta") + aliases = tmp_path / "resources" / "view" + resource_view = ResourceView( + root=aliases, + agent=aliases / "agent", + media=aliases / "media", + package=aliases / "package", + ) + + loader = SkillsLoader( + workspace, + builtin_skills_dir=builtin, + resource_view=resource_view, + ) + entries = loader.list_skills(filter_unavailable=False) + summary = loader.build_skills_summary() + + assert {entry["path"] for entry in entries} == { + str(workspace_path), + str(builtin_path), + } + assert f"`{resource_view.agent / 'skills'}`" in summary + assert f"`{resource_view.package / 'skills'}`" in summary + assert str(workspace_path) not in summary + assert str(builtin_path) not in summary + + def test_bundled_update_setup_description_is_valid_yaml(tmp_path: Path) -> None: metadata = SkillsLoader(tmp_path).get_skill_metadata("update-setup") diff --git a/tests/agent/test_subagent.py b/tests/agent/test_subagent.py index 551dd9860..1349cdaa3 100644 --- a/tests/agent/test_subagent.py +++ b/tests/agent/test_subagent.py @@ -11,6 +11,7 @@ from nanobot.agent.tools.filesystem import FileToolsConfig from nanobot.bus.queue import MessageBus from nanobot.config.schema import ToolsConfig from nanobot.providers.base import GenerationSettings, LLMProvider +from nanobot.resource_links import ResourceView from nanobot.security.workspace_access import build_workspace_scope from nanobot.utils.llm_runtime import LLMRuntime @@ -109,6 +110,51 @@ def test_subagent_prompt_explains_grouped_skill_paths(tmp_path): assert "project-custom" not in prompt +def test_subagent_prompt_uses_restricted_resource_aliases(tmp_path): + agent_workspace = tmp_path / "agent" + aliases = tmp_path / "resources" / "view" + resource_view = ResourceView( + root=aliases, + agent=aliases / "agent", + media=aliases / "media", + package=aliases / "package", + ) + manager = SubagentManager( + workspace=agent_workspace, + bus=MessageBus(), + max_tool_result_chars=16_000, + resource_view=resource_view, + ) + + prompt = manager._build_subagent_prompt(resource_view_mode="restricted") + + assert f"Custom skills: `{resource_view.agent / 'skills'}`" in prompt + assert f"Media: `{resource_view.media}`" in prompt + assert f"Built-in skills: `{resource_view.package / 'skills'}`" in prompt + assert f"Agent workspace: `{resource_view.agent}`" not in prompt + assert f"Nanobot package: `{resource_view.package}`" not in prompt + assert f"History log: {agent_workspace.resolve() / 'memory' / 'history.jsonl'}" in prompt + + +def test_subagent_prompt_uses_agent_alias_for_full_history_path(tmp_path): + agent_workspace = tmp_path / "agent" + aliases = tmp_path / "resources" / "view" + resource_view = ResourceView( + root=aliases, + agent=aliases / "agent", + ) + manager = SubagentManager( + workspace=agent_workspace, + bus=MessageBus(), + max_tool_result_chars=16_000, + resource_view=resource_view, + ) + + prompt = manager._build_subagent_prompt(resource_view_mode="full") + + assert f"History log: {resource_view.agent / 'memory' / 'history.jsonl'}" in prompt + + @pytest.mark.asyncio async def test_subagent_keeps_project_runtime_scope_with_agent_owned_tools(tmp_path): agent_workspace = tmp_path / "agent" diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index 356ef8108..e664ba3fd 100644 --- a/tests/cli/test_commands.py +++ b/tests/cli/test_commands.py @@ -365,7 +365,7 @@ def test_status_help_shows_workspace_and_config_options(): assert "-c" in stripped_output -def test_status_uses_explicit_config_and_workspace(tmp_path: Path): +def test_status_uses_explicit_config_and_workspace(tmp_path: Path, monkeypatch): config_path = tmp_path / "instance" / "config.json" config_workspace = tmp_path / "config-workspace" override_workspace = tmp_path / "override-workspace" @@ -373,6 +373,11 @@ def test_status_uses_explicit_config_and_workspace(tmp_path: Path): config.agents.defaults.workspace = str(config_workspace) config_path.parent.mkdir(parents=True) config_path.write_text(json.dumps(config.model_dump(mode="json", by_alias=True))) + monkeypatch.setattr( + cli_commands, + "_prepare_resource_view", + lambda _config: pytest.fail("status must not prepare runtime resource links"), + ) result = runner.invoke( app, @@ -387,6 +392,58 @@ def test_status_uses_explicit_config_and_workspace(tmp_path: Path): assert str(config_workspace) not in compact_output +def test_prepare_resource_view_uses_active_config_and_workspace( + monkeypatch, + tmp_path: Path, +) -> None: + from nanobot import resource_links + + config_path = (tmp_path / "instance" / "config.json").resolve() + workspace = (tmp_path / "workspace").resolve() + config = Config() + config.agents.defaults.workspace = str(workspace) + expected = SimpleNamespace(warnings=()) + captured: dict[str, Path] = {} + + monkeypatch.setattr( + "nanobot.config.loader.get_config_path", + lambda: config_path, + ) + + def _fake_ensure_resource_view(**kwargs): + captured.update(kwargs) + return expected + + monkeypatch.setattr(resource_links, "ensure_resource_view", _fake_ensure_resource_view) + + assert cli_commands._prepare_resource_view(config) is expected + assert captured == { + "data_dir": config_path.parent, + "config_path": config_path, + "agent_workspace": workspace, + } + + +def test_prepare_resource_view_failure_does_not_block_runtime( + monkeypatch, + tmp_path: Path, +) -> None: + from nanobot import resource_links + + config_path = tmp_path / "config.json" + monkeypatch.setattr( + "nanobot.config.loader.get_config_path", + lambda: config_path, + ) + + def _fail(**_kwargs): + raise OSError("read-only filesystem") + + monkeypatch.setattr(resource_links, "ensure_resource_view", _fail) + + assert cli_commands._prepare_resource_view(Config()) is None + + def test_onboard_interactive_discard_does_not_save_or_create_workspace(mock_paths, monkeypatch): config_file, workspace_dir, _ = mock_paths @@ -1442,10 +1499,15 @@ def mock_agent_runtime(tmp_path): """Mock agent command dependencies for focused CLI tests.""" config = Config() config.agents.defaults.workspace = str(tmp_path / "default-workspace") + resource_view = object() with patch("nanobot.config.loader.load_config", return_value=config) as mock_load_config, \ patch("nanobot.config.loader.resolve_config_env_vars", side_effect=lambda c: c), \ patch("nanobot.cli.commands.sync_workspace_templates") as mock_sync_templates, \ + patch( + "nanobot.cli.commands._prepare_resource_view", + return_value=resource_view, + ) as mock_prepare_resource_view, \ patch("nanobot.providers.factory.make_provider", return_value=_fake_provider()), \ patch("nanobot.cli.commands._print_agent_response") as mock_print_response, \ patch("nanobot.bus.queue.MessageBus"), \ @@ -1463,6 +1525,8 @@ def mock_agent_runtime(tmp_path): "config": config, "load_config": mock_load_config, "sync_templates": mock_sync_templates, + "prepare_resource_view": mock_prepare_resource_view, + "resource_view": resource_view, "from_config": mock_from_config, "agent_loop": agent_loop, "print_response": mock_print_response, @@ -1490,6 +1554,9 @@ def test_agent_uses_default_config_when_no_workspace_or_config_flags(mock_agent_ ) passed_config = mock_agent_runtime["from_config"].call_args.args[0] assert passed_config.workspace_path == mock_agent_runtime["config"].workspace_path + assert mock_agent_runtime["from_config"].call_args.kwargs["resource_view"] is ( + mock_agent_runtime["resource_view"] + ) mock_agent_runtime["agent_loop"].process_direct.assert_awaited_once() mock_agent_runtime["print_response"].assert_called_once_with( "mock-response", render_markdown=True, metadata={}, @@ -1520,6 +1587,7 @@ def test_agent_config_sets_active_path(monkeypatch, tmp_path: Path) -> None: ) monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config) monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None) + monkeypatch.setattr("nanobot.cli.commands._prepare_resource_view", lambda _config: None) monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider()) monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object()) monkeypatch.setattr("nanobot.cron.service.CronService", lambda _store: object()) @@ -1558,6 +1626,7 @@ def test_agent_uses_workspace_directory_for_cron_store(monkeypatch, tmp_path: Pa monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None) monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config) monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None) + monkeypatch.setattr("nanobot.cli.commands._prepare_resource_view", lambda _config: None) monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider()) monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object()) @@ -1607,6 +1676,7 @@ def test_agent_workspace_override_does_not_migrate_legacy_cron( monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None) monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config) monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None) + monkeypatch.setattr("nanobot.cli.commands._prepare_resource_view", lambda _config: None) monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider()) monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object()) monkeypatch.setattr("nanobot.config.paths.get_cron_dir", lambda: legacy_dir) @@ -1663,6 +1733,7 @@ def test_agent_custom_config_workspace_does_not_migrate_legacy_cron( monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None) monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config) monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None) + monkeypatch.setattr("nanobot.cli.commands._prepare_resource_view", lambda _config: None) monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider()) monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object()) monkeypatch.setattr("nanobot.config.paths.get_cron_dir", lambda: legacy_dir) @@ -1858,6 +1929,7 @@ def _patch_cli_command_runtime( session_manager=None, cron_service=None, get_cron_dir=None, + prepare_resource_view=None, ) -> None: provider_factory = make_provider or (lambda _config: _fake_provider()) @@ -1871,6 +1943,10 @@ def _patch_cli_command_runtime( "nanobot.cli.commands.sync_workspace_templates", sync_templates or (lambda _path: None), ) + monkeypatch.setattr( + "nanobot.cli.commands._prepare_resource_view", + prepare_resource_view or (lambda _config: None), + ) monkeypatch.setattr( "nanobot.providers.factory.make_provider", provider_factory, @@ -2429,6 +2505,8 @@ def test_webui_foreground_refuses_occupied_webui_port(monkeypatch, tmp_path: Pat def _patch_serve_runtime(monkeypatch, config: Config, seen: dict[str, object]) -> None: pytest.importorskip("aiohttp") + resource_view = object() + seen["expected_resource_view"] = resource_view class _FakeApiApp: def __init__(self) -> None: @@ -2441,6 +2519,7 @@ def _patch_serve_runtime(monkeypatch, config: Config, seen: dict[str, object]) - return cls(workspace=config.workspace_path, **extra) def __init__(self, **kwargs) -> None: seen["workspace"] = kwargs["workspace"] + seen["resource_view"] = kwargs["resource_view"] async def _connect_mcp(self) -> None: return None @@ -2470,6 +2549,7 @@ def _patch_serve_runtime(monkeypatch, config: Config, seen: dict[str, object]) - config, message_bus=lambda: object(), session_manager=lambda _workspace: object(), + prepare_resource_view=lambda _config: resource_view, ) monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop) monkeypatch.setattr("nanobot.api.server.create_app", _fake_create_app) @@ -2887,6 +2967,7 @@ def test_gateway_local_trigger_queue_submits_agent_turns( config.gateway.heartbeat.enabled = False bus = MagicMock() seen: dict[str, object] = {} + resource_view = object() _patch_cli_command_runtime( monkeypatch, @@ -2894,6 +2975,7 @@ def test_gateway_local_trigger_queue_submits_agent_turns( message_bus=lambda: bus, session_manager=lambda _workspace: _FakeSessionManager(), cron_service=lambda _store_path: _FakeCronService(), + prepare_resource_view=lambda _config: resource_view, ) class _FakeMemory: @@ -2997,6 +3079,7 @@ def test_gateway_local_trigger_queue_submits_agent_turns( agent_kwargs = seen["agent_from_config_kwargs"] kwargs = seen["local_trigger_queue_kwargs"] assert isinstance(agent_kwargs["provider"], UnconfiguredProvider) is bool(setup_error) + assert agent_kwargs["resource_view"] is resource_view assert "local_trigger_store" in agent_kwargs assert kwargs["store"] is agent_kwargs["local_trigger_store"] assert "bus" not in kwargs @@ -3608,6 +3691,7 @@ def test_serve_uses_api_config_defaults_and_workspace_override( assert result.exit_code == 0 assert seen["workspace"] == override_workspace + assert seen["resource_view"] is seen["expected_resource_view"] assert seen["host"] == "127.0.0.2" assert seen["port"] == 18900 assert seen["request_timeout"] == 45.0 diff --git a/tests/security/test_resource_view_access.py b/tests/security/test_resource_view_access.py new file mode 100644 index 000000000..78e095439 --- /dev/null +++ b/tests/security/test_resource_view_access.py @@ -0,0 +1,104 @@ +from pathlib import Path + +import pytest + +from nanobot.resource_links import ensure_resource_view +from nanobot.security.workspace_policy import WorkspaceBoundaryError, resolve_allowed_path + + +@pytest.fixture +def resource_targets(tmp_path: Path) -> tuple[Path, Path, Path, Path]: + data_dir = tmp_path / "data" + workspace = tmp_path / "agent" + package = tmp_path / "package" / "nanobot" + project = tmp_path / "project" + (workspace / "skills" / "custom").mkdir(parents=True) + (workspace / "memory").mkdir() + (package / "skills" / "builtin").mkdir(parents=True) + (package / "templates").mkdir() + project.mkdir() + (workspace / "skills" / "custom" / "SKILL.md").write_text("custom", encoding="utf-8") + (workspace / "memory" / "history.jsonl").write_text("{}\n", encoding="utf-8") + (package / "skills" / "builtin" / "SKILL.md").write_text("builtin", encoding="utf-8") + (package / "templates" / "identity.md").write_text("identity", encoding="utf-8") + return data_dir, workspace, package, project + + +def _view_for(targets: tuple[Path, Path, Path, Path]): + data_dir, workspace, package, _ = targets + view = ensure_resource_view( + data_dir=data_dir, + config_path=data_dir / "config.json", + agent_workspace=workspace, + package_root=package, + ) + if view.agent is None or view.media is None or view.package is None: + pytest.skip(f"directory links unavailable: {view.warnings}") + return view + + +def test_restricted_access_follows_resource_alias_targets( + resource_targets: tuple[Path, Path, Path, Path], +) -> None: + _, workspace, package, project = resource_targets + view = _view_for(resource_targets) + + custom_skill = resolve_allowed_path( + view.agent / "skills" / "custom" / "SKILL.md", + workspace=project, + allowed_root=project, + extra_allowed_roots=[workspace / "skills", package / "skills"], + strict=True, + ) + builtin_skill = resolve_allowed_path( + view.package / "skills" / "builtin" / "SKILL.md", + workspace=project, + allowed_root=project, + extra_allowed_roots=[workspace / "skills", package / "skills"], + strict=True, + ) + media_root = resolve_allowed_path( + view.media, + workspace=project, + allowed_root=project, + extra_allowed_roots=[resource_targets[0] / "media"], + strict=True, + ) + + assert custom_skill == (workspace / "skills" / "custom" / "SKILL.md").resolve() + assert builtin_skill == (package / "skills" / "builtin" / "SKILL.md").resolve() + assert media_root == (resource_targets[0] / "media").resolve() + + +def test_alias_does_not_expand_restricted_package_or_agent_access( + resource_targets: tuple[Path, Path, Path, Path], +) -> None: + _, workspace, _, project = resource_targets + view = _view_for(resource_targets) + + with pytest.raises(WorkspaceBoundaryError): + resolve_allowed_path( + view.package / "templates" / "identity.md", + workspace=project, + allowed_root=project, + extra_allowed_roots=[workspace / "skills"], + strict=True, + ) + + history = workspace / "memory" / "history.jsonl" + with pytest.raises(WorkspaceBoundaryError): + resolve_allowed_path( + view.agent / "memory" / "history.jsonl", + workspace=project, + allowed_root=project, + extra_allowed_files=[history], + strict=True, + ) + + assert resolve_allowed_path( + history, + workspace=project, + allowed_root=project, + extra_allowed_files=[history], + strict=True, + ) == history.resolve() diff --git a/tests/test_nanobot_facade.py b/tests/test_nanobot_facade.py index d66c41558..97803cb52 100644 --- a/tests/test_nanobot_facade.py +++ b/tests/test_nanobot_facade.py @@ -10,6 +10,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, patch import pytest +from nanobot.config.schema import Config from nanobot.nanobot import ( STREAM_EVENT_REASONING_COMPLETED, STREAM_EVENT_REASONING_DELTA, @@ -30,6 +31,7 @@ from nanobot.nanobot import ( StreamEvent, StreamEventType, ) +from nanobot.nanobot import _prepare_resource_view as prepare_resource_view from nanobot.runtime_context import ( RUNTIME_CONTEXT_HISTORY_META, RuntimeContextBlock, @@ -39,6 +41,15 @@ from nanobot.session.manager import FILE_MAX_MESSAGES from nanobot.utils.llm_runtime import runtime_from_provider_snapshot +@pytest.fixture(autouse=True) +def _disable_sdk_resource_view_creation(monkeypatch) -> None: + """Keep facade tests from creating runtime links unless a test opts in.""" + monkeypatch.setattr( + "nanobot.nanobot._prepare_resource_view", + lambda _config, _config_path: None, + ) + + def _write_config(tmp_path: Path, overrides: dict | None = None) -> Path: data = { "providers": {"openrouter": {"apiKey": "sk-test-key"}}, @@ -138,6 +149,72 @@ def test_from_config_default_path(): mock_load.assert_called_once_with(None) +def test_from_config_scopes_resource_view_to_custom_config_without_global_mutation( + monkeypatch, + tmp_path: Path, +) -> None: + from nanobot.config import loader + + instance_dir = tmp_path / "instance" + instance_dir.mkdir() + config_path = _write_config(instance_dir) + workspace = tmp_path / "workspace" + unrelated_config = tmp_path / "other" / "config.json" + monkeypatch.setattr(loader, "_current_config_path", unrelated_config) + resource_view = object() + + with patch( + "nanobot.nanobot._prepare_resource_view", + return_value=resource_view, + ) as mock_prepare, patch("nanobot.nanobot.AgentLoop.from_config") as mock_loop: + bot = Nanobot.from_config(config_path, workspace=workspace) + + prepared_config, prepared_path = mock_prepare.call_args.args + assert prepared_path == config_path.resolve() + assert prepared_config.workspace_path == workspace.resolve() + assert mock_loop.call_args.kwargs["resource_view"] is resource_view + assert loader.get_config_path() == unrelated_config + assert bot._loop is mock_loop.return_value + + +def test_sdk_resource_view_failure_is_non_fatal( + monkeypatch, + tmp_path: Path, +) -> None: + from nanobot import resource_links + + config = Config() + config.agents.defaults.workspace = str(tmp_path / "workspace") + + def _fail(**_kwargs): + raise PermissionError("read-only") + + monkeypatch.setattr(resource_links, "ensure_resource_view", _fail) + + assert prepare_resource_view(config, tmp_path / "config.json") is None + + +def test_sdk_resource_view_prepares_fresh_workspace_before_linking( + monkeypatch, + tmp_path: Path, +) -> None: + from nanobot import resource_links + + config = Config() + workspace = tmp_path / "fresh-workspace" + config.agents.defaults.workspace = str(workspace) + expected = SimpleNamespace(warnings=()) + + def _capture(**kwargs): + assert workspace.is_dir() + assert kwargs["agent_workspace"] == workspace + return expected + + monkeypatch.setattr(resource_links, "ensure_resource_view", _capture) + + assert prepare_resource_view(config, tmp_path / "config.json") is expected + + @pytest.mark.asyncio async def test_run_returns_result(tmp_path): config_path = _write_config(tmp_path) diff --git a/tests/test_resource_links.py b/tests/test_resource_links.py new file mode 100644 index 000000000..f063ddb18 --- /dev/null +++ b/tests/test_resource_links.py @@ -0,0 +1,332 @@ +from __future__ import annotations + +import json +import os +import subprocess +from pathlib import Path + +import pytest +from filelock import Timeout + +from nanobot import resource_links +from nanobot.resource_links import ResourceView, ensure_resource_view + + +def _targets(tmp_path: Path) -> tuple[Path, Path, Path, Path]: + data_dir = tmp_path / "state" + config_path = data_dir / "config.json" + agent_workspace = tmp_path / "agent" + package_root = tmp_path / "package" + agent_workspace.mkdir() + package_root.mkdir() + return data_dir, config_path, agent_workspace, package_root + + +def _ensure( + data_dir: Path, + config_path: Path, + agent_workspace: Path, + package_root: Path, +) -> ResourceView: + return ensure_resource_view( + data_dir=data_dir, + config_path=config_path, + agent_workspace=agent_workspace, + package_root=package_root, + ) + + +def _remove_directory_link(path: Path) -> None: + try: + path.unlink() + except OSError: + os.rmdir(path) + + +def test_ensure_resource_view_is_stable_and_idempotent(tmp_path: Path) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + + first = _ensure(data_dir, config_path, agent_workspace, package_root) + second = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert first == second + assert first.warnings == () + assert first.root is not None + assert len(first.root.name) == 16 + assert first.agent is not None + assert first.agent.resolve(strict=True) == agent_workspace.resolve(strict=True) + assert first.media is not None + assert first.media.resolve(strict=True) == (data_dir / "media").resolve(strict=True) + assert first.package is not None + assert first.package.resolve(strict=True) == package_root.resolve(strict=True) + + +def test_resource_view_id_isolated_by_config_workspace_and_package(tmp_path: Path) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + other_workspace = tmp_path / "other-agent" + other_package = tmp_path / "other-package" + other_workspace.mkdir() + other_package.mkdir() + + baseline = _ensure(data_dir, config_path, agent_workspace, package_root) + config_variant = _ensure( + data_dir, + data_dir / "other-config.json", + agent_workspace, + package_root, + ) + workspace_variant = _ensure(data_dir, config_path, other_workspace, package_root) + package_variant = _ensure(data_dir, config_path, agent_workspace, other_package) + + roots = { + baseline.root, + config_variant.root, + workspace_variant.root, + package_variant.root, + } + assert None not in roots + assert len(roots) == 4 + + +def test_partial_link_failure_only_degrades_that_alias( + monkeypatch, + tmp_path: Path, +) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + real_create = resource_links._create_directory_link + + def fail_media(alias: Path, target: Path) -> None: + if alias.name == "media": + raise PermissionError("media denied") + real_create(alias, target) + + monkeypatch.setattr(resource_links, "_create_directory_link", fail_media) + + view = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert view.root is not None + assert view.agent is not None + assert view.media is None + assert view.package is not None + assert any("media denied" in warning for warning in view.warnings) + + +def test_existing_alias_collision_is_never_replaced(tmp_path: Path) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + first = _ensure(data_dir, config_path, agent_workspace, package_root) + assert first.agent is not None + _remove_directory_link(first.agent) + first.agent.write_text("user-owned", encoding="utf-8") + + second = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert second.root == first.root + assert second.agent is None + assert second.media is not None + assert second.package is not None + assert first.agent.read_text(encoding="utf-8") == "user-owned" + assert any("alias collision for agent" in warning for warning in second.warnings) + + +def test_wrong_link_is_never_repointed(tmp_path: Path) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + wrong_target = tmp_path / "wrong-agent" + wrong_target.mkdir() + first = _ensure(data_dir, config_path, agent_workspace, package_root) + assert first.agent is not None + _remove_directory_link(first.agent) + resource_links._create_directory_link(first.agent, wrong_target) + + second = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert second.agent is None + assert first.agent.resolve(strict=True) == wrong_target.resolve(strict=True) + assert any("alias collision for agent" in warning for warning in second.warnings) + + +def test_unmanaged_namespace_collision_is_not_modified(tmp_path: Path) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + namespace = data_dir / "resources" + namespace.mkdir(parents=True) + user_file = namespace / "notes.txt" + user_file.write_text("keep me", encoding="utf-8") + + view = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert view.root is None + assert view.agent is None + assert user_file.read_text(encoding="utf-8") == "keep me" + assert list(namespace.iterdir()) == [user_file] + assert any("ownership marker missing" in warning for warning in view.warnings) + + +def test_mismatched_view_marker_is_not_repaired(tmp_path: Path) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + first = _ensure(data_dir, config_path, agent_workspace, package_root) + assert first.root is not None + marker = first.root / ".nanobot-resource-view.json" + payload = json.loads(marker.read_text(encoding="utf-8")) + payload["targets"]["agent"] = str(tmp_path / "someone-else") + marker.write_text(json.dumps(payload), encoding="utf-8") + + second = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert second.root is None + assert second.agent is None + assert any("marker does not match" in warning for warning in second.warnings) + + +def test_invalid_marker_encoding_degrades_without_raising(tmp_path: Path) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + first = _ensure(data_dir, config_path, agent_workspace, package_root) + assert first.root is not None + marker = first.root / ".nanobot-resource-view.json" + marker.write_bytes(b"\xff") + + second = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert second.root is None + assert any("Could not read resource view marker" in warning for warning in second.warnings) + + +def test_failed_marker_write_removes_only_new_empty_view_directory( + monkeypatch, + tmp_path: Path, +) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + real_write_marker = resource_links._write_marker + + def fail_view_marker(marker_path: Path, payload: dict) -> None: + if marker_path.name == resource_links._VIEW_MARKER: + raise PermissionError("view marker denied") + real_write_marker(marker_path, payload) + + monkeypatch.setattr(resource_links, "_write_marker", fail_view_marker) + + view = _ensure(data_dir, config_path, agent_workspace, package_root) + + namespace = data_dir / "resources" + assert view.root is None + assert namespace.is_dir() + assert [entry.name for entry in namespace.iterdir()] == [ + resource_links._NAMESPACE_MARKER + ] + assert any("view marker denied" in warning for warning in view.warnings) + + +def test_view_inside_agent_target_is_fully_disabled_to_avoid_recursive_walk( + tmp_path: Path, +) -> None: + agent_workspace = tmp_path / "agent" + data_dir = agent_workspace / ".nanobot" + config_path = data_dir / "config.json" + package_root = tmp_path / "package" + agent_workspace.mkdir() + package_root.mkdir() + + view = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert view.root is None + assert view.agent is None + assert view.media is None + assert view.package is None + assert not (data_dir / "resources").exists() + assert any("recursive traversal unsafe" in warning for warning in view.warnings) + + +def test_unverified_new_link_is_removed_without_touching_target( + monkeypatch, + tmp_path: Path, +) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + real_points_to = resource_links._link_points_to + + def fail_agent_verification(alias: Path, target: Path) -> bool: + if alias.name == "agent": + return False + return real_points_to(alias, target) + + monkeypatch.setattr(resource_links, "_link_points_to", fail_agent_verification) + + view = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert view.root is not None + assert view.agent is None + assert not os.path.lexists(view.root / "agent") + assert agent_workspace.is_dir() + assert view.media is not None + assert view.package is not None + assert any("could not be verified" in warning for warning in view.warnings) + + +def test_lock_timeout_is_nonfatal_and_finite(monkeypatch, tmp_path: Path) -> None: + data_dir, config_path, agent_workspace, package_root = _targets(tmp_path) + observed_timeouts: list[float] = [] + + def fail_lock(lock_path: str, *, timeout: float): + observed_timeouts.append(timeout) + raise Timeout(lock_path) + + monkeypatch.setattr(resource_links, "FileLock", fail_lock) + + view = _ensure(data_dir, config_path, agent_workspace, package_root) + + assert observed_timeouts == [resource_links._LOCK_TIMEOUT_SECONDS] + assert view == ResourceView( + warnings=( + f"Timed out waiting for resource view lock: " + f"{data_dir.resolve() / '.nanobot-resource-links.lock'}", + ) + ) + + +def test_windows_symlink_failure_falls_back_to_junction(monkeypatch, tmp_path: Path) -> None: + alias = tmp_path / "alias" + target = tmp_path / "target" + target.mkdir() + junction_calls: list[tuple[Path, Path]] = [] + + def fail_symlink(self: Path, target: Path, *, target_is_directory: bool = False) -> None: + assert target_is_directory is True + raise PermissionError("symlinks unavailable") + + def record_junction(link: Path, junction_target: Path) -> None: + junction_calls.append((link, junction_target)) + + monkeypatch.setattr(Path, "symlink_to", fail_symlink) + monkeypatch.setattr(resource_links, "_is_windows", lambda: True) + monkeypatch.setattr(resource_links, "_create_windows_junction", record_junction) + + resource_links._create_directory_link(alias, target) + + assert junction_calls == [(alias, target)] + + +def test_windows_junction_command_timeout_is_bounded(monkeypatch, tmp_path: Path) -> None: + observed_timeouts: list[float] = [] + + def time_out(command: str, **kwargs): + observed_timeouts.append(kwargs["timeout"]) + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + + monkeypatch.setattr(resource_links.subprocess, "run", time_out) + + with pytest.raises(OSError, match="Timed out creating Windows junction"): + resource_links._create_windows_junction(tmp_path / "alias", tmp_path / "target") + + assert observed_timeouts == [resource_links._JUNCTION_TIMEOUT_SECONDS] + + +def test_default_package_root_points_to_installed_nanobot_package(tmp_path: Path) -> None: + data_dir = tmp_path / "state" + agent_workspace = tmp_path / "agent" + agent_workspace.mkdir() + + view = ensure_resource_view( + data_dir=data_dir, + config_path=data_dir / "config.json", + agent_workspace=agent_workspace, + ) + + assert view.package is not None + assert view.package.resolve(strict=True) == Path(resource_links.__file__).parent.resolve(strict=True)