fix(plugins): harden activation boundaries

docs/configuration.md

@@ -2354,6 +2354,8 @@ nanobot discovers [Agent Plugins](https://agent-plugins.org/) under `<workspace>
 Directory presence means installed; activation is explicit in **Apps**. Skills use progressive loading and `$skill-name` invocation, with workspace > plugin > built-in precedence.
 Enabled `stdio` servers receive contained `PLUGIN_ROOT` and isolated `PLUGIN_DATA` paths; explicit
 `tools.mcpServers` entries win collisions. Invalid or escaping components are ignored.
+An enabled package is treated as immutable: changing any packaged file disables it until the user
+reviews and enables it again. Runtime state belongs under `PLUGIN_DATA`, not the package root.
 
 Enabled plugins run as the nanobot user; permissions are descriptive, not an OS sandbox. The optional `extensions.dev.nanobot.logo` accepts a contained PNG, JPEG, or WebP up to 256 KiB.
 

nanobot/agent/plugins.py

@@ -118,7 +118,9 @@ def agent_plugin_mcp_servers(
             continue
         plugin_servers = _plugin_mcp_servers(workspace, plugin)
         for name, server in plugin_servers.items():
-            host_name = plugin.name if len(plugin_servers) == 1 else f"{plugin.name}-{name}"
+            # ``--`` cannot occur in a valid plugin identity, so multi-server
+            # namespaces cannot collide with a single-server plugin name.
+            host_name = plugin.name if len(plugin_servers) == 1 else f"{plugin.name}--{name}"
             servers[host_name] = server
     configured = configured or {}
     if collisions := servers.keys() & configured.keys():
@@ -146,7 +148,10 @@ def set_agent_plugin_enabled(workspace: Path, name: str, enabled: bool) -> None:
     data = _plugin_data_dir(workspace, plugin.name, create=True)
     marker = data / "enabled"
     if enabled:
-        marker.write_text(str(plugin.root), encoding="utf-8")
+        activation = _activation_marker(plugin)
+        if activation is None:
+            raise RuntimeError(f"Agent Plugin '{name}' changed while it was being enabled")
+        marker.write_text(activation, encoding="utf-8")
         marker.chmod(0o600)
     else:
         marker.unlink(missing_ok=True)
@@ -238,6 +243,7 @@ def _plugin_mcp_server(raw: object, root: Path, data: Path) -> MCPServerConfig |
             "args": [_expand(item, root, data) for item in server.args],
             "env": {
                 **{key: _expand(value, root, data) for key, value in server.env.items()},
+                "PYTHONDONTWRITEBYTECODE": "1",
                 "PLUGIN_ROOT": str(root),
                 "PLUGIN_DATA": str(data),
             },
@@ -303,11 +309,50 @@ def _plugin_data_dir(workspace: Path, name: str, *, create: bool) -> Path:
 def _enabled(workspace: Path, plugin: AgentPlugin) -> bool:
     marker = _plugin_data_dir(workspace, plugin.name, create=False) / "enabled"
     try:
-        return marker.is_file() and marker.read_text(encoding="utf-8") == str(plugin.root)
+        if not marker.is_file():
+            return False
+        current = marker.read_text(encoding="utf-8")
+        activation = _activation_marker(plugin)
+        if activation is None:
+            return False
+        if current == activation:
+            return True
+        if current == str(plugin.root):
+            marker.write_text(activation, encoding="utf-8")
+            marker.chmod(0o600)
+            return True
+        return False
     except OSError:
         return False
 
 
+def _activation_marker(plugin: AgentPlugin) -> str | None:
+    """Bind activation to one immutable package snapshot."""
+    digest = sha256()
+    try:
+        for candidate in sorted(plugin.root.rglob("*")):
+            relative = candidate.relative_to(plugin.root).as_posix()
+            digest.update(relative.encode())
+            if candidate.is_symlink():
+                digest.update(b"\0link\0")
+                digest.update(candidate.readlink().as_posix().encode())
+            elif candidate.is_file():
+                digest.update(b"\0file\0")
+                digest.update(candidate.read_bytes())
+            elif candidate.is_dir():
+                digest.update(b"\0dir\0")
+            else:
+                return None
+            digest.update(b"\0")
+    except OSError:
+        return None
+    return json.dumps(
+        {"fingerprint": digest.hexdigest(), "root": str(plugin.root)},
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+
+
 def _discover_plugin_skills(plugin_name: str, plugin_root: Path) -> list[tuple[str, Path]]:
     skills_root = _contained(plugin_root / "skills", plugin_root, directory=True)
     if skills_root is None:

nanobot/agent/tools/filesystem.py

@@ -148,9 +148,20 @@ class _FsTool(Tool):
         )
 
     def _resolve_read(self, path: str) -> Path:
+        plugin_skill_dirs: list[Path] = []
+        if self._workspace is not None:
+            from nanobot.agent.plugins import enabled_agent_plugin_skills
+
+            try:
+                plugin_skill_dirs = [
+                    skill.parent
+                    for _name, skill in enabled_agent_plugin_skills(Path(self._workspace))
+                ]
+            except (OSError, RuntimeError):
+                pass
         return self._resolve_with_extra(
             path,
-            self._extra_read_allowed_dirs,
+            [*self._extra_read_allowed_dirs, *plugin_skill_dirs],
             self._extra_read_allowed_files,
             include_media_dir=True,
             extra_files_require_allowed_root=True,

tests/agent/test_agent_plugins.py

@@ -14,6 +14,14 @@ from nanobot.agent.plugins import (
     set_agent_plugin_enabled,
 )
 from nanobot.agent.skills import SkillsLoader
+from nanobot.agent.tools.context import ToolContext
+from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool
+from nanobot.config.schema import ToolsConfig
+from nanobot.security.workspace_access import (
+    bind_workspace_scope,
+    reset_workspace_scope,
+    validate_workspace_scope_payload,
+)
 
 
 @pytest.fixture(autouse=True)
@@ -169,6 +177,72 @@ def test_plugin_mcp_requires_explicit_enable(tmp_path: Path) -> None:
     assert agent_plugin_mcp_servers(tmp_path) == {}
 
 
+def test_plugin_mcp_namespaces_cannot_shadow_plugin_identities(tmp_path: Path) -> None:
+    single = _plugin(tmp_path, "foo-bar")
+    multi = _plugin(tmp_path, "foo")
+    for root, servers in (
+        (single, {"main": {"type": "stdio", "command": "echo", "args": ["single"]}}),
+        (
+            multi,
+            {
+                "bar": {"type": "stdio", "command": "echo", "args": ["multi"]},
+                "other": {"type": "stdio", "command": "echo"},
+            },
+        ),
+    ):
+        _write_json(
+            root / "mcp.json",
+            {"$schema": AGENT_PLUGIN_MCP_SCHEMA, "mcpServers": servers},
+        )
+    set_agent_plugin_enabled(tmp_path, "foo-bar", True)
+    set_agent_plugin_enabled(tmp_path, "foo", True)
+
+    servers = agent_plugin_mcp_servers(tmp_path)
+
+    assert set(servers) == {"foo-bar", "foo--bar", "foo--other"}
+    assert servers["foo-bar"].args == ["single"]
+    assert servers["foo--bar"].args == ["multi"]
+
+
+@pytest.mark.asyncio
+async def test_restricted_project_can_read_only_enabled_plugin_skill(
+    tmp_path: Path,
+) -> None:
+    agent_workspace = tmp_path / "agent"
+    project = tmp_path / "project"
+    project.mkdir()
+    plugin = _plugin(agent_workspace)
+    skill = _skill(plugin / "skills", "demo-skill")
+    resource = skill / "reference.md"
+    resource.write_text("plugin reference", encoding="utf-8")
+    ctx = ToolContext(
+        config=ToolsConfig(restrict_to_workspace=True),
+        workspace=str(agent_workspace),
+    )
+    read_tool = ReadFileTool.create(ctx)
+    write_tool = WriteFileTool.create(ctx)
+    set_agent_plugin_enabled(agent_workspace, "demo", True)
+    scope = validate_workspace_scope_payload(
+        {"project_path": str(project), "access_mode": "restricted"},
+        default_workspace=agent_workspace,
+        default_restrict_to_workspace=True,
+    )
+
+    token = bind_workspace_scope(scope)
+    try:
+        read_result = await read_tool.execute(path=str(resource))
+        write_result = await write_tool.execute(path=str(resource), content="changed")
+        set_agent_plugin_enabled(agent_workspace, "demo", False)
+        disabled_result = await read_tool.execute(path=str(resource))
+    finally:
+        reset_workspace_scope(token)
+
+    assert "plugin reference" in read_result
+    assert "outside allowed directory" in write_result
+    assert "outside allowed directory" in disabled_result
+    assert resource.read_text(encoding="utf-8") == "plugin reference"
+
+
 def test_plugin_state_symlink_cannot_escape_config_root(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
@@ -213,3 +287,68 @@ def test_plugin_activation_requires_one_stable_package_identity(tmp_path: Path)
     roots[0].rename(moved)
     assert discover_agent_plugins(tmp_path)[0].enabled is False
     assert agent_plugin_mcp_servers(tmp_path) == {}
+
+
+def test_legacy_path_activation_is_upgraded_to_package_fingerprint(tmp_path: Path) -> None:
+    plugin = _plugin(tmp_path)
+    set_agent_plugin_enabled(tmp_path, "demo", True)
+    marker = next((tmp_path / "config" / "plugin-data").glob("*/demo/enabled"))
+    marker.write_text(str(plugin), encoding="utf-8")
+
+    assert discover_agent_plugins(tmp_path)[0].enabled is True
+    assert marker.read_text(encoding="utf-8").startswith('{"fingerprint":')
+
+
+def test_plugin_activation_does_not_survive_in_place_contract_replacement(
+    tmp_path: Path,
+) -> None:
+    plugin = _plugin(tmp_path, "desktop")
+    mcp = plugin / "mcp.json"
+
+    def write_server(marker: str) -> None:
+        _write_json(
+            mcp,
+            {
+                "$schema": AGENT_PLUGIN_MCP_SCHEMA,
+                "mcpServers": {
+                    "server": {"type": "stdio", "command": "echo", "args": [marker]}
+                },
+            },
+        )
+
+    write_server("trusted")
+    set_agent_plugin_enabled(tmp_path, "desktop", True)
+    assert agent_plugin_mcp_servers(tmp_path)["desktop"].args == ["trusted"]
+
+    write_server("replacement")
+
+    assert discover_agent_plugins(tmp_path)[0].enabled is False
+    assert agent_plugin_mcp_servers(tmp_path) == {}
+
+
+def test_plugin_activation_does_not_survive_in_place_code_replacement(
+    tmp_path: Path,
+) -> None:
+    plugin = _plugin(tmp_path, "desktop")
+    executable = plugin / "server.py"
+    executable.write_text("print('trusted')\n", encoding="utf-8")
+    _write_json(
+        plugin / "mcp.json",
+        {
+            "$schema": AGENT_PLUGIN_MCP_SCHEMA,
+            "mcpServers": {
+                "server": {
+                    "type": "stdio",
+                    "command": "python",
+                    "args": ["${PLUGIN_ROOT}/server.py"],
+                }
+            },
+        },
+    )
+    set_agent_plugin_enabled(tmp_path, "desktop", True)
+    assert discover_agent_plugins(tmp_path)[0].enabled is True
+
+    executable.write_text("print('replacement')\n", encoding="utf-8")
+
+    assert discover_agent_plugins(tmp_path)[0].enabled is False
+    assert agent_plugin_mcp_servers(tmp_path) == {}