From b744029c3741e6a09e1a71a183f86693da2a2ace Mon Sep 17 00:00:00 2001 From: Herik Webb Date: Fri, 9 Oct 2026 13:49:57 -0400 Subject: [PATCH] fix(acp)!: require trusted preinstalled adapters Prevent workspace npm projects and shared-ancestor configuration from choosing the adapter before ACP tool approval. Resolve preinstalled executables outside the workspace, filter their PATH, and use shared nonexecuting interpreter checks for readiness and startup. Keep process startup in NBI's user directory while passing the notebook workspace explicitly to ACP sessions and the MCP workspace tool. Cover path aliases, interpreter lookup, invalid overrides and startup cleanup with regression tests and document installation and trust requirements. BREAKING CHANGE: Experimental ACP no longer installs its adapter through npx at startup. Install the adapter and runtime in trusted locations. NBI_ACP_AGENT_COMMAND must name an absolute installed executable; unset the variable to restore discovery rather than assigning an empty value. --- CHANGELOG.md | 6 + PRIVACY.md | 26 +- README.md | 10 +- docs/acp.md | 5 +- docs/admin-guide.md | 38 +- notebook_intelligence/acp_agent.py | 67 ++-- notebook_intelligence/acp_mcp_server.py | 16 +- notebook_intelligence/acp_registry.py | 148 +++++++- notebook_intelligence/acp_runtime.py | 112 ++++++ notebook_intelligence/extension.py | 2 +- notebook_intelligence/readiness.py | 103 +++--- src/components/settings-panel.tsx | 14 +- tests/test_acp_agent.py | 409 ++++++++++++++++++++- tests/test_acp_registry.py | 452 ++++++++++++++++++++++++ tests/test_acp_runtime.py | 203 +++++++++++ tests/test_readiness.py | 325 ++++++++++++++++- 16 files changed, 1800 insertions(+), 136 deletions(-) create mode 100644 notebook_intelligence/acp_runtime.py create mode 100644 tests/test_acp_registry.py create mode 100644 tests/test_acp_runtime.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 20d194ad..e73a0023 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,12 @@ For each release we list user-facing changes grouped as **Added**, **Changed**, +## [Unreleased] + +### Fixed + +- **ACP startup no longer lets npm projects choose the adapter.** Starting an adapter through `npx` could execute a package planted in the workspace or a shared ancestor of the user's home before any approval prompt. ACP now launches a preinstalled `codex-acp` executable directly from an absolute server `PATH` directory, excluding workspace paths and aliases from adapter discovery and the `PATH` inherited by agent commands. **Migration:** install the tested `@zed-industries/codex-acp@0.16.0` in a trusted location and maintain its version there; NBI does not verify versions found on `PATH` or through an override. `NBI_ACP_AGENT_COMMAND` must point to the installed adapter itself by absolute path, not `npx`, `npm exec`, `pnpm dlx`, `bunx`, or an `env` wrapper. An override preserves the executable's name for trusted shims and retains the full server `PATH`; it is an administrator trust decision covering the executable, its runtime and packages, and their parent directories. Readiness and startup use the same resolver, so relative, missing, malformed, and set-but-empty or whitespace-only commands are consistently blocked. Unset `NBI_ACP_AGENT_COMMAND` to restore default discovery; assigning an empty value does not reset it. The adapter starts in NBI's user directory, workspace sessions retain their explicit root, and the `nbi_workspace_root` MCP tool receives that root through an environment variable. See the [admin guidance](docs/admin-guide.md#gating-the-experimental-acp-agent-378). + ## [6.0.0] - unreleased 6.0.0 adds Chatbook, a notebook whose cells can be written in English and executed as generated code, together with its admin policy, its own kernel, and the context, mention, and ruleset plumbing around it. The rest of the release is hardening: the experimental ACP agent mode gets a tighter sandbox and better approval cards, and a long tail of chat, settings, MCP, and readiness fixes. The major version marks Chatbook's arrival rather than a compatibility break: no traitlet, env var, REST route, command id, or on-disk format was renamed or removed, `requires-python` is unchanged at `>=3.10`, and the supported JupyterLab range widened rather than narrowed. Upgrading needs no migration. diff --git a/PRIVACY.md b/PRIVACY.md index 2a3c32cc..3c8766aa 100644 --- a/PRIVACY.md +++ b/PRIVACY.md @@ -29,19 +29,19 @@ If your cells contain sensitive outputs (PHI, PII, secrets), clear them before i Hosts NBI may contact, depending on which features are enabled: -| Host | Purpose | -| ----------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `api.githubcopilot.com` | GitHub Copilot chat and inline completion | -| `api.github.com` | GitHub Copilot device-flow login; managed-skills manifest fetches when hosted on github.com; skill imports | -| `github.com`, `codeload.github.com` | Skill tarball downloads (Import from GitHub and the managed-skills reconciler) | -| `raw.githubusercontent.com` | Manifest fetches when `NBI_SKILLS_MANIFEST` points at a `raw.githubusercontent.com` URL | -| `api.anthropic.com` | Anthropic API for Claude-mode inline chat and auto-complete; also the default destination of the Claude Code CLI | -| `api.openai.com` | OpenAI-compatible provider (default Base URL) | -| Your configured Base URL | OpenAI-compatible, LiteLLM-compatible, or Claude when pointed at a self-hosted endpoint | -| `localhost:11434` (or your Ollama host) | Ollama local model serving | -| `registry.npmjs.org` and configured npm mirrors | MCP servers configured to launch via `npx -y`, and ACP mode, which launches its adapter with `npx` on every agent start unless `NBI_ACP_AGENT_COMMAND` points elsewhere | - -For the configurable destinations above (Base URLs, Ollama host, MCP `npx` packages), the destination is whatever you or your admin set. NBI itself starts no other network activity. In ACP mode the agent subprocess makes its own calls, which NBI neither proxies nor enumerates: the destinations there are the agent's to document, and blocking egress at the network layer is the only way to bound them from outside. +| Host | Purpose | +| ----------------------------------------------- | ---------------------------------------------------------------------------------------------------------------- | +| `api.githubcopilot.com` | GitHub Copilot chat and inline completion | +| `api.github.com` | GitHub Copilot device-flow login; managed-skills manifest fetches when hosted on github.com; skill imports | +| `github.com`, `codeload.github.com` | Skill tarball downloads (Import from GitHub and the managed-skills reconciler) | +| `raw.githubusercontent.com` | Manifest fetches when `NBI_SKILLS_MANIFEST` points at a `raw.githubusercontent.com` URL | +| `api.anthropic.com` | Anthropic API for Claude-mode inline chat and auto-complete; also the default destination of the Claude Code CLI | +| `api.openai.com` | OpenAI-compatible provider (default Base URL) | +| Your configured Base URL | OpenAI-compatible, LiteLLM-compatible, or Claude when pointed at a self-hosted endpoint | +| `localhost:11434` (or your Ollama host) | Ollama local model serving | +| `registry.npmjs.org` and configured npm mirrors | MCP servers configured to launch via `npx -y` | + +For the configurable destinations above (Base URLs, Ollama host, MCP `npx` packages), the destination is whatever you or your admin set. NBI itself starts no other network activity. ACP uses a preinstalled adapter; NBI does not fetch it from npm at startup. In ACP mode the agent subprocess makes its own calls, which NBI neither proxies nor enumerates: the destinations there are the agent's to document, and blocking egress at the network layer is the only way to bound them from outside. For air-gapped or egress-restricted environments, see [`docs/admin-guide.md`](docs/admin-guide.md#air-gap-deployment). diff --git a/README.md b/README.md index 8408ddbe..ce20e401 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ NBI is free and open-source. Connect it to a free or paid LLM provider of your c - Python 3.10+ - JupyterLab 4.x -- Node.js — required for [Claude mode](#claude-mode) (the Claude Code CLI), for [ACP agent mode](#acp-agent-mode), which launches its adapter with `npx` on every start, and for MCP servers that launch via `npx`. +- Node.js — required for [Claude mode](#claude-mode) (the Claude Code CLI), for the preinstalled `codex-acp` adapter used by [ACP agent mode](#acp-agent-mode), and for MCP servers that launch via `npx`. ACP mode requires a preinstalled adapter in a trusted location; NBI does not download it at startup. - A fresh virtualenv or conda env is recommended so NBI doesn't conflict with system Python. ## Quick start @@ -292,7 +292,9 @@ Per-user preferences (default on for the cell-output features) live in `config.j Inline chat calls the Anthropic API directly while the other Claude modes go through the Claude Code CLI, so a deployment can need a different model id for each. It uses `inline_chat_model`, falling back to `chat_model` when that is unset. `NBI_CLAUDE_CHAT_MODEL` therefore governs both surfaces: while it is set, a stored `inline_chat_model` is ignored and its control is disabled. Set `NBI_CLAUDE_INLINE_CHAT_MODEL` as well to pin inline chat to a different model — it takes precedence and re-enables nothing, so both surfaces stay administrator-controlled. -`NBI_ACP_AGENT_COMMAND` (not a settings lock) overrides the adapter command NBI launches for the selected ACP agent type; the value is shell-split. For Codex, NBI appends its `-c` approval and sandbox options to that command, so a replacement must pass them through to codex-acp. Leave it unset to use the agent's registered default command. +`NBI_ACP_AGENT_COMMAND` (not a settings lock) selects an installed adapter by absolute executable path, for example `/opt/nbi/acp/bin/codex-acp`; the value is shell-split, so quote paths containing spaces. Point it at the installed adapter itself, not `npx`, `npm exec`, `pnpm dlx`, `bunx`, or an `env` wrapper: making a package runner's path absolute does not prevent it from selecting code from a project. For Codex, the adapter must honor the `-c` approval and sandbox options NBI appends. An override is an administrator trust decision covering the command, its runtime and packages, inherited `PATH`, and their parent directories. + +Leave the override unset to discover a preinstalled `codex-acp` in an absolute server `PATH` directory outside the Jupyter workspace. Empty or whitespace-only values are errors; use `unset NBI_ACP_AGENT_COMMAND` in a POSIX shell to restore default discovery. NBI follows symlinks before launching it; installations using a shim that depends on the name `codex-acp` need an explicit absolute path to that trusted shim. Default startup removes workspace directories from the `PATH` inherited by the agent and its commands, which can exclude a project's `.venv/bin`; an explicit override retains the full server `PATH`. Install the tested `@zed-industries/codex-acp@0.16.0` and maintain that version yourself: NBI does not verify or update versions for either discovery method. See the [adapter installation and migration guidance](docs/admin-guide.md#gating-the-experimental-acp-agent-378) for a concrete install command, home-directory workspace layouts, and the explicit adapter-path requirement for Windows with Python 3.10 or 3.11. Provider IDs: `github-copilot`, `openai-compatible`, `litellm-compatible`, `ollama`, `none`. The `*_MODEL_ID` value is whatever the chosen provider exposes (e.g. `gpt-4o`, `llama3:latest`). Claude model IDs are the literal IDs from the Anthropic API (e.g. `claude-opus-4-7`, `claude-sonnet-4-6`); empty string = "Default (recommended)"; `NBI_CLAUDE_INLINE_COMPLETION_MODEL` also accepts `none` (no inline completion in Claude mode) or `inherit` (use the General-tab Auto-complete model). @@ -470,7 +472,7 @@ c.NotebookIntelligence.enable_chat_feedback_always_visible = True NBI Settings opens on a **Status** card that answers one question: is this deployment configured to work, and if not, which specific piece is missing? Each check reports `ok`, `warn`, or `blocked`, and anything that is not `ok` carries the next action rather than just the failure ("the claude CLI was not found" plus "install it, or set `NBI_CLAUDE_CLI_PATH` and restart"). -It runs on open and costs nothing: it resolves your provider and model, checks the model list is reachable and that your selected model is one the endpoint actually serves, and for the agent modes checks that the CLI answers `--version` and that credentials are present. A missing key is a warning rather than an error, because the Claude CLI and ACP agents can hold their own subscription logins that NBI cannot see. +It runs on open and costs nothing: it resolves your provider and model, checks the model list is reachable and that your selected model is one the endpoint actually serves, checks the Claude CLI with `--version`, and checks ACP's adapter executable without running it. Both agent modes also check whether credentials are present. A missing key is a warning rather than an error, because the Claude CLI and ACP agents can hold their own subscription logins that NBI cannot see. **Test the endpoint** sends one short request to your configured model, after asking first. It is the only way to catch two failures nothing cheaper can see: a gateway that returns 200s but buffers instead of streaming, and a proxy that strips the `tools` field so agent mode silently never calls a tool. No model output is kept, so the result is safe to paste into a support ticket. @@ -523,7 +525,7 @@ Full guide, including the span and event reference, the probe thresholds, worked ## Roadmap -NBI 6.x is the current line. New features land in minor releases (6.1, 6.2, …); breaking changes are reserved for a major release and announced in the [changelog](CHANGELOG.md). 6.0 itself is not a breaking release: it marks Chatbook's arrival, and no traitlet, env var, REST route, or on-disk format was renamed or removed, so upgrading from 5.x needs no migration. Upgrading from 4.x? See the [5.0.0 migration note](CHANGELOG.md#migration-note) for the `fastmcp` → `mcp` dependency swap, the new path sandboxes, and the workspace-file-attach behavior change. +NBI 6.x is the current line. New features land in minor releases (6.1, 6.2, …); breaking changes are reserved for a major release and announced in the [changelog](CHANGELOG.md). 6.0 marks Chatbook's arrival, and no traitlet, env var, REST route, or on-disk format was renamed or removed. Deployments using the experimental ACP mode must now preinstall its adapter; see the [migration guidance](docs/admin-guide.md#gating-the-experimental-acp-agent-378). Upgrading from 4.x? See the [5.0.0 migration note](CHANGELOG.md#migration-note) for the `fastmcp` → `mcp` dependency swap, the new path sandboxes, and the workspace-file-attach behavior change. ## License diff --git a/docs/acp.md b/docs/acp.md index ed8879af..ad019db9 100644 --- a/docs/acp.md +++ b/docs/acp.md @@ -10,7 +10,7 @@ For the admin side of this feature (the two policies, the approval posture NBI p The **ACP** tab in Settings is hidden until an admin allows the mode, because `acp_mode_policy` defaults to `force-off`. If you do not see the tab, that is why; ask for `NBI_ACP_MODE_POLICY=user-choice`. -With the tab visible, pick the agent type, then set the chat model, API key, and base URL. The adapter is launched with `npx` on every agent start, so the server needs Node.js on its `PATH` even though nothing is installed permanently. +With the tab visible, pick the agent type, then set the chat model, API key, and base URL. An administrator must first install the tested `@zed-industries/codex-acp@0.16.0` in a trusted location outside the Jupyter workspace. NBI finds its `codex-acp` executable through absolute directories on the server's `PATH`, or uses an absolute path to the installed adapter configured through `NBI_ACP_AGENT_COMMAND`. Do not use package runners such as `npx`, `npm exec`, `pnpm dlx`, or `bunx`, or an `env` wrapper, for that override. NBI does not download an adapter at startup or verify its version; the administrator maintains the installation. Settings → Status reports a missing or invalid adapter command or declared script interpreter. See the [installation and migration guidance](admin-guide.md#gating-the-experimental-acp-agent-378), including a concrete install command, home-directory workspace layouts, and the requirements for shims and Windows with Python 3.10 or 3.11. Empty or whitespace-only `NBI_ACP_AGENT_COMMAND` values are invalid; unset the variable to restore default discovery. ## What a turn looks like @@ -22,7 +22,8 @@ Previous sessions are listed by the history control, and picking one resumes it ## Limits worth knowing -- **The first start is slow, and it is bounded.** An agent has 60 seconds to spawn, download the adapter through `npx`, initialize, authenticate, and create a session. On a cold cache over a slow link the download alone can exceed that, and the failure reads `ACP agent did not start in time`. Run the adapter command once by hand to warm the cache, or point `NBI_ACP_AGENT_COMMAND` at a pre-installed copy. +- **Agent commands use a filtered `PATH` by default.** Workspace directories, including a project's `.venv/bin`, are removed. Commands such as `python` or `pytest` may therefore select another installation or be unavailable. An administrator's explicit adapter override retains the full server `PATH` and requires trusting its contents. +- **Startup is bounded.** An agent has 60 seconds to spawn, initialize, authenticate, and create a session. If this takes too long, the failure reads `ACP agent did not start in time`. Check the adapter installation and authentication before retrying. - **A turn is capped at 30 minutes**, after which it fails with an agent response timeout. - **Session listing is optional.** Resuming depends on the agent advertising the capability, and listing is an optional protocol extension. An agent that implements neither reports that it does not support listing sessions; this is the agent's limitation, not a misconfiguration. - **Chatbook uses ACP differently.** When ACP mode is active, Chatbook generation runs in a separate session with full access forced off, no MCP servers, and permission requests denied, so a Chatbook cell cannot approve its way into running commands. diff --git a/docs/admin-guide.md b/docs/admin-guide.md index a894f33d..98a51356 100644 --- a/docs/admin-guide.md +++ b/docs/admin-guide.md @@ -126,7 +126,7 @@ The full surface, in one table. | `NBI_GITHUB_COPILOT_CLI_PATH` | str | unset | env | Absolute path to the GitHub Copilot CLI. When unset, NBI looks up `copilot` on `PATH`. Gates the GitHub Copilot launcher tile. | | `NBI_CODEX_CLI_PATH` | str | unset | env | Absolute path to the OpenAI Codex CLI. When unset, NBI looks up `codex` on `PATH`. Gates the Codex launcher tile. | | `NBI_CLAUDE_INLINE_COMPLETION_MAX_TOKENS` | int | `1024` | env | Output cap for Claude-mode auto-complete suggestions. Clamped to `[1, 4096]`; a value outside that range is pulled to the nearest bound rather than rejected. The default was lowered from 10,000 in 5.3.0 because a suggestion is at most a few dozen lines and the old ceiling let a rambling response generate for seconds before the extraction regex discarded most of it. | -| `NBI_ACP_AGENT_COMMAND` | str | unset | env | Overrides the adapter command NBI launches for the selected ACP agent type (shell-split). For Codex, NBI appends its `-c` approval and sandbox options, so a replacement must pass them through to codex-acp. Unset uses the agent's registered default command. See [Gating the experimental ACP agent](#gating-the-experimental-acp-agent-378). | +| `NBI_ACP_AGENT_COMMAND` | str | unset | env | Absolute path to the installed adapter executable (shell-split; quote paths containing spaces). Do not use a package runner or `env` wrapper. The adapter must honor NBI's appended `-c` options. Unset discovers `codex-acp` outside the workspace through absolute server `PATH` directories and filters the agent's `PATH`. Empty or whitespace-only values are errors; unset the variable to restore discovery. An override retains the full server `PATH` and trusts the executable, its runtime and packages, and their parent directories. The command starts in `~/.jupyter/nbi`. See [Gating the experimental ACP agent](#gating-the-experimental-acp-agent-378). | | `NBI_GHE_SUBDOMAIN` | str | `""` | env | GitHub Enterprise subdomain for GitHub Copilot users on a GHE tenant. Empty selects github.com. | | `NBI_GITHUB_ENTERPRISE_HOSTS` | csv | `""` | env | Comma-separated hostnames the plugin marketplace detector treats as GitHub. Cookie-domain shape: bare token (`github.acme.com`) matches exactly; leading-dot token (`.acme.com`) matches any subdomain of `acme.com`. Independent of `NBI_GHE_SUBDOMAIN`, which only configures the Copilot OAuth tenant. Required so `allow_github_plugin_import = False` actually gates GHE marketplace adds and so the `GITHUB_TOKEN` / `gh auth token` chain injects on GHE sources. | | `NBI_LOG_LEVEL` | str | `INFO` | env | Python logging level for the `notebook_intelligence` logger. | @@ -280,7 +280,7 @@ Steps for deploying to a network with no general internet egress: 2. **Manifest hosting.** Set `NBI_SKILLS_MANIFEST` to either a `file://` path (a manifest baked into the image) or an internal `https://` URL on a network the pod can reach. 3. **Skill bundles.** Either bake the skills into the image under `~/.claude/skills/` (managed status will reset since they aren't from the manifest), or host the GitHub-style tarballs at an internal mirror and write the manifest URLs to point at it. 4. **Disable user-initiated GitHub imports** at the network layer — block `github.com`, `codeload.github.com`, and `raw.githubusercontent.com`. Users can still install skills from the local filesystem by dropping bundles into `~/.claude/skills/`. -5. **ACP mode is `npx`-only by default.** The adapter is fetched with `npx` on every agent start, so an air-gapped deployment must either leave ACP off (it is `force-off` by default) or pre-install the adapter and point `NBI_ACP_AGENT_COMMAND` at it. A replacement command must still accept the `-c` overrides NBI passes, or the approval posture those pin is lost. +5. **Preinstall the ACP adapter if ACP mode is needed.** Include the tested `@zed-industries/codex-acp@0.16.0` and its runtime dependencies in a trusted location in the image and expose its executable through an absolute server `PATH` directory, or set `NBI_ACP_AGENT_COMMAND` to the installed adapter's absolute path. Do not use a package runner or `env` wrapper. NBI does not download the adapter or verify its version at startup. The adapter must honor the `-c` overrides NBI passes, or the approval posture those pin is lost. See the [installation and migration guidance](#gating-the-experimental-acp-agent-378) for the trust and `PATH` requirements. ACP remains `force-off` by default. 6. **MCP `npx -y` is incompatible with air-gap.** Pre-install the server binary and reference it directly: ```json @@ -623,11 +623,39 @@ Independent of this policy, NBI refuses bypass whenever Claude Code's enterprise ### Gating the experimental ACP agent (#378) -ACP mode runs an external coding agent over the Agent Client Protocol; Codex (via the codex-acp adapter) is the first selectable agent type. ACP mode and Claude mode are mutually exclusive: enabling one disables the other on save, and a hand-edited config that enables both resolves to Claude. The admin controls that apply: +ACP mode runs an external coding agent over the Agent Client Protocol; Codex (via the codex-acp adapter) is the first selectable agent type. ACP mode and Claude mode are mutually exclusive: enabling one disables the other on save, and a hand-edited config that enables both resolves to Claude. + +**Adapter installation and migration.** Install the tested `@zed-industries/codex-acp@0.16.0` in a trusted location before enabling ACP. NBI now launches a preinstalled `codex-acp` executable directly; it no longer downloads an adapter at startup. Deployments upgrading from the previous automatic download must install the adapter. If you used `NBI_ACP_AGENT_COMMAND`, replace any package-runner command with the absolute path to the installed adapter itself, for example `/opt/nbi/acp/bin/codex-acp`; quote paths containing spaces. Do not use `npx`, `npm exec`, `pnpm dlx`, `bunx`, or an `env` wrapper. Making a package runner's path absolute does not prevent it from selecting code through project or parent-directory package lookup. + +For example, on POSIX systems an administrator can provision `/opt/nbi/acp` outside every Jupyter workspace and install the adapter there. Run the installation from a trusted working directory, using trusted npm configuration and permissions to write that prefix. Install Node.js in a trusted location outside the workspace as well, and make its `node` executable available on the Jupyter server's `PATH`. + +```bash +npm install --global --prefix /opt/nbi/acp @zed-industries/codex-acp@0.16.0 +export PATH="/opt/nbi/acp/bin:$PATH" +unset NBI_ACP_AGENT_COMMAND +``` + +This produces `/opt/nbi/acp/bin/codex-acp`; configure the server environment to retain that `PATH` on restart. Unsetting the override selects default discovery. Setting it to an empty or whitespace-only value is an error, not a reset. + +On Windows with Python 3.10 or 3.11, set `NBI_ACP_AGENT_COMMAND` to the absolute path of the installed `.cmd` or `.exe` adapter instead of relying on default discovery, which may select npm's extensionless POSIX shim. ACP startup has not been validated on Windows; verify the selected adapter and runtime in your deployment before enabling the mode. + +Operators own the installed version and its upgrades. NBI neither verifies nor updates the adapter version, whether it finds the executable on `PATH` or uses an explicit override. Its approval and credential protections were tested with codex-acp 0.16.0; other versions may handle the settings differently. Keep the complete installation under trusted control: the adapter, any shim, its interpreter and runtime packages (including platform-specific dependencies), symlink targets, and configured `PATH` directories, together with their parent directories. An absolute adapter path alone does not establish trust in everything it loads or runs. Administrators must also control the mount layout: a workspace subtree mounted separately outside the workspace is not necessarily identifiable as part of it. + +By default, NBI searches only absolute directories on the Jupyter server's `PATH`. It skips empty, relative, and invalid entries and excludes workspace directories, symlinks into the workspace, case and Unicode aliases, and alternate mounts of the workspace root. It also rejects an adapter whose path or resolved target lies in the workspace. NBI launches the resolved target, so a shim that chooses its program from its own name needs an explicit override pointing to the installed `codex-acp` shim's absolute path. The override preserves that name; use it only after trusting the shim and the installation it selects. + +The default adapter receives a `PATH` without workspace directories or directory aliases into the workspace. This affects both its interpreter lookup (such as `env node`) and commands the agent runs. A project's `.venv/bin`, for example, is removed: `python`, `pip`, or `pytest` can resolve to a different installation or be unavailable. Install required runtimes and tools in trusted locations outside the workspace. + +An explicit `NBI_ACP_AGENT_COMMAND` override retains the full inherited server `PATH` and can select a location the default workspace check excludes. It is an administrator trust decision covering the executable and the entire runtime, package, and directory chain described above. Missing executables, relative overrides, and empty or malformed commands block both readiness and startup. Both also check a script adapter's declared interpreter against the launch environment without executing it. For default discovery, the interpreter path and its resolved target must also lie outside the workspace. For example, `node` must be available on the adapter's filtered `PATH` for an `env node` launcher. These checks do not validate a native binary's dependencies or everything a shim loads, and do not guarantee that the agent will start successfully. + +If the Jupyter root is `/home/alice`, default discovery excludes installations such as `/home/alice/.npm-global/bin/codex-acp` and an nvm-managed `node` under `/home/alice/.nvm`, even when the adapter itself is installed elsewhere. Either install both Node.js and the adapter outside that home directory, or explicitly trust the home installation with `NBI_ACP_AGENT_COMMAND=/home/alice/.npm-global/bin/codex-acp`. The override retains the full server `PATH`, including the trusted Node.js installation. Use `unset NBI_ACP_AGENT_COMMAND` to return to default discovery after moving both installations outside the workspace. + +The adapter process starts in NBI's user directory (`~/.jupyter/nbi`). The notebook workspace is passed separately through ACP `session/new` and `session/load`; NBI's MCP workspace tool receives it through `NBI_WORKSPACE_ROOT`, so it does not depend on the adapter or MCP process working directory. A custom adapter command must accept the `-c` settings NBI appends. + +The admin controls that apply: - **`NBI_ACP_MODE_POLICY`** (default `force-off`): whether ACP mode can be enabled at all. While `force-off`, the ACP settings tab is hidden and `enabled` is clamped to false on every config write, so it is the nuclear off switch. - **`NBI_ACP_FULL_ACCESS_POLICY`** (default `force-off`): whether the agent may run tools without asking. While full access is off, NBI pins Codex to `approval_policy = untrusted`, so it asks before anything beyond trusted read-only commands and each request surfaces through NBI's per-tool confirmation. With full access on, NBI pins `approval_policy = never` and `sandbox_mode = workspace-write`, so Codex edits files and runs commands without asking, inside Codex's workspace-write sandbox. That sandbox limits writes to the workspace (the Jupyter server's root directory) and temp directories and keeps outbound network access off, unless Codex's own config widens it. Because Codex never asks under full access, anything the sandbox blocks, such as a `pip install` that needs the network or a write outside the workspace, fails instead of prompting. `user-choice` exposes the "Full access" toggle in the ACP tab; `force-on` requires unattended runs. Like bypass, the value is clamped server-side, so a hand-crafted config write can't relax it. -- **Credentials and model** follow the same env-override locks as Claude: `OPENAI_API_KEY`, `OPENAI_BASE_URL`, and `NBI_ACP_CHAT_MODEL` pin and hide the corresponding fields. When NBI supplies the API key, Codex no longer writes the key to disk or passes it into the environment of the commands it runs: the key is held in memory (`cli_auth_credentials_store = "ephemeral"`) rather than in `auth.json`, so a changed key takes effect the next time the agent starts; Codex's shell environment snapshots are turned off (`features.shell_snapshot = false`); and the key's variable is removed from commands' environment (`shell_environment_policy.exclude`). The next time the agent starts, NBI also removes the `auth.json` and shell snapshot files that earlier versions left in its `codex-home` directory. Removing them does not undo earlier exposure, so rotate any key used with ACP mode before this version. What this does not cover: a key typed into the ACP tab is still saved in NBI's `config.json`, which commands running as the same user can read, so prefer supplying it through `OPENAI_API_KEY`; the key stays in the environment of the codex-acp process (and of the Jupyter server, when it comes from `OPENAI_API_KEY`), where a command running outside Codex's sandbox as the same user can read it; a shell startup file that exports the key and that the login shell reads (`~/.zshenv`, `~/.zprofile`, `~/.bash_profile`, `~/.profile`, or a file they source, such as a `~/.bashrc` sourced from `~/.profile`) sets it again for commands; the `exclude` list replaces, rather than adds to, one set in system Codex configuration; and an adapter other than codex-acp 0.16.0 launched through `NBI_ACP_AGENT_COMMAND` may ignore these settings. With snapshots off, Codex no longer runs commands from a snapshot of an interactive shell, so PATH changes, aliases, and environment managers (conda, pyenv, nvm) set up only in `~/.zshrc` or in a `~/.bashrc` that the login shell does not source do not apply to them. +- **Credentials and model** follow the same env-override locks as Claude: `OPENAI_API_KEY`, `OPENAI_BASE_URL`, and `NBI_ACP_CHAT_MODEL` pin and hide the corresponding fields. When NBI supplies the API key, Codex no longer writes the key to disk or passes it into the environment of the commands it runs: the key is held in memory (`cli_auth_credentials_store = "ephemeral"`) rather than in `auth.json`, so a changed key takes effect the next time the agent starts; Codex's shell environment snapshots are turned off (`features.shell_snapshot = false`); and the key's variable is removed from commands' environment (`shell_environment_policy.exclude`). The next time the agent starts, NBI also removes the `auth.json` and shell snapshot files that earlier versions left in its `codex-home` directory. Removing them does not undo earlier exposure, so rotate any key used with ACP mode before this version. What this does not cover: a key typed into the ACP tab is still saved in NBI's `config.json`, which commands running as the same user can read, so prefer supplying it through `OPENAI_API_KEY`; the key stays in the environment of the codex-acp process (and of the Jupyter server, when it comes from `OPENAI_API_KEY`), where a command running outside Codex's sandbox as the same user can read it; a shell startup file that exports the key and that the login shell reads (`~/.zshenv`, `~/.zprofile`, `~/.bash_profile`, `~/.profile`, or a file they source, such as a `~/.bashrc` sourced from `~/.profile`) sets it again for commands; the `exclude` list replaces, rather than adds to, one set in system Codex configuration; and an adapter other than the tested codex-acp 0.16.0 may ignore these settings, whether discovered on `PATH` or selected through `NBI_ACP_AGENT_COMMAND`. With snapshots off, Codex no longer runs commands from a snapshot of an interactive shell, so PATH changes, aliases, and environment managers (conda, pyenv, nvm) set up only in `~/.zshrc` or in a `~/.bashrc` that the login shell does not source do not apply to them. How the pins are applied: NBI passes `approval_policy` (and, with full access, `sandbox_mode`) to codex via the codex-acp binary's `-c key=value` command-line override, which takes precedence over Codex's user and project config (`$CODEX_HOME/config.toml` and a workspace's `.codex/config.toml`). System and managed Codex configuration is still read, and managed configuration outranks NBI's pins: on Linux and macOS that is `/etc/codex` (including `managed_config.toml` and `requirements.toml`), plus MDM-delivered configuration on macOS. A managed `requirements.toml` that disallows `workspace-write` makes Codex fall back to read-only, which brings back refused edits under full access. When an API key is configured, NBI also runs codex with an isolated `CODEX_HOME` under the NBI user directory, so neither the workspace's nor the user's `~/.codex` config is read (this isolates configuration only; the agent still operates on workspace files, subject to the approval and sandbox pins). With ChatGPT (OAuth) auth, codex uses the user's own `~/.codex`. The `-c` pins still override its top-level `approval_policy` and `sandbox_mode`, so a broader `sandbox_mode` set there, such as `danger-full-access`, is narrowed to `workspace-write` while full access is on. The rest of that directory still applies, including `[sandbox_workspace_write]` settings like `writable_roots` and `network_access` and any command rules, and a user who controls that home directory is also the account the agent runs as. On shared or locked-down deployments, prefer API-key auth (so `CODEX_HOME` is isolated) and keep `NBI_ACP_MODE_POLICY=force-off` unless the agent is explicitly needed. @@ -833,7 +861,7 @@ A misconfigured NBI fails at the far end of a chat turn, where the user sees a g It is deliberately **not** gated behind the perf-diagnostics setting the way the two `/perf` routes are: a user who cannot tell "the admin has not set this up" from "I did something wrong" needs the answer whether or not diagnostics are enabled, and an admin verifying a rollout needs it before anyone has opened the settings dialog. It reports configuration state the authenticated user can already read from the capabilities response and the settings panel, so it exposes nothing new. Exception messages are never included, only exception class names, because provider errors routinely carry the base URL and sometimes the key. -**What runs by default (bills nothing):** which path is serving chat; for the native provider path, whether a provider and model are selected, whether the provider's model name was left blank, whether the provider is registered (a provider removed by `disabled_providers` looks identical to a typo from the user's side, so the remedy names both), whether the model list is reachable and non-empty, and whether the selected model id is one the endpoint actually serves; for Claude mode, whether the CLI resolves **and answers `--version`** (a CLI that resolves and then fails to start is the usual cause of a turn hanging on "Thinking...", which a path check alone would call healthy) and whether an Anthropic key is present; for ACP mode, the resolved adapter command, whether `npx` is available when the adapter needs it, and whether an agent key is present. +**What runs by default (bills nothing):** which path is serving chat; for the native provider path, whether a provider and model are selected, whether the provider's model name was left blank, whether the provider is registered (a provider removed by `disabled_providers` looks identical to a typo from the user's side, so the remedy names both), whether the model list is reachable and non-empty, and whether the selected model id is one the endpoint actually serves; for Claude mode, whether the CLI resolves **and answers `--version`** (a CLI that resolves and then fails to start is the usual cause of a turn hanging on "Thinking...", which a path check alone would call healthy) and whether an Anthropic key is present; for ACP mode, whether the adapter command resolves to an allowed absolute executable and its declared script interpreter is available in the launch environment, using the same rules as startup, and whether an agent key is present. The ACP check reads a script's interpreter line but does not execute the adapter, verify its version, or download packages; it cannot verify a native binary's dependencies or the internals of an administrator-selected shim. Missing credentials are reported as a **warning, not a blocker**, in both agent modes. The Claude CLI and the ACP agent can each hold a subscription or OAuth login that NBI cannot see, and calling that configuration broken would be wrong. diff --git a/notebook_intelligence/acp_agent.py b/notebook_intelligence/acp_agent.py index f70e7057..c92c156b 100644 --- a/notebook_intelligence/acp_agent.py +++ b/notebook_intelligence/acp_agent.py @@ -55,12 +55,14 @@ from notebook_intelligence import perf from notebook_intelligence.acp_registry import ( AcpAgentSpec, + acp_adapter_path, codex_approval_args, codex_auth_args, codex_model_args, resolve_acp_agent, resolve_acp_agent_command, ) +from notebook_intelligence.acp_runtime import validate_acp_adapter_runtime from notebook_intelligence.base_chat_participant import BaseChatParticipant from notebook_intelligence.claude_sessions import CONTROL_SLASH_COMMANDS, NBI_CONTEXT_PREFIX from notebook_intelligence.util import BIDI_CONTROL_CODEPOINTS, ThreadSafeWebSocketConnector, get_jupyter_root_dir @@ -790,15 +792,18 @@ def _mcp_servers(self) -> list: """The NBI MCP server config passed to every session create/load.""" if self._force_safe_mode: return [] - # Launched by file path rather than ``-m``: the agent starts this server - # outside its sandbox with the workspace as cwd, and ``-m`` puts the cwd - # first on sys.path, so a ``notebook_intelligence`` package the agent - # wrote into the workspace would run in place of this one. + # Launch by absolute file path: the adapter controls this server's cwd, + # and ``-m`` could import a workspace package instead of NBI's server. from notebook_intelligence import acp_mcp_server + # Pass the workspace explicitly instead of trusting the adapter's cwd. return [ schema.McpServerStdio( name="nbi", command=sys.executable, - args=[os.path.abspath(acp_mcp_server.__file__)], env=[], + args=[os.path.abspath(acp_mcp_server.__file__)], + env=[schema.EnvVariable( + name=acp_mcp_server.WORKSPACE_ROOT_ENV, + value=get_jupyter_root_dir() or "", + )], ) ] @@ -852,25 +857,28 @@ async def _serve(self): self._shutdown = asyncio.Event() workdir = get_jupyter_root_dir() spec = self.agent_spec - env = self._child_env(spec) - cmd = list(resolve_acp_agent_command(spec)) - if spec.id == "codex": - # First, so they stay clear of the pins that follow. - cmd += codex_auth_args(spec.api_key_env if self._api_key(spec) else "") - cmd += codex_approval_args( - False - if self._force_safe_mode - else bool(self.acp_settings.get("full_access", False)) - ) - # acp_settings already folds in the OPENAI_BASE_URL / - # NBI_ACP_CHAT_MODEL env overrides (ACP_SETTINGS_OVERRIDES), - # so these -c flags are the single delivery path to codex. - cmd += codex_model_args(self.acp_settings) - log.info("Starting ACP agent (%s): %s", spec.id, " ".join(cmd)) try: + env = self._child_env(spec) + cmd = resolve_acp_agent_command(spec, workspace_root=workdir) + launch_cwd = self._adapter_cwd() + validate_acp_adapter_runtime( + cmd, env_path=env.get("PATH", os.defpath), cwd=launch_cwd, + workspace_root=workdir if "NBI_ACP_AGENT_COMMAND" not in os.environ else None, + ) + if spec.id == "codex": + # First, so they stay clear of the pins that follow. + cmd += codex_auth_args(spec.api_key_env if self._api_key(spec) else "") + cmd += codex_approval_args( + False + if self._force_safe_mode + else bool(self.acp_settings.get("full_access", False)) + ) + # acp_settings folds in the environment overrides too. + cmd += codex_model_args(self.acp_settings) + log.info("Starting ACP agent (%s): %s", spec.id, " ".join(cmd)) self._proc = await asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, cwd=workdir, env=env, + stderr=asyncio.subprocess.PIPE, cwd=launch_cwd, env=env, ) self._stderr_task = asyncio.create_task(self._drain_stderr()) self._client = _NbiAcpClient(self) @@ -907,9 +915,26 @@ def _api_key(self, spec: AcpAgentSpec) -> str: return (self.acp_settings.get("api_key") or "").strip() \ or os.environ.get(spec.api_key_env, "").strip() + def _adapter_cwd(self) -> str: + """The directory the adapter process is started in. + + Keep the process outside the workspace; session/new and session/load + receive the actual workspace explicitly. This directory can have a + shared ancestor, so cwd alone is not a package-resolution boundary. + resolve_acp_agent_command instead selects a preinstalled executable + by absolute path, without invoking npm at startup. + """ + cwd = self._host.nbi_config.nbi_user_dir + os.makedirs(cwd, exist_ok=True) + return cwd + def _child_env(self, spec: AcpAgentSpec) -> dict: env = {k: v for k, v in os.environ.items() if k != "CLAUDECODE" and not k.startswith("CLAUDE_CODE_")} + if "NBI_ACP_AGENT_COMMAND" not in os.environ: + # npm-installed adapters use /usr/bin/env node. Resolving the + # adapter alone must not leave its interpreter workspace-controlled. + env["PATH"] = acp_adapter_path(get_jupyter_root_dir()) if spec.id == "codex": # Whatever the sign-in, so old copies do not outlive a switch away # from an API key. diff --git a/notebook_intelligence/acp_mcp_server.py b/notebook_intelligence/acp_mcp_server.py index 9ce03a4a..ff865525 100644 --- a/notebook_intelligence/acp_mcp_server.py +++ b/notebook_intelligence/acp_mcp_server.py @@ -6,10 +6,13 @@ MCP server Claude mode uses (``create_sdk_mcp_server``), so NBI's tools must run as a real subprocess for the agent-mode framework (issue #378, Phase 1). This module is that subprocess: newline-delimited JSON-RPC over stdio, launched as -``python `` with its cwd set to the JupyterLab working -directory by the participant. It runs by path rather than ``-m`` so that a -package planted in that working directory cannot shadow it (see +``python `` by the agent. The agent chooses the MCP +process cwd; this server does not depend on it. It runs by path rather than +``-m`` so that a package planted in the workspace cannot shadow it (see ``AcpAgentClient._mcp_servers``), which is also why it imports only the stdlib. +The workspace root arrives in the ``NBI_WORKSPACE_ROOT`` environment variable +the participant sets on the server entry, independently of the adapter's +launch directory and the agent's MCP launch behavior. Phase 1 ships one safe, dependency-free tool (``nbi_workspace_root``) so the end-to-end MCP path is exercised; richer Jupyter UI tools that need the live @@ -22,6 +25,10 @@ PROTOCOL_VERSION = "2025-06-18" +# Set by the participant on the McpServerStdio entry; the cwd is the fallback +# for a launch that predates it. +WORKSPACE_ROOT_ENV = "NBI_WORKSPACE_ROOT" + TOOLS = [ { "name": "nbi_workspace_root", @@ -49,7 +56,8 @@ def _error(mid, code, message): def _call_tool(name, arguments): if name == "nbi_workspace_root": - return {"content": [{"type": "text", "text": os.getcwd()}]} + root = os.environ.get(WORKSPACE_ROOT_ENV) or os.getcwd() + return {"content": [{"type": "text", "text": root}]} raise KeyError(name) diff --git a/notebook_intelligence/acp_registry.py b/notebook_intelligence/acp_registry.py index f9e0ee19..aa54e1c7 100644 --- a/notebook_intelligence/acp_registry.py +++ b/notebook_intelligence/acp_registry.py @@ -15,7 +15,11 @@ import base64 import os +import shlex +import shutil +import stat from dataclasses import dataclass +from pathlib import Path from typing import Optional # Pinned in Phase 1 (the version the spike validated); revisit per release. @@ -55,7 +59,8 @@ class AcpAgentSpec: id: str label: str description: str - package: str # npx package spec that launches the ACP adapter + package: str # pinned package to install before starting the adapter + executable: str # preinstalled adapter binary, resolved before spawning icon_url: str api_key_env: str # env var the agent reads its API key from auth_method: str # preferred ACP auth method id when a key is present @@ -67,6 +72,7 @@ class AcpAgentSpec: label="Codex", description="OpenAI Codex (via the Agent Client Protocol)", package=CODEX_ACP_PACKAGE, + executable="codex-acp", icon_url=CODEX_AGENT_ICON_URL, api_key_env="OPENAI_API_KEY", auth_method="openai-api-key", @@ -81,18 +87,136 @@ def resolve_acp_agent(agent_id: Optional[str]) -> AcpAgentSpec: return ACP_AGENTS.get(agent_id or "", None) or ACP_AGENTS[DEFAULT_ACP_AGENT] -def resolve_acp_agent_command(spec: AcpAgentSpec) -> list[str]: - """The command that launches the agent's ACP adapter. - - ``NBI_ACP_AGENT_COMMAND`` overrides (shell-split); otherwise run the - spec's pinned package via ``npx``. Kept separate from the Claude CLI - resolver because the adapters are npm packages, not binaries on PATH. +@dataclass(frozen=True) +class _WorkspaceBoundary: + paths: tuple[Path, Path] + identity: tuple[int, int] + + def external_target(self, path: Path) -> Optional[Path]: + """Resolve a path only when neither its spelling nor identity is in the workspace. + + ``resolve`` does not normalize case or Unicode on every filesystem, and + a mount alias can name the same directory without using a symlink. + Compare each ancestor's filesystem identity as well as both path + spellings. Unknown identities fail closed; callers skip that entry. + """ + target = path.resolve(strict=True) + for candidate in (path, target): + if any(candidate.is_relative_to(root) for root in self.paths): + return None + for ancestor in (candidate, *candidate.parents): + info = ancestor.stat() + if not info.st_ino: + raise ValueError("ACP PATH entry has no stable filesystem identity") + if (info.st_dev, info.st_ino) == self.identity: + return None + return target + + +def _workspace_boundary(workspace_root: str) -> _WorkspaceBoundary: + """Require a known workspace before deciding that any default PATH is safe.""" + try: + if not workspace_root: + raise ValueError("the workspace root is unset") + path = Path(os.path.abspath(workspace_root)) + target = path.resolve(strict=True) + info = target.stat() + if not stat.S_ISDIR(info.st_mode) or not info.st_ino: + raise ValueError("the workspace root is not an identifiable directory") + return _WorkspaceBoundary((path, target), (info.st_dev, info.st_ino)) + except (OSError, RuntimeError, ValueError) as exc: + raise ValueError( + "ACP workspace root must be an existing, inspectable directory before " + f"resolving a default adapter: {workspace_root!r}" + ) from exc + + +def _adapter_directories(boundary: _WorkspaceBoundary) -> list[str]: + directories = [] + for directory in os.environ.get("PATH", os.defpath).split(os.pathsep): + try: + if not Path(directory).is_absolute(): + continue + target = boundary.external_target(Path(os.path.abspath(directory))) + # A symlink can resolve to a directory whose name contains the PATH + # separator. Serializing that target would create additional, + # unchecked search entries for the adapter's interpreter. + if target is not None and os.pathsep not in str(target) and target.is_dir(): + directories.append(str(target)) + except (OSError, RuntimeError, ValueError): + # Missing, malformed, unreadable, or looping entries cannot establish + # a trusted location. Keep looking so one bad entry cannot hide a + # later trusted installation. + continue + return directories + + +def acp_adapter_path(workspace_root: str) -> str: + """PATH for default adapters and their interpreters, excluding workspace aliases.""" + return os.pathsep.join(_adapter_directories(_workspace_boundary(workspace_root))) + + +def validate_acp_executable_path(executable: str, workspace_root: str) -> None: + """Require a default adapter interpreter to stay outside the workspace. + + Check the original spelling as well as the resolved target, including + filesystem identity aliases. Explicit administrator overrides do not use + this default-launch restriction. + """ + try: + path = Path(executable) + if not path.is_absolute(): + raise ValueError("the executable path is not absolute") + if _workspace_boundary(workspace_root).external_target(path) is None: + raise ValueError("the executable path or its target is in the workspace") + except (OSError, RuntimeError, ValueError) as exc: + raise ValueError( + "ACP executable must be an existing, inspectable absolute path outside " + f"the workspace: {executable!r}. Install the adapter and its interpreter " + "in a trusted location outside the workspace." + ) from exc + + +def resolve_acp_agent_command(spec: AcpAgentSpec, *, workspace_root: Optional[str] = None) -> list[str]: + """Resolve a preinstalled adapter independently of the process cwd. + + Never bootstrap with ``npx`` here: npm can select packages, project + configuration, and executable paths from untrusted ancestors even when + its cwd is a private directory. Installation and PATH are administrator + responsibilities; neither startup nor readiness installs a package. """ - override = os.environ.get("NBI_ACP_AGENT_COMMAND", "").strip() - if override: - import shlex - return shlex.split(override) - return ["npx", "-y", spec.package] + override = os.environ.get("NBI_ACP_AGENT_COMMAND") + if override is not None: + command = shlex.split(override) + if not command or not Path(command[0]).is_absolute(): + raise ValueError("NBI_ACP_AGENT_COMMAND must name an absolute executable path") + resolved = shutil.which(command[0]) + if resolved is None: + raise FileNotFoundError(f"ACP adapter '{command[0]}' was not found or is not executable") + return [resolved, *command[1:]] + + if workspace_root is None: + from notebook_intelligence.util import get_jupyter_root_dir + workspace_root = get_jupyter_root_dir() + boundary = _workspace_boundary(workspace_root) + + # Check absolute candidates individually. On Windows, which(name, path=...) + # can prepend the current directory even when PATH contains only absolute + # entries. A candidate with a directory component bypasses that search. + for directory in _adapter_directories(boundary): + try: + resolved = shutil.which(os.path.join(directory, spec.executable)) + if resolved is not None: + target = boundary.external_target(Path(os.path.abspath(resolved))) + if target is not None: + return [str(target)] + except (OSError, RuntimeError, ValueError): + continue + raise FileNotFoundError( + f"ACP adapter '{spec.executable}' was not found on an absolute PATH entry outside the workspace. " + f"Install {spec.package} in a trusted location, or set " + "NBI_ACP_AGENT_COMMAND to its absolute executable path." + ) def codex_approval_args(full_access: bool) -> list[str]: diff --git a/notebook_intelligence/acp_runtime.py b/notebook_intelligence/acp_runtime.py new file mode 100644 index 00000000..c84966e3 --- /dev/null +++ b/notebook_intelligence/acp_runtime.py @@ -0,0 +1,112 @@ +# Copyright (c) Mehmet Bektas + +"""Nonexecuting checks for an ACP adapter's declared interpreter.""" + +import os +import re +import shutil +import stat +from typing import Optional + +from notebook_intelligence.acp_registry import validate_acp_executable_path + + +_SHEBANG_LIMIT = 4096 + + +def _resolve_interpreter(program: str, env_path: str, cwd: Optional[str]) -> Optional[str]: + if cwd is not None and not os.path.isabs(cwd): + cwd = os.path.join(os.getcwd(), cwd) + if os.path.isabs(program): + return shutil.which(program) + if os.path.dirname(program): + if cwd is None: + raise ValueError("The adapter working directory is needed to check its relative interpreter") + return shutil.which(os.path.join(cwd, program)) + for directory in env_path.split(os.pathsep): + if not os.path.isabs(directory): + if cwd is None: + raise ValueError("The adapter working directory is needed to check its relative PATH entries") + directory = os.path.join(cwd, directory) + # Absolute candidates avoid shutil.which's implicit current-directory + # search on Windows and model env's lookup from the adapter's cwd. + # Keep symlink/.. traversal intact: env and the kernel do not collapse + # it lexically, so doing so here could validate a different executable. + resolved = shutil.which(os.path.join(directory, program)) + if resolved is not None: + return resolved + return None + + +def validate_acp_adapter_runtime( + command: list[str], *, env_path: str, cwd: Optional[str] = None, + workspace_root: Optional[str] = None, +) -> None: + """Check a shebang using the same PATH and cwd as the eventual launch. + + Read at most one bounded line, never execute the adapter or interpreter, + and report unsupported env syntax instead of guessing how env expands it. + This checks the declared interpreter's availability, not adapter behavior + or the internal dependencies of a native binary or administrator shim. + Default startup supplies workspace_root to exclude interpreters in the + workspace, including executable symlinks. Explicit administrator overrides + leave it unset and retain their trusted runtime choice. + """ + executable = command[0] + try: + flags = os.O_RDONLY | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_BINARY", 0) + with os.fdopen(os.open(executable, flags), "rb") as adapter: + if not stat.S_ISREG(os.fstat(adapter.fileno()).st_mode): + raise ValueError(f"ACP adapter '{executable}' is not a regular file") + first_line = adapter.readline(_SHEBANG_LIMIT + 1) + except OSError as exc: + raise ValueError( + f"Cannot read ACP adapter '{executable}' to check its interpreter ({type(exc).__name__}). " + "Check the installed executable's read permissions and filesystem links." + ) from exc + if not first_line.startswith(b"#!"): + return + if len(first_line) > _SHEBANG_LIMIT or any( + char < 0x20 and char not in (0x09, 0x0A) for char in first_line + ): + raise ValueError("ACP adapter has an invalid or overlong interpreter line; reinstall the adapter") + try: + shebang = re.split(r"[ \t]+", first_line[2:].decode("utf-8").strip(" \t\n"), maxsplit=1) + except UnicodeError as exc: + raise ValueError("ACP adapter's interpreter line is not valid UTF-8; reinstall the adapter") from exc + if not shebang or not os.path.isabs(shebang[0]): + raise ValueError("ACP adapter must declare an absolute interpreter path; reinstall the adapter") + interpreter = shebang[0] + if shutil.which(interpreter) is None: + raise ValueError( + f"ACP adapter requires interpreter '{interpreter}', which is missing or not executable. " + "Install the declared interpreter or reinstall the adapter." + ) + if workspace_root is not None: + validate_acp_executable_path(interpreter, workspace_root) + if os.path.basename(interpreter) != "env": + return + + arguments = re.split(r"[ \t]+", shebang[1]) if len(shebang) == 2 else [] + # With no -S, kernels pass the remaining shebang text as one argument. + # For -S support the common plain command/flag form, but do not attempt + # GNU/BSD env's quoting, expansion, assignment, or option grammars. + if arguments and arguments[0] == "-S": + arguments = arguments[1:] + supported = not any(char in shebang[1] for char in "\\\"'$#") + else: + supported = len(arguments) == 1 + if not supported or not arguments or arguments[0].startswith("-") or "=" in arguments[0]: + raise ValueError( + "Cannot validate ACP adapter's env interpreter line. Use an installed adapter with " + "a simple '#!/usr/bin/env node' or '#!/usr/bin/env -S node ' line." + ) + program = arguments[0] + resolved_program = _resolve_interpreter(program, env_path, cwd) + if resolved_program is None: + raise ValueError( + f"ACP adapter requires interpreter '{program}', which is missing or not executable on " + "the adapter PATH. Install it in a trusted directory available to the adapter." + ) + if workspace_root is not None: + validate_acp_executable_path(resolved_program, workspace_root) diff --git a/notebook_intelligence/extension.py b/notebook_intelligence/extension.py index f1241133..f4b49afc 100644 --- a/notebook_intelligence/extension.py +++ b/notebook_intelligence/extension.py @@ -2526,7 +2526,7 @@ async def get(self): return try: # list_sessions may cold-start the agent subprocess (up to a - # minute through npx); keep the IO loop free while it does. + # minute to initialize); keep the IO loop free while it does. loop = asyncio.get_event_loop() sessions, error = await loop.run_in_executor( None, participant.list_sessions diff --git a/notebook_intelligence/readiness.py b/notebook_intelligence/readiness.py index d04de59d..433daa45 100644 --- a/notebook_intelligence/readiness.py +++ b/notebook_intelligence/readiness.py @@ -536,64 +536,75 @@ def _claude_rows(nbi_config: Any, pool: concurrent.futures.ThreadPoolExecutor) - def _acp_rows(nbi_config: Any) -> list: from notebook_intelligence.acp_registry import ( + acp_adapter_path, resolve_acp_agent, resolve_acp_agent_command, ) + from notebook_intelligence.acp_runtime import validate_acp_adapter_runtime + from notebook_intelligence.util import get_jupyter_root_dir rows = [] settings = nbi_config.acp_settings or {} spec = resolve_acp_agent(settings.get("agent")) - command = resolve_acp_agent_command(spec) - - rows.append( - _row( - "acp.agent", - "acp", - LEVEL_OK, - "ACP agent", - f"{spec.label}, launched as: {' '.join(command)}", + try: + # Launch and readiness share the same absolute executable resolution. + # In particular, neither accepts a command relative to the server cwd. + workspace_root = get_jupyter_root_dir() + command = resolve_acp_agent_command(spec, workspace_root=workspace_root) + env_path = ( + os.environ.get("PATH", os.defpath) + if "NBI_ACP_AGENT_COMMAND" in os.environ + else acp_adapter_path(workspace_root) ) - ) - - # Whatever argv[0] is, it has to exist and be executable. Special-casing - # the literal "npx" would let an NBI_ACP_AGENT_COMMAND override point at a - # nonexistent binary and still report Ready. - if command: - binary = command[0] - from shutil import which - - resolved = which(binary) if not os.path.sep in binary else ( - binary if os.access(binary, os.X_OK) else None + validate_acp_adapter_runtime( + command, env_path=env_path, cwd=getattr(nbi_config, "nbi_user_dir", None), + workspace_root=workspace_root if "NBI_ACP_AGENT_COMMAND" not in os.environ else None, ) - if resolved is None: - hint = ( - "The default adapter is an npm package, so this usually means " - "Node.js is not installed in the environment running " - "JupyterLab. " - if binary == "npx" - else "" + except (ValueError, OSError, RuntimeError) as exc: + rows.append( + _row( + "acp.agent", + "acp", + LEVEL_OK, + "ACP agent", + spec.label, ) - rows.append( - _row( - "acp.runtime", - "acp", - LEVEL_BLOCKED, - "Adapter runtime", - f"'{binary}' was not found or is not executable.", - hint + "Install it, or set NBI_ACP_AGENT_COMMAND to a " - "command that does exist.", - ) + ) + rows.append( + _row( + "acp.runtime", + "acp", + LEVEL_BLOCKED, + "Adapter runtime", + str(exc), + f"Preinstall the pinned adapter package {spec.package} in a " + "trusted location outside the workspace and put its absolute " + "executable directory on PATH, " + "or set NBI_ACP_AGENT_COMMAND to the absolute path of an " + "installed adapter executable. Install its required interpreter " + "on the adapter PATH; default discovery excludes workspace directories. " + "Do not use a package runner such as npx. Relative commands are not supported.", ) - else: - rows.append( - _row( - "acp.runtime", - "acp", - LEVEL_OK, - "Adapter runtime", - f"'{binary}' is available.", - ) + ) + else: + rows.append( + _row( + "acp.agent", + "acp", + LEVEL_OK, + "ACP agent", + f"{spec.label}, launched as: {' '.join(command)}", ) + ) + rows.append( + _row( + "acp.runtime", + "acp", + LEVEL_OK, + "Adapter runtime", + f"'{command[0]}' and its declared interpreter are available.", + ) + ) source = _credential_source(settings.get("api_key"), spec.api_key_env) if source: diff --git a/src/components/settings-panel.tsx b/src/components/settings-panel.tsx index 75142a47..38645744 100644 --- a/src/components/settings-panel.tsx +++ b/src/components/settings-panel.tsx @@ -2164,9 +2164,17 @@ function SettingsPanelComponentAcp(props: any) { > Agent Client Protocol - . Requires npx available on the server. ACP mode - and Claude mode are mutually exclusive: enabling one turns the - other off. + . Requires a preinstalled codex-acp adapter ( + @zed-industries/codex-acp@0.16.0). See the{' '} + + administrator installation guide + + . ACP mode and Claude mode are mutually exclusive: enabling one + turns the other off.
diff --git a/tests/test_acp_agent.py b/tests/test_acp_agent.py index 7291dcac..7e8781ee 100644 --- a/tests/test_acp_agent.py +++ b/tests/test_acp_agent.py @@ -12,6 +12,7 @@ import json import os import subprocess +import sys from types import SimpleNamespace import pytest @@ -28,6 +29,24 @@ from notebook_intelligence.api import ChatResponse, ResponseStreamDataType +@pytest.fixture(autouse=True) +def installed_adapter(tmp_path, monkeypatch): + """Keep launch unit tests independent of adapters installed on the host.""" + import notebook_intelligence.util as util + + binary_dir = tmp_path / "trusted-bin" + binary_dir.mkdir() + adapter = binary_dir / "codex-acp" + adapter.write_text("#!/bin/sh\nexit 99\n") + adapter.chmod(0o700) + monkeypatch.setenv("PATH", str(binary_dir) + os.pathsep + os.environ.get("PATH", "")) + monkeypatch.delenv("NBI_ACP_AGENT_COMMAND", raising=False) + workspace = tmp_path / "default-workspace" + workspace.mkdir() + monkeypatch.setattr(util, "_jupyter_root_dir", str(workspace)) + return adapter + + class FakeResponse(ChatResponse): """Captures streamed data and supports the user-input signal round trip.""" @@ -894,8 +913,7 @@ def test_initialize_does_not_advertise_fs(self, tmp_path, monkeypatch): import notebook_intelligence.acp_agent as mod - # A missed patch must fail fast rather than launch a real agent. - monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", "nbi-test-must-not-spawn") + # The installed_adapter fixture exits immediately if a patch is missed. monkeypatch.delenv("OPENAI_API_KEY", raising=False) captured = {} @@ -1008,14 +1026,14 @@ def test_args_come_exactly_with_the_isolated_codex_home( monkeypatch.setenv("OPENAI_API_KEY", env_key) captured = self._launch(tmp_path, {"api_key": settings_key}) codex_home = str(tmp_path / "codex-home") - assert (captured["cmd"][3:9] == self.AUTH_ARGS) is supplied + assert (captured["cmd"][1:7] == self.AUTH_ARGS) is supplied assert (captured["env"].get("CODEX_HOME") == codex_home) is supplied def test_auth_args_come_before_the_approval_pin(self, tmp_path, monkeypatch): monkeypatch.setenv("OPENAI_API_KEY", "sk-env") captured = self._launch(tmp_path, {}) - assert captured["cmd"][3:9] == self.AUTH_ARGS - assert captured["cmd"][9:11] == ["-c", 'approval_policy="untrusted"'] + assert captured["cmd"][1:7] == self.AUTH_ARGS + assert captured["cmd"][7:9] == ["-c", 'approval_policy="untrusted"'] def test_saved_key_copies_left_by_earlier_versions_are_removed(self, tmp_path, monkeypatch): monkeypatch.delenv("OPENAI_API_KEY", raising=False) @@ -1083,7 +1101,8 @@ def test_an_unreadable_snapshot_directory_does_not_stop_the_launch(self, tmp_pat captured = self._launch(tmp_path, {}) finally: snapshots.chmod(0o755) - assert captured["cmd"][:3] == ["npx", "-y", "@zed-industries/codex-acp@0.16.0"] + assert os.path.isabs(captured["cmd"][0]) + assert os.path.basename(captured["cmd"][0]) == "codex-acp" assert not (codex_home / "auth.json").exists() assert "Could not list old Codex shell snapshots" in caplog.text @@ -1160,22 +1179,21 @@ def test_force_on_turns_full_access_on(self, mock_nbi_config): cmd = _captured_launch_cmd(mock_nbi_config) assert cmd[-4:] == self._FULL_ACCESS_PINS - def test_agent_command_override_keeps_the_pins(self, tmp_path, monkeypatch): + def test_agent_command_override_keeps_the_pins(self, tmp_path, monkeypatch, installed_adapter): """An admin-pinned adapter binary must not lose the approval and sandbox posture.""" - monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", "/opt/codex-acp --verbose") + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", f"{installed_adapter} --verbose") cmd = _captured_launch_cmd(SimpleNamespace( acp_settings={"enabled": True, "agent": "codex", "full_access": True}, nbi_user_dir=str(tmp_path), )) - assert cmd[:2] == ["/opt/codex-acp", "--verbose"] + assert cmd[:2] == [str(installed_adapter), "--verbose"] assert cmd[-4:] == self._FULL_ACCESS_PINS class TestNbiMcpServerLaunch: - """Codex starts NBI's MCP server outside its sandbox with the workspace as - cwd, and full access lets the agent write that workspace, so nothing - written there may run in place of the server.""" + """The adapter controls the MCP server cwd. Workspace files must never + shadow the server, regardless of which cwd the adapter supplies.""" @staticmethod def _server(): @@ -1211,6 +1229,373 @@ def test_a_package_planted_in_the_workspace_does_not_run(self, tmp_path): reply = json.loads(proc.stdout.splitlines()[0]) assert reply["result"]["serverInfo"]["name"] == "nbi" + def test_workspace_root_travels_in_the_server_env(self, monkeypatch, tmp_path): + import notebook_intelligence.acp_agent as mod + from notebook_intelligence import acp_mcp_server + + workspace = tmp_path / "workspace" + workspace.mkdir() + monkeypatch.setattr(mod, "get_jupyter_root_dir", lambda: str(workspace)) + server = self._server() + + env = {e.name: e.value for e in server.env} + assert env[acp_mcp_server.WORKSPACE_ROOT_ENV] == str(workspace) + + def test_nbi_workspace_root_answers_from_the_env_not_the_cwd(self, tmp_path): + """The adapter (and so this server) no longer runs in the workspace, + so the tool must not report whatever directory it happens to be in.""" + from notebook_intelligence import acp_mcp_server + + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir() + server = self._server() + env = dict(os.environ, **{acp_mcp_server.WORKSPACE_ROOT_ENV: "/srv/workspace"}) + request = json.dumps({ + "jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "nbi_workspace_root", "arguments": {}}, + }) + + proc = subprocess.run( + [server.command, *server.args], + input=request + "\n", cwd=elsewhere, env=env, + capture_output=True, text=True, timeout=30, + ) + + reply = json.loads(proc.stdout.splitlines()[0]) + assert reply["result"]["content"][0]["text"] == "/srv/workspace" + + +class TestAdapterLaunchCwd: + """Startup must ignore npm projects even above the user's private home.""" + + @staticmethod + def _captured_launch(monkeypatch, tmp_path): + import notebook_intelligence.acp_agent as mod + + workspace = tmp_path / "workspace" + workspace.mkdir() + user_dir = tmp_path / "nbi-user-dir" + monkeypatch.setattr(mod, "get_jupyter_root_dir", lambda: str(workspace)) + host = SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace( + acp_settings={"api_key": ""}, nbi_user_dir=str(user_dir) + ), + ) + client = mod.AcpAgentClient(host) + captured = {} + + async def fake_exec(*cmd, **kw): + captured["cmd"] = list(cmd) + captured["cwd"] = kw.get("cwd") + raise RuntimeError("test: launch captured, subprocess intentionally not started") + + monkeypatch.setattr(mod.asyncio, "create_subprocess_exec", fake_exec) + asyncio.run(client._serve()) + assert "cmd" in captured, "_serve returned before launching the adapter" + return captured, str(workspace), str(user_dir) + + def test_adapter_starts_in_the_nbi_user_dir_not_the_workspace(self, monkeypatch, tmp_path): + captured, workspace, user_dir = self._captured_launch(monkeypatch, tmp_path) + + assert captured["cwd"] == user_dir + assert os.path.realpath(captured["cwd"]) != os.path.realpath(workspace) + assert os.path.isdir(user_dir), "the launch cwd must exist before spawn" + + def test_workspace_still_reaches_the_agent_as_the_session_cwd(self, monkeypatch, tmp_path): + """Moving the process cwd must not move the agent's workspace.""" + import notebook_intelligence.acp_agent as mod + + workspace = tmp_path / "workspace" + workspace.mkdir() + monkeypatch.setattr(mod, "get_jupyter_root_dir", lambda: str(workspace)) + client = mod.AcpAgentClient(SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace( + acp_settings={}, nbi_user_dir=str(tmp_path / "nbi-user-dir") + ), + )) + calls = {} + + class FakeConn: + async def new_session(self, **kw): + calls.update(kw) + return SimpleNamespace(session_id="s-1") + + client._conn = FakeConn() + asyncio.run(client._new_session_coro()) + + assert calls["cwd"] == str(workspace) + assert client._session_id == "s-1" + + @pytest.mark.parametrize("installed", [True, False]) + def test_shared_ancestor_cannot_choose_the_adapter( + self, tmp_path, monkeypatch, installed_adapter, installed + ): + import notebook_intelligence.acp_agent as mod + from notebook_intelligence.acp_registry import ACP_AGENTS + + package = ACP_AGENTS["codex"].package + name, _, version = package.rpartition("@") + bin_name = name.rpartition("/")[2] + workspace = tmp_path / "shared" + workspace.mkdir(mode=0o1777) + home = workspace / "private-home" + home.mkdir(mode=0o700) + user_dir = home / ".jupyter" / "nbi" + user_dir.mkdir(parents=True) + result = tmp_path / "adapter-result.json" + marker = tmp_path / "untrusted-adapter-ran" + planted = workspace / "node_modules" / name + planted.mkdir(parents=True) + (planted / "package.json").write_text(json.dumps({ + "name": name, "version": version, "bin": {bin_name: "./adapter"}, + })) + (planted / "adapter").write_text( + f"#!{sys.executable}\nfrom pathlib import Path\n" + f"Path({str(marker)!r}).touch()\n" + ) + (planted / "adapter").chmod(0o755) + bin_dir = workspace / "node_modules" / ".bin" + bin_dir.mkdir() + (bin_dir / bin_name).symlink_to(planted / "adapter") + # No environment registry override: the malicious project config must + # be irrelevant because startup does not invoke npm at all. + (workspace / ".npmrc").write_text("registry=https://untrusted.invalid/\n") + fake_npx = installed_adapter.parent / "npx" + fake_npx.write_text((planted / "adapter").read_text()) + fake_npx.chmod(0o755) + # Model the real npm shim's /usr/bin/env node shebang. A trusted + # adapter must not pick an interpreter from the workspace either. + planted_node = bin_dir / "node" + planted_node.write_text((planted / "adapter").read_text()) + planted_node.chmod(0o755) + trusted_node = installed_adapter.parent / "node" + trusted_node.write_text( + f"#!{sys.executable}\nimport os, sys\n" + f"os.execv({sys.executable!r}, [{sys.executable!r}, *sys.argv[1:]])\n" + ) + trusted_node.chmod(0o755) + if installed: + installed_adapter.write_text( + "#!/usr/bin/env node\nimport json, os, sys\nfrom pathlib import Path\n" + f"Path({str(result)!r}).write_text(json.dumps({{'cwd': os.getcwd(), 'args': sys.argv[1:]}}))\n" + ) + else: + installed_adapter.unlink() + # Include the planted absolute PATH entry, not just npm's implicit + # search path. Discovery must skip the workspace even in this case. + monkeypatch.setenv("PATH", str(bin_dir) + os.pathsep + str(installed_adapter.parent)) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.chdir(workspace) + monkeypatch.setattr(mod, "get_jupyter_root_dir", lambda: str(workspace)) + client = mod.AcpAgentClient(SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace(acp_settings={}, nbi_user_dir=str(user_dir)), + )) + + class FakeConn: + async def initialize(self, **kw): + await asyncio.wait_for(client._proc.wait(), timeout=5) + raise RuntimeError("test: marker adapter exited before ACP initialization") + + monkeypatch.setattr(mod.acp, "connect_to_agent", lambda *args: FakeConn()) + asyncio.run(client._serve()) + assert client._started.is_set() + assert not marker.exists() + if installed: + launch = json.loads(result.read_text()) + assert launch["cwd"] == str(user_dir) + assert launch["args"] == ["-c", 'approval_policy="untrusted"'] + else: + assert not result.exists() + assert "Install @zed-industries/codex-acp@0.16.0" in client._start_error + + def test_relative_override_fails_before_spawning(self, tmp_path, monkeypatch, installed_adapter): + import notebook_intelligence.acp_agent as mod + + monkeypatch.chdir(installed_adapter.parent) + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", "./codex-acp") + client = mod.AcpAgentClient(SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace(acp_settings={}, nbi_user_dir=str(tmp_path / "user")), + )) + + async def unexpected_spawn(*args, **kwargs): + pytest.fail("a relative override must not reach subprocess startup") + + monkeypatch.setattr(mod.asyncio, "create_subprocess_exec", unexpected_spawn) + asyncio.run(client._serve()) + assert client._started.is_set() + assert "absolute executable path" in client._start_error + + @pytest.mark.parametrize("override", [False, True]) + def test_child_path_filters_workspace_only_for_default_launch( + self, tmp_path, monkeypatch, installed_adapter, override + ): + import notebook_intelligence.acp_agent as mod + + workspace = tmp_path / "workspace" + workspace_bin = workspace / ".venv" / "bin" + workspace_bin.mkdir(parents=True) + inherited_path = os.pathsep.join([str(workspace_bin), str(installed_adapter.parent)]) + monkeypatch.setenv("PATH", inherited_path) + monkeypatch.setattr(mod, "get_jupyter_root_dir", lambda: str(workspace)) + if override: + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", str(installed_adapter)) + client = mod.AcpAgentClient(SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace(acp_settings={}, nbi_user_dir=str(tmp_path / "user")), + )) + + env = client._child_env(client.agent_spec) + + assert env["PATH"] == (inherited_path if override else str(installed_adapter.parent.resolve())) + + def test_environment_failure_signals_start_and_tears_down(self, tmp_path, monkeypatch): + import notebook_intelligence.acp_agent as mod + + client = mod.AcpAgentClient(SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace(acp_settings={}, nbi_user_dir=str(tmp_path / "user")), + )) + + def broken_environment(spec): + raise ValueError("Workspace root cannot be inspected") + + async def unexpected_spawn(*args, **kwargs): + pytest.fail("environment validation must finish before spawning") + + monkeypatch.setattr(client, "_child_env", broken_environment) + monkeypatch.setattr(mod.asyncio, "create_subprocess_exec", unexpected_spawn) + + asyncio.run(client._serve()) + + assert client._started.is_set() + assert "Workspace root cannot be inspected" in client._start_error + assert client._loop is None + assert client._proc is None + + def test_missing_interpreter_fails_before_spawning( + self, tmp_path, monkeypatch, installed_adapter + ): + import notebook_intelligence.acp_agent as mod + + installed_adapter.write_text("#!/usr/bin/env node\n") + monkeypatch.setenv("PATH", str(installed_adapter.parent)) + client = mod.AcpAgentClient(SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace(acp_settings={}, nbi_user_dir=str(tmp_path / "user")), + )) + + async def unexpected_spawn(*args, **kwargs): + pytest.fail("missing interpreter must be diagnosed before spawning") + + monkeypatch.setattr(mod.asyncio, "create_subprocess_exec", unexpected_spawn) + asyncio.run(client._serve()) + + assert client._started.is_set() + assert "node" in client._start_error + assert "PATH" in client._start_error + assert client._loop is None + + @pytest.mark.parametrize("override", [False, True]) + @pytest.mark.parametrize("interpreter_kind", ["absolute", "symlink"]) + def test_workspace_interpreter_requires_explicit_override( + self, tmp_path, monkeypatch, installed_adapter, override, interpreter_kind + ): + import notebook_intelligence.acp_agent as mod + + workspace = tmp_path / "workspace" + workspace.mkdir() + node = workspace / "node" + node.write_bytes(b"") + node.chmod(0o755) + if interpreter_kind == "absolute": + installed_adapter.write_text(f"#!{node}\n") + else: + (installed_adapter.parent / "node").symlink_to(node) + installed_adapter.write_text("#!/usr/bin/env node\n") + monkeypatch.setenv("PATH", str(installed_adapter.parent)) + monkeypatch.setattr(mod, "get_jupyter_root_dir", lambda: str(workspace)) + if override: + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", str(installed_adapter)) + client = mod.AcpAgentClient(SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace(acp_settings={}, nbi_user_dir=str(tmp_path / "user")), + )) + captured = [] + + async def capture_launch(*args, **kwargs): + captured.append(args) + raise RuntimeError("fixture captured launch without executing") + + monkeypatch.setattr(mod.asyncio, "create_subprocess_exec", capture_launch) + asyncio.run(client._serve()) + + assert bool(captured) is override + assert client._started.is_set() + if not override: + assert "workspace" in client._start_error.lower() + assert client._loop is None + + @pytest.mark.skipif(os.name != "posix", reason="POSIX env shebang startup fixture") + @pytest.mark.parametrize("trusted_node_available", [False, True]) + def test_separator_in_resolved_path_cannot_launch_workspace_interpreter( + self, tmp_path, monkeypatch, installed_adapter, trusted_node_available + ): + import notebook_intelligence.acp_agent as mod + + workspace = tmp_path / "workspace" + workspace_bin = workspace / "bin" + workspace_bin.mkdir(parents=True) + # This is one external directory on disk, but naïvely serializing its + # name into PATH would introduce the separate workspace/bin entry. + target = tmp_path / ("prefix" + os.pathsep + str(workspace_bin)) + target.mkdir(parents=True) + alias = tmp_path / "trusted-alias" + alias.symlink_to(target, target_is_directory=True) + unwanted = tmp_path / "workspace-interpreter-ran" + workspace_node = workspace_bin / "node" + workspace_node.write_text( + f"#!{sys.executable}\nfrom pathlib import Path\nPath({str(unwanted)!r}).touch()\n" + ) + workspace_node.chmod(0o755) + expected = tmp_path / "trusted-adapter-ran" + installed_adapter.write_text( + "#!/usr/bin/env node\nfrom pathlib import Path\n" + f"Path({str(expected)!r}).touch()\n" + ) + if trusted_node_available: + trusted_node = installed_adapter.parent / "node" + trusted_node.write_text( + f"#!{sys.executable}\nimport os, sys\n" + f"os.execv({sys.executable!r}, [{sys.executable!r}, *sys.argv[1:]])\n" + ) + trusted_node.chmod(0o755) + monkeypatch.setenv("PATH", str(alias) + os.pathsep + str(installed_adapter.parent)) + monkeypatch.setattr(mod, "get_jupyter_root_dir", lambda: str(workspace)) + client = mod.AcpAgentClient(SimpleNamespace( + websocket_connector=None, + nbi_config=SimpleNamespace(acp_settings={}, nbi_user_dir=str(tmp_path / "user")), + )) + + class FakeConn: + async def initialize(self, **kwargs): + await asyncio.wait_for(client._proc.wait(), timeout=5) + raise RuntimeError("fixture adapter exited before initialization") + + monkeypatch.setattr(mod.acp, "connect_to_agent", lambda *args: FakeConn()) + assert client._child_env(client.agent_spec)["PATH"] == str(installed_adapter.parent.resolve()) + asyncio.run(client._serve()) + + assert client._started.is_set() + assert not unwanted.exists() + assert expected.exists() is trusted_node_available + if not trusted_node_available: + assert "requires interpreter 'node'" in client._start_error + assert "missing or not executable on the adapter PATH" in client._start_error + class TestAssembleQuery: """The turn's context lines (attachments, current-file pointer, output diff --git a/tests/test_acp_registry.py b/tests/test_acp_registry.py new file mode 100644 index 00000000..7c6d8f8e --- /dev/null +++ b/tests/test_acp_registry.py @@ -0,0 +1,452 @@ +# Copyright (c) Mehmet Bektas + +"""Adapter resolution checks use inert executable fixtures, never real adapters.""" + +import os +import shlex +import shutil +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from notebook_intelligence import acp_registry as registry + + +@pytest.fixture +def layout(tmp_path, monkeypatch): + workspace = tmp_path / "workspace" + workspace.mkdir() + trusted = tmp_path / "trusted-bin" + trusted.mkdir() + monkeypatch.delenv("NBI_ACP_AGENT_COMMAND", raising=False) + monkeypatch.setenv("PATH", str(trusted)) + return workspace, trusted + + +def _executable(path): + path.write_text("#!/bin/sh\nexit 99\n") + path.chmod(0o755) + return path + + +def _resolve(workspace): + return registry.resolve_acp_agent_command( + registry.resolve_acp_agent("codex"), workspace_root=str(workspace) + ) + + +def test_default_resolves_preinstalled_adapter_to_absolute_target(layout): + workspace, trusted = layout + installed = _executable(trusted / "codex-acp") + assert _resolve(workspace) == [str(installed.resolve())] + assert registry.acp_adapter_path(str(workspace)) == str(trusted.resolve()) + + +def test_default_command_uses_configured_workspace_when_not_passed(layout, monkeypatch): + from notebook_intelligence import util + + workspace, trusted = layout + installed = _executable(trusted / "codex-acp") + monkeypatch.setattr(util, "_jupyter_root_dir", str(workspace)) + assert registry.resolve_acp_agent_command(registry.resolve_acp_agent(None)) == [ + str(installed.resolve()) + ] + + +@pytest.mark.parametrize("path_entry", ["", ".", "bin"]) +def test_relative_path_entries_never_select_current_directory( + layout, monkeypatch, path_entry +): + workspace, trusted = layout + server_cwd = workspace.parent / "server-cwd" + server_cwd.mkdir() + candidate_dir = server_cwd / path_entry if path_entry else server_cwd + candidate_dir.mkdir(exist_ok=True) + _executable(candidate_dir / "codex-acp") + installed = _executable(trusted / "codex-acp") + monkeypatch.chdir(server_cwd) + monkeypatch.setenv("PATH", path_entry + os.pathsep + str(trusted)) + assert registry.acp_adapter_path(str(workspace)) == str(trusted.resolve()) + assert _resolve(workspace) == [str(installed.resolve())] + + +@pytest.mark.parametrize("later_trusted", [False, True]) +@pytest.mark.parametrize("kind", ["direct", "directory-link", "adapter-link"]) +def test_workspace_candidates_are_skipped_or_fail_closed( + layout, monkeypatch, kind, later_trusted +): + workspace, trusted = layout + planted = _executable(workspace / "codex-acp") + entry = workspace + if kind == "directory-link": + entry = workspace.parent / "workspace-alias" + entry.symlink_to(workspace, target_is_directory=True) + elif kind == "adapter-link": + entry = workspace.parent / "first-bin" + entry.mkdir() + (entry / "codex-acp").symlink_to(planted) + search_path = str(entry) + if later_trusted: + installed = _executable(trusted / "codex-acp") + search_path += os.pathsep + str(trusted) + monkeypatch.setenv("PATH", search_path) + if later_trusted: + assert _resolve(workspace) == [str(installed.resolve())] + else: + with pytest.raises(FileNotFoundError, match="outside the workspace"): + _resolve(workspace) + if kind != "adapter-link": + assert str(workspace) not in registry.acp_adapter_path(str(workspace)) + + +@pytest.mark.parametrize("symlinked_parent", [False, True]) +def test_workspace_root_can_be_reached_through_a_symlink( + layout, monkeypatch, symlinked_parent +): + workspace, trusted = layout + _executable(workspace / "codex-acp") + installed = _executable(trusted / "codex-acp") + alias = workspace.parent / "root-link" + if symlinked_parent: + alias.symlink_to(workspace.parent, target_is_directory=True) + alias = alias / workspace.name + else: + alias.symlink_to(workspace, target_is_directory=True) + monkeypatch.setenv("PATH", str(workspace) + os.pathsep + str(trusted)) + assert registry.acp_adapter_path(str(alias)) == str(trusted.resolve()) + assert _resolve(alias) == [str(installed.resolve())] + + +def test_workspace_directory_linking_out_is_still_excluded(layout, monkeypatch): + workspace, trusted = layout + _executable(trusted / "codex-acp") + link = workspace / "external-bin" + link.symlink_to(trusted, target_is_directory=True) + monkeypatch.setenv("PATH", str(link)) + assert registry.acp_adapter_path(str(workspace)) == "" + with pytest.raises(FileNotFoundError): + _resolve(workspace) + + +@pytest.mark.parametrize("spelling", ["case", "unicode"]) +@pytest.mark.parametrize("linking_out", [False, True]) +def test_filesystem_aliases_of_workspace_remain_excluded( + layout, monkeypatch, spelling, linking_out +): + old_workspace, trusted = layout + name, alternate = ( + ("MixedCaseWorkspace", "mixedcaseworkspace") + if spelling == "case" else ("caf\u00e9", "cafe\u0301") + ) + workspace = old_workspace.parent / name + workspace.mkdir() + alias = workspace.parent / alternate + if not alias.exists() or not alias.samefile(workspace): + pytest.skip(f"this filesystem does not alias {spelling} spellings") + if linking_out: + _executable(trusted / "codex-acp") + (workspace / "bin").symlink_to(trusted, target_is_directory=True) + alias = alias / "bin" + else: + _executable(workspace / "codex-acp") + monkeypatch.setenv("PATH", str(alias)) + assert registry.acp_adapter_path(str(workspace)) == "" + with pytest.raises(FileNotFoundError): + _resolve(workspace) + + +@pytest.mark.parametrize("linking_out", [False, True]) +def test_root_identity_alias_is_excluded_without_textual_or_symlink_match( + layout, monkeypatch, linking_out +): + """Model a second mount of the root on filesystems without case aliases.""" + workspace, trusted = layout + alias = workspace.parent / "identity-alias" + alias.mkdir() + if linking_out: + _executable(trusted / "codex-acp") + (alias / "bin").symlink_to(trusted, target_is_directory=True) + entry = alias / "bin" + else: + _executable(alias / "codex-acp") + entry = alias + root_info = workspace.stat() + original_stat = Path.stat + + def aliased_stat(path, *args, **kwargs): + return root_info if path == alias else original_stat(path, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", aliased_stat) + monkeypatch.setenv("PATH", str(entry)) + assert registry.acp_adapter_path(str(workspace)) == "" + with pytest.raises(FileNotFoundError): + _resolve(workspace) + + +def test_trusted_directory_and_adapter_symlinks_keep_resolved_targets(layout, monkeypatch): + workspace, trusted = layout + installed = _executable(trusted / "adapter-target") + (trusted / "codex-acp").symlink_to(installed) + directory_link = workspace.parent / "trusted-link" + directory_link.symlink_to(trusted, target_is_directory=True) + monkeypatch.setenv("PATH", str(directory_link)) + assert registry.acp_adapter_path(str(workspace)) == str(trusted.resolve()) + assert _resolve(workspace) == [str(installed.resolve())] + + +def _separator_alias(workspace): + """The canonical name would serialize a workspace bin as a second entry.""" + workspace_bin = workspace / "bin" + workspace_bin.mkdir() + target = Path(str(workspace.parent / "external") + os.pathsep + str(workspace_bin)) + target.mkdir(parents=True) + alias = workspace.parent / "trusted-alias" + alias.symlink_to(target, target_is_directory=True) + return alias, target, workspace_bin + + +@pytest.mark.parametrize("later_trusted", [False, True]) +def test_separator_in_canonical_directory_cannot_select_adapter( + layout, monkeypatch, later_trusted +): + workspace, trusted = layout + alias, target, _ = _separator_alias(workspace) + _executable(target / "codex-acp") + search_path = str(alias) + if later_trusted: + installed = _executable(trusted / "codex-acp") + search_path += os.pathsep + str(trusted) + monkeypatch.setenv("PATH", search_path) + + if later_trusted: + assert _resolve(workspace) == [str(installed.resolve())] + else: + with pytest.raises(FileNotFoundError, match="outside the workspace"): + _resolve(workspace) + + +@pytest.mark.parametrize("trusted_interpreter", [False, True]) +def test_separator_in_canonical_directory_cannot_inject_interpreter_path( + layout, monkeypatch, trusted_interpreter +): + workspace, trusted = layout + alias, target, workspace_bin = _separator_alias(workspace) + _executable(workspace_bin / "node") + installed = _executable(trusted / "codex-acp") + trusted_node = _executable(trusted / "node") if trusted_interpreter else None + monkeypatch.setenv("PATH", str(alias) + os.pathsep + str(trusted)) + + # The directory is external before serialization; only the separator + # check prevents it from injecting the workspace directory into PATH. + assert not target.is_relative_to(workspace) + assert str(workspace_bin) in str(target).split(os.pathsep) + assert _resolve(workspace) == [str(installed.resolve())] + child_path = registry.acp_adapter_path(str(workspace)) + assert child_path.split(os.pathsep) == [str(trusted.resolve())] + assert shutil.which("node", path=child_path) == ( + str(trusted_node.resolve()) if trusted_node else None + ) + + +@pytest.mark.parametrize("kind", ["direct", "link-in", "link-out"]) +def test_default_executable_validation_rejects_workspace_paths(layout, kind): + workspace, trusted = layout + candidate = _executable(workspace / "node") + if kind == "link-in": + candidate = trusted / "node" + candidate.symlink_to(workspace / "node") + elif kind == "link-out": + candidate.unlink() + candidate.symlink_to(_executable(trusted / "node")) + with pytest.raises(ValueError, match="outside the workspace.*Install the adapter"): + registry.validate_acp_executable_path(str(candidate), str(workspace)) + + +def test_default_executable_validation_accepts_trusted_target(layout): + workspace, trusted = layout + installed = _executable(trusted / "node-real") + alias = trusted / "node" + alias.symlink_to(installed) + registry.validate_acp_executable_path(str(alias), str(workspace)) + + +@pytest.mark.parametrize("kind", ["relative", "missing", "loop", "invalid"]) +def test_default_executable_validation_rejects_uninspectable_paths(layout, kind): + workspace, trusted = layout + candidate = trusted / "node" + if kind == "relative": + candidate = Path("node") + elif kind == "loop": + candidate.symlink_to(candidate) + elif kind == "invalid": + candidate = "invalid\x00path" + with pytest.raises(ValueError, match="existing, inspectable absolute path"): + registry.validate_acp_executable_path(str(candidate), str(workspace)) + + +def test_default_executable_validation_rejects_workspace_identity_alias(layout, monkeypatch): + workspace, trusted = layout + alias = workspace.parent / "identity-alias" + alias.mkdir() + interpreter = _executable(alias / "node") + root_info = workspace.stat() + original_stat = Path.stat + + def aliased_stat(path, *args, **kwargs): + return root_info if path == alias else original_stat(path, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", aliased_stat) + with pytest.raises(ValueError, match="outside the workspace"): + registry.validate_acp_executable_path(str(interpreter), str(workspace)) + + +def test_missing_files_and_symlink_loops_do_not_hide_later_installation(layout, monkeypatch): + workspace, trusted = layout + installed = _executable(trusted / "codex-acp") + loop = workspace.parent / "loop" + loop.symlink_to(loop) + not_directory = _executable(workspace.parent / "not-a-directory") + monkeypatch.setenv("PATH", os.pathsep.join(map(str, [ + workspace.parent / "missing", not_directory, loop, trusted, + ]))) + assert registry.acp_adapter_path(str(workspace)) == str(trusted.resolve()) + assert _resolve(workspace) == [str(installed.resolve())] + + +@pytest.mark.parametrize("error_type", [OSError, RuntimeError, ValueError]) +@pytest.mark.parametrize("failure_at", ["directory", "executable"]) +def test_uninspectable_entries_do_not_hide_later_installation( + layout, monkeypatch, error_type, failure_at +): + workspace, trusted = layout + installed = _executable(trusted / "codex-acp") + broken = workspace.parent / "broken-bin" + broken.mkdir() + broken_adapter = _executable(broken / "codex-acp") + failing_path = broken if failure_at == "directory" else broken_adapter + original_resolve = Path.resolve + + def failing_resolve(path, *args, **kwargs): + if path == failing_path: + raise error_type("uninspectable fixture") + return original_resolve(path, *args, **kwargs) + + monkeypatch.setattr(Path, "resolve", failing_resolve) + monkeypatch.setenv("PATH", str(broken) + os.pathsep + str(trusted)) + assert _resolve(workspace) == [str(installed.resolve())] + if failure_at == "directory": + assert registry.acp_adapter_path(str(workspace)) == str(trusted.resolve()) + + +def test_unreadable_ancestor_is_skipped(layout, monkeypatch): + workspace, trusted = layout + installed = _executable(trusted / "codex-acp") + unreadable = workspace.parent / "unreadable" + directory = unreadable / "bin" + directory.mkdir(parents=True) + _executable(directory / "codex-acp") + original_stat = Path.stat + + def failing_stat(path, *args, **kwargs): + if path == unreadable: + raise PermissionError("unreadable fixture") + return original_stat(path, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", failing_stat) + monkeypatch.setenv("PATH", str(directory) + os.pathsep + str(trusted)) + assert registry.acp_adapter_path(str(workspace)) == str(trusted.resolve()) + assert _resolve(workspace) == [str(installed.resolve())] + + +@pytest.mark.parametrize("root_kind", ["empty", "missing", "file", "loop"]) +def test_unknown_workspace_root_fails_closed(layout, monkeypatch, root_kind): + workspace, trusted = layout + _executable(trusted / "codex-acp") + root = workspace.parent / root_kind + if root_kind == "file": + root.write_text("not a directory") + elif root_kind == "loop": + root.symlink_to(root) + elif root_kind == "empty": + root = "" + with pytest.raises(ValueError, match="workspace root.*existing, inspectable"): + registry.acp_adapter_path(str(root)) + with pytest.raises(ValueError, match="workspace root.*existing, inspectable"): + _resolve(root) + + +@pytest.mark.parametrize("zero_identity_at", ["root", "entry"]) +def test_unknown_filesystem_identity_fails_closed(layout, monkeypatch, zero_identity_at): + workspace, trusted = layout + _executable(trusted / "codex-acp") + failing_path = workspace if zero_identity_at == "root" else trusted + original_stat = Path.stat + + def zero_identity_stat(path, *args, **kwargs): + info = original_stat(path, *args, **kwargs) + if path == failing_path: + return SimpleNamespace(st_mode=info.st_mode, st_dev=0, st_ino=0) + return info + + monkeypatch.setattr(Path, "stat", zero_identity_stat) + if zero_identity_at == "root": + with pytest.raises(ValueError, match="workspace root"): + _resolve(workspace) + else: + assert registry.acp_adapter_path(str(workspace)) == "" + with pytest.raises(FileNotFoundError): + _resolve(workspace) + + +@pytest.mark.parametrize("command", ["", " ", "'unclosed", "codex-acp", "./codex-acp"]) +def test_override_requires_nonempty_absolute_command(layout, monkeypatch, command): + workspace, _ = layout + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", command) + with pytest.raises(ValueError): + _resolve(workspace) + + +def test_missing_absolute_override_fails_without_path_fallback(layout, monkeypatch): + workspace, trusted = layout + _executable(trusted / "codex-acp") + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", str(trusted / "missing")) + with pytest.raises(FileNotFoundError, match="missing"): + _resolve(workspace) + + +def test_explicit_override_preserves_arguments_and_admin_workspace_trust(layout, monkeypatch): + workspace, _ = layout + adapter = _executable(workspace / "custom adapter") + monkeypatch.setenv( + "NBI_ACP_AGENT_COMMAND", shlex.join([str(adapter), "--label", "two words"]) + ) + assert _resolve(workspace) == [str(adapter), "--label", "two words"] + # This explicit administrator decision does not depend on workspace discovery. + assert _resolve("") == [str(adapter), "--label", "two words"] + + +@pytest.mark.parametrize("error_type", [PermissionError, RuntimeError]) +def test_uninspectable_workspace_root_has_a_specific_error(layout, monkeypatch, error_type): + workspace, trusted = layout + _executable(trusted / "codex-acp") + original_resolve = Path.resolve + + def failing_resolve(path, *args, **kwargs): + if path == workspace: + raise error_type("uninspectable root fixture") + return original_resolve(path, *args, **kwargs) + + monkeypatch.setattr(Path, "resolve", failing_resolve) + with pytest.raises(ValueError, match="workspace root.*existing, inspectable"): + _resolve(workspace) + + +def test_explicit_override_preserves_shim_name(layout, monkeypatch): + workspace, trusted = layout + target = _executable(trusted / "toolchain-shim") + adapter = trusted / "codex-acp" + adapter.symlink_to(target) + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", shlex.quote(str(adapter))) + assert _resolve(workspace) == [str(adapter)] + assert str(adapter) != str(target) diff --git a/tests/test_acp_runtime.py b/tests/test_acp_runtime.py new file mode 100644 index 00000000..43200f5d --- /dev/null +++ b/tests/test_acp_runtime.py @@ -0,0 +1,203 @@ +"""ACP interpreter checks inspect inert fixtures without launching programs.""" + +import os +import subprocess + +import pytest + +from notebook_intelligence.acp_runtime import validate_acp_adapter_runtime + + +def _executable(path, contents=b"inert executable fixture\n"): + path.write_bytes(contents) + path.chmod(0o700) + return path + + +@pytest.fixture +def runtime_files(tmp_path, monkeypatch): + trusted = tmp_path / "trusted-bin" + trusted.mkdir() + workspace = tmp_path / "workspace" + workspace.mkdir() + env = _executable(trusted / "env") + node = _executable(trusted / "node") + adapter = trusted / "adapter" + + def fail_if_executed(*args, **kwargs): + pytest.fail("runtime validation must not launch adapters or interpreters") + + monkeypatch.setattr(subprocess, "Popen", fail_if_executed) + return trusted, workspace, env, node, adapter + + +def test_non_utf8_shebang_names_the_decoding_failure(runtime_files): + trusted, _, env, _, adapter = runtime_files + _executable(adapter, f"#!{env} ".encode() + b"\xff\n") + with pytest.raises(ValueError, match="interpreter line is not valid UTF-8"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted)) + + +@pytest.mark.parametrize("suffix", [b"\r", b"\x00", b"\x01", b"\x0b", b" " * 4096]) +def test_control_characters_and_overlong_shebangs_fail_before_lookup(runtime_files, suffix): + trusted, _, env, _, adapter = runtime_files + _executable(adapter, f"#!{env} node".encode() + suffix + b"\n") + with pytest.raises(ValueError, match="invalid or overlong interpreter line"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted)) + + +def test_relative_shebang_interpreter_is_rejected_even_when_executable_exists(runtime_files, monkeypatch): + trusted, _, _, _, adapter = runtime_files + monkeypatch.chdir(trusted) + _executable(adapter, b"#!node\n") + with pytest.raises(ValueError, match="must declare an absolute interpreter path"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted)) + + +@pytest.mark.parametrize( + "arguments", + [ + "", + "node --no-warnings", + "-i node", + "-S", + "-S PATH=/tmp node", + "-S ${NODE}", + "-S 'node' --no-warnings", + '-S "node" --no-warnings', + "-S node\\ --no-warnings", + "-S node #comment", + "-S -i node", + ], +) +def test_ambiguous_env_syntax_is_rejected_with_a_parser_error(runtime_files, arguments): + trusted, _, env, _, adapter = runtime_files + _executable(adapter, f"#!{env} {arguments}\n".encode()) + with pytest.raises(ValueError, match="Cannot validate ACP adapter's env interpreter line"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted)) + + +@pytest.mark.parametrize("arguments", ["node", "-S node --no-warnings --stack-size=2048"]) +def test_plain_env_program_and_split_string_flags_are_supported(runtime_files, arguments): + trusted, _, env, _, adapter = runtime_files + _executable(adapter, f"#!{env} {arguments}\n".encode()) + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted)) + + +@pytest.mark.parametrize("contents", [b"", b"plain text without a shebang\n", b"\x7fELF\xff\x00", b"MZ\x00\xff"]) +def test_no_shebang_and_native_bytes_do_not_require_an_interpreter(runtime_files, contents): + _, _, _, _, adapter = runtime_files + _executable(adapter, contents) + validate_acp_adapter_runtime([str(adapter)], env_path="") + + +@pytest.mark.skipif(os.name == "nt", reason="opening directories as file descriptors is POSIX-specific") +def test_directory_adapter_reports_its_read_failure(runtime_files): + trusted, _, _, _, adapter = runtime_files + adapter.mkdir() + with pytest.raises(ValueError, match="Cannot read ACP adapter.*IsADirectoryError"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted)) + + +@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="FIFO fixtures require POSIX") +def test_fifo_adapter_is_rejected_without_waiting_for_a_writer(runtime_files): + trusted, _, _, _, adapter = runtime_files + os.mkfifo(adapter) + with pytest.raises(ValueError, match="is not a regular file"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted)) + + +def test_env_relative_program_is_resolved_from_launch_cwd(runtime_files, tmp_path, monkeypatch): + trusted, _, env, _, adapter = runtime_files + launch = tmp_path / "launch" + (launch / "bin").mkdir(parents=True) + _executable(launch / "bin" / "node") + server = tmp_path / "server" + server.mkdir() + monkeypatch.chdir(server) + _executable(adapter, f"#!{env} bin/node\n".encode()) + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted), cwd=str(launch)) + with pytest.raises(ValueError, match="working directory is needed to check its relative interpreter"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted)) + with pytest.raises(ValueError, match="requires interpreter 'bin/node'.*adapter PATH"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted), cwd=str(server)) + + +def test_env_relative_program_preserves_symlink_parent_traversal(runtime_files, tmp_path): + trusted, _, env, _, adapter = runtime_files + launch = tmp_path / "launch" + launch.mkdir() + target = tmp_path / "linked-target" + (target / "nested").mkdir(parents=True) + (launch / "bin-link").symlink_to(target / "nested", target_is_directory=True) + actual_node = _executable(target / "node") + _executable(adapter, f"#!{env} bin-link/../node\n".encode()) + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted), cwd=str(launch)) + actual_node.unlink() + _executable(launch / "node") + # Lexically collapsing '..' would incorrectly select launch/node. The + # kernel traverses the symlink first, so this interpreter is absent. + with pytest.raises(ValueError, match="requires interpreter 'bin-link/../node'.*adapter PATH"): + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted), cwd=str(launch)) + + +@pytest.mark.parametrize("search_path", ["", ".", "bin"]) +def test_env_relative_path_is_resolved_from_launch_cwd(runtime_files, tmp_path, monkeypatch, search_path): + _, _, env, _, adapter = runtime_files + launch = tmp_path / "launch" + directory = launch / search_path if search_path else launch + directory.mkdir(parents=True) + _executable(directory / "node") + server = tmp_path / "server" + server.mkdir() + monkeypatch.chdir(server) + _executable(adapter, f"#!{env} node\n".encode()) + validate_acp_adapter_runtime([str(adapter)], env_path=search_path, cwd=str(launch)) + with pytest.raises(ValueError, match="working directory is needed to check its relative PATH entries"): + validate_acp_adapter_runtime([str(adapter)], env_path=search_path) + with pytest.raises(ValueError, match="requires interpreter 'node'.*adapter PATH"): + validate_acp_adapter_runtime([str(adapter)], env_path=search_path, cwd=str(server)) + + +@pytest.mark.parametrize("via_symlink", [False, True]) +def test_default_rejects_absolute_workspace_interpreter_but_override_may_trust_it(runtime_files, via_symlink): + trusted, workspace, _, _, adapter = runtime_files + interpreter = _executable(workspace / "runtime") + if via_symlink: + alias = trusted / "runtime" + alias.symlink_to(interpreter) + interpreter = alias + _executable(adapter, f"#!{interpreter}\n".encode()) + with pytest.raises(ValueError, match="outside the workspace"): + validate_acp_adapter_runtime( + [str(adapter)], env_path=str(trusted), workspace_root=str(workspace) + ) + validate_acp_adapter_runtime([str(adapter)], env_path=str(trusted), workspace_root=None) + + +def test_default_rejects_first_env_interpreter_symlink_into_workspace(runtime_files, tmp_path): + trusted, workspace, env, node, adapter = runtime_files + node.unlink() + node.symlink_to(_executable(workspace / "node")) + fallback = tmp_path / "fallback-bin" + fallback.mkdir() + _executable(fallback / "node") + env_path = os.pathsep.join((str(trusted), str(fallback))) + _executable(adapter, f"#!{env} node\n".encode()) + # The actual env process selects the first node. A later safe candidate + # cannot repair its selection, so preflight must reject the first match. + with pytest.raises(ValueError, match="outside the workspace"): + validate_acp_adapter_runtime( + [str(adapter)], env_path=env_path, workspace_root=str(workspace) + ) + validate_acp_adapter_runtime([str(adapter)], env_path=env_path, workspace_root=None) + + +@pytest.mark.parametrize("use_env", [False, True]) +def test_default_accepts_interpreter_outside_workspace(runtime_files, use_env): + trusted, workspace, env, node, adapter = runtime_files + shebang = f"#!{env} node" if use_env else f"#!{node}" + _executable(adapter, (shebang + "\n").encode()) + validate_acp_adapter_runtime( + [str(adapter)], env_path=str(trusted), workspace_root=str(workspace) + ) diff --git a/tests/test_readiness.py b/tests/test_readiness.py index 337cba7e..eec41187 100644 --- a/tests/test_readiness.py +++ b/tests/test_readiness.py @@ -12,6 +12,7 @@ import pytest import notebook_intelligence.readiness as rd +import notebook_intelligence.util as util from notebook_intelligence.checks import CheckTimeout as _CheckTimeout @@ -20,6 +21,15 @@ # --------------------------------------------------------------------------- +@pytest.fixture(autouse=True) +def _readiness_workspace(tmp_path, monkeypatch): + """Never let ACP readiness depend on another test's process-global root.""" + workspace = tmp_path / "readiness-workspace" + workspace.mkdir() + monkeypatch.setattr(util, "_jupyter_root_dir", str(workspace)) + return workspace + + class _Model: def __init__(self, model_id): self.id = model_id @@ -42,6 +52,7 @@ def _config(**kw): chat_model=kw.get("chat_model", {}), claude_settings=kw.get("claude_settings", {}), acp_settings=kw.get("acp_settings", {}), + nbi_user_dir=kw.get("nbi_user_dir"), ) @@ -311,38 +322,144 @@ def test_claude_mode_does_not_run_provider_checks(): # --------------------------------------------------------------------------- -def test_acp_missing_npx_blocks_and_names_the_override(): - with patch("shutil.which", return_value=None), patch.dict(rd.os.environ, {}, clear=True): - doc = rd.run_readiness(_config(), _manager(mode="acp")) +def test_acp_missing_adapter_blocks_and_explains_preinstallation(tmp_path, monkeypatch): + monkeypatch.delenv("NBI_ACP_AGENT_COMMAND", raising=False) + monkeypatch.setenv("PATH", str(tmp_path)) + doc = rd.run_readiness(_config(), _manager(mode="acp")) row = _by_id(doc)["acp.runtime"] assert row["level"] == rd.LEVEL_BLOCKED + assert "'codex-acp' was not found on an absolute PATH entry outside the workspace" in row["detail"] + assert "Preinstall" in row["remedy"] + assert "@zed-industries/codex-acp@0.16.0" in row["remedy"] assert "NBI_ACP_AGENT_COMMAND" in row["remedy"] + assert "absolute path" in row["remedy"] + assert doc["verdict"] == rd.VERDICT_NOT_READY def test_acp_command_override_pointing_at_a_missing_binary_is_blocked(): - """Special-casing the literal "npx" would let an override point at a - nonexistent binary and still report Ready.""" with patch.dict( rd.os.environ, {"NBI_ACP_AGENT_COMMAND": "/opt/nope/agent --acp"}, clear=True ): doc = rd.run_readiness(_config(), _manager(mode="acp")) row = _by_id(doc)["acp.runtime"] assert row["level"] == rd.LEVEL_BLOCKED - assert "/opt/nope/agent" in row["detail"] + assert row["detail"] == "ACP adapter '/opt/nope/agent' was not found or is not executable" assert doc["verdict"] == rd.VERDICT_NOT_READY -def test_acp_command_override_pointing_at_a_real_binary_is_ok(): - with patch.dict( - rd.os.environ, {"NBI_ACP_AGENT_COMMAND": "/bin/sh -c true"}, clear=True - ): - doc = rd.run_readiness(_config(), _manager(mode="acp")) +@pytest.mark.parametrize("workspace_is_home", [False, True]) +def test_acp_absolute_override_explicitly_trusts_workspace_installation( + tmp_path, monkeypatch, workspace_is_home +): + if workspace_is_home: + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setattr(util, "_jupyter_root_dir", str(tmp_path)) + adapter = tmp_path / "adapter" + adapter.write_text("#!/bin/sh\nexit 0\n") + adapter.chmod(0o700) + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", f"{adapter} --acp") + doc = rd.run_readiness(_config(), _manager(mode="acp")) assert _by_id(doc)["acp.runtime"]["level"] == rd.LEVEL_OK + assert _by_id(doc)["acp.agent"]["detail"].endswith("/adapter --acp") + + +@pytest.mark.parametrize("command", ["./adapter --acp", "adapter --acp"]) +def test_acp_relative_override_is_blocked_even_if_present_in_server_cwd( + tmp_path, monkeypatch, command +): + adapter = tmp_path / "adapter" + adapter.write_text("#!/bin/sh\nexit 0\n") + adapter.chmod(0o700) + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("PATH", str(tmp_path)) + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", command) + doc = rd.run_readiness(_config(), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + assert row["level"] == rd.LEVEL_BLOCKED + assert row["detail"] == "NBI_ACP_AGENT_COMMAND must name an absolute executable path" + assert "absolute" in row["remedy"] + assert doc["verdict"] == rd.VERDICT_NOT_READY + + +@pytest.mark.parametrize( + "command, expected_reason", + [ + ("", "NBI_ACP_AGENT_COMMAND must name an absolute executable path"), + (" ", "NBI_ACP_AGENT_COMMAND must name an absolute executable path"), + ("'unclosed", "No closing quotation"), + ], +) +def test_acp_invalid_override_is_a_runtime_failure(command, expected_reason, monkeypatch): + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", command) + doc = rd.run_readiness(_config(), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + assert row["level"] == rd.LEVEL_BLOCKED + assert row["detail"] == expected_reason + assert not any(row["id"].endswith(".unavailable") for row in doc["checks"]) + + +@pytest.mark.parametrize("search_path", ["", ".", "bin"]) +def test_acp_cwd_relative_path_does_not_report_false_ready( + tmp_path, monkeypatch, search_path, _readiness_workspace +): + server_cwd = tmp_path / "server-cwd" + server_cwd.mkdir() + assert not server_cwd.is_relative_to(_readiness_workspace) + adapter_dir = server_cwd / search_path if search_path else server_cwd + adapter_dir.mkdir(exist_ok=True) + adapter = adapter_dir / "codex-acp" + adapter.write_text("#!/bin/sh\nexit 0\n") + adapter.chmod(0o700) + monkeypatch.chdir(server_cwd) + monkeypatch.delenv("NBI_ACP_AGENT_COMMAND", raising=False) + monkeypatch.setenv("PATH", search_path) + doc = rd.run_readiness(_config(), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + assert row["level"] == rd.LEVEL_BLOCKED + assert "'codex-acp' was not found on an absolute PATH entry outside the workspace" in row["detail"] + assert doc["verdict"] == rd.VERDICT_NOT_READY + + +@pytest.mark.parametrize("via_symlink", [False, True]) +@pytest.mark.parametrize("has_external_adapter", [False, True]) +def test_acp_default_skips_workspace_executables_and_symlinks_into_workspace( + tmp_path, monkeypatch, via_symlink, has_external_adapter +): + workspace = tmp_path / "workspace" + workspace.mkdir() + planted = workspace / "codex-acp" + planted.write_text("#!/bin/sh\nexit 0\n") + planted.chmod(0o700) + first_directory = workspace + if via_symlink: + first_directory = tmp_path / "first-bin" + first_directory.mkdir() + (first_directory / "codex-acp").symlink_to(planted) + search_path = str(first_directory) + if has_external_adapter: + trusted = tmp_path / "trusted-bin" + trusted.mkdir() + installed = trusted / "codex-acp" + installed.write_text("#!/bin/sh\nexit 0\n") + installed.chmod(0o700) + search_path += rd.os.pathsep + str(trusted) + monkeypatch.setattr(util, "_jupyter_root_dir", str(workspace)) + monkeypatch.delenv("NBI_ACP_AGENT_COMMAND", raising=False) + monkeypatch.setenv("PATH", search_path) + doc = rd.run_readiness(_config(), _manager(mode="acp")) + if has_external_adapter: + assert _by_id(doc)["acp.runtime"]["level"] == rd.LEVEL_OK + assert _by_id(doc)["acp.agent"]["detail"].endswith("/trusted-bin/codex-acp") + else: + row = _by_id(doc)["acp.runtime"] + assert row["level"] == rd.LEVEL_BLOCKED + assert "'codex-acp' was not found on an absolute PATH entry outside the workspace" in row["detail"] + assert doc["verdict"] == rd.VERDICT_NOT_READY def test_acp_credentials_absent_warns_for_the_oauth_case(): - with patch("shutil.which", return_value="/usr/bin/npx"), patch.dict( - rd.os.environ, {}, clear=True + with patch.dict( + rd.os.environ, {"NBI_ACP_AGENT_COMMAND": "/bin/sh"}, clear=True ): doc = rd.run_readiness(_config(), _manager(mode="acp")) row = _by_id(doc)["acp.credentials"] @@ -350,6 +467,188 @@ def test_acp_credentials_absent_warns_for_the_oauth_case(): assert "OAuth" in row["remedy"] +def _acp_script_fixture(tmp_path, monkeypatch, shebang="#!/usr/bin/env node"): + workspace = tmp_path / "workspace" + workspace.mkdir() + trusted = tmp_path / "trusted-bin" + trusted.mkdir() + adapter = trusted / "codex-acp" + adapter.write_text(shebang + "\nthrow new Error('must not execute');\n") + adapter.chmod(0o700) + monkeypatch.setattr(util, "_jupyter_root_dir", str(workspace)) + monkeypatch.delenv("NBI_ACP_AGENT_COMMAND", raising=False) + monkeypatch.setenv("PATH", str(trusted)) + return workspace, trusted, adapter + + +@pytest.mark.parametrize("node_location", ["missing", "workspace", "directory-alias", "trusted"]) +def test_acp_checks_node_against_the_filtered_launch_path( + tmp_path, monkeypatch, node_location +): + workspace, trusted, _ = _acp_script_fixture(tmp_path, monkeypatch) + if node_location != "missing": + node_directory = trusted if node_location == "trusted" else workspace + node = node_directory / "node" + node.write_text("#!/bin/sh\necho must-not-execute\n") + node.chmod(0o700) + if node_location == "directory-alias": + alias = tmp_path / "linked-bin" + alias.symlink_to(workspace, target_is_directory=True) + node_directory = alias + monkeypatch.setenv("PATH", str(node_directory) + rd.os.pathsep + str(trusted)) + with patch("subprocess.Popen", side_effect=AssertionError("readiness must not execute the adapter")): + doc = rd.run_readiness(_config(), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + if node_location == "trusted": + assert row["level"] == rd.LEVEL_OK + else: + assert row["level"] == rd.LEVEL_BLOCKED + assert "node" in row["detail"] + assert "adapter PATH" in row["detail"] + assert "interpreter" in row["remedy"] + assert "workspace" in row["remedy"] + assert doc["verdict"] == rd.VERDICT_NOT_READY + + +@pytest.mark.parametrize("relative_path", [False, True]) +def test_acp_override_checks_interpreter_using_inherited_path_and_launch_cwd( + tmp_path, monkeypatch, relative_path +): + workspace, _, adapter = _acp_script_fixture(tmp_path, monkeypatch) + launch_directory = tmp_path / "nbi-user-dir" + launch_directory.mkdir() + if relative_path: + node_directory = launch_directory / "bin" + node_directory.mkdir() + monkeypatch.setenv("PATH", "bin") + else: + node_directory = workspace + monkeypatch.setenv("PATH", str(workspace)) + node = node_directory / "node" + node.write_text("#!/bin/sh\nexit 1\n") + node.chmod(0o700) + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", str(adapter)) + doc = rd.run_readiness(_config(nbi_user_dir=str(launch_directory)), _manager(mode="acp")) + assert _by_id(doc)["acp.runtime"]["level"] == rd.LEVEL_OK + + +@pytest.mark.parametrize("interpreter_kind", ["absolute-shebang", "env-symlink"]) +@pytest.mark.parametrize("explicit_override", [False, True]) +def test_acp_workspace_interpreter_requires_an_explicit_administrator_override( + tmp_path, monkeypatch, interpreter_kind, explicit_override +): + workspace, trusted, adapter = _acp_script_fixture(tmp_path, monkeypatch) + node = workspace / "node" + node.write_text("#!/bin/sh\nexit 1\n") + node.chmod(0o700) + if interpreter_kind == "absolute-shebang": + adapter.write_text(f"#!{node}\nnever execute\n") + else: + (trusted / "node").symlink_to(node) + if explicit_override: + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", str(adapter)) + doc = rd.run_readiness(_config(), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + if explicit_override: + assert row["level"] == rd.LEVEL_OK + else: + assert row["level"] == rd.LEVEL_BLOCKED + assert "outside the workspace" in row["detail"] + assert "/workspace/node'" in row["detail"] or "/trusted-bin/node'" in row["detail"] + + +def test_acp_relative_override_path_is_not_resolved_from_the_server_cwd(tmp_path, monkeypatch): + workspace, _, adapter = _acp_script_fixture(tmp_path, monkeypatch) + server_bin = workspace / "bin" + server_bin.mkdir() + node = server_bin / "node" + node.write_text("#!/bin/sh\nexit 1\n") + node.chmod(0o700) + launch_directory = tmp_path / "nbi-user-dir" + launch_directory.mkdir() + monkeypatch.chdir(workspace) + monkeypatch.setenv("PATH", "bin") + monkeypatch.setenv("NBI_ACP_AGENT_COMMAND", str(adapter)) + doc = rd.run_readiness(_config(nbi_user_dir=str(launch_directory)), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + assert row["level"] == rd.LEVEL_BLOCKED + assert "requires interpreter 'node', which is missing or not executable on the adapter PATH" in row["detail"] + + +@pytest.mark.parametrize( + "shebang", + ["#!/usr/bin/env node", "#!/usr/bin/env -S node --no-warnings"], +) +def test_acp_common_env_interpreter_lines_are_checked_without_execution(tmp_path, monkeypatch, shebang): + _, trusted, _ = _acp_script_fixture(tmp_path, monkeypatch, shebang) + node = trusted / "node" + node.write_text("#!/bin/sh\nexit 1\n") + node.chmod(0o700) + doc = rd.run_readiness(_config(), _manager(mode="acp")) + assert _by_id(doc)["acp.runtime"]["level"] == rd.LEVEL_OK + + +@pytest.mark.parametrize( + "shebang, expected_reason", + [ + ("#!/usr/bin/env", "Cannot validate ACP adapter's env interpreter line"), + ("#!/usr/bin/env node --no-warnings", "Cannot validate ACP adapter's env interpreter line"), + ("#!/usr/bin/env -S PATH=/tmp node", "Cannot validate ACP adapter's env interpreter line"), + ("#!/usr/bin/env -S ${NODE}", "Cannot validate ACP adapter's env interpreter line"), + ("#!/usr/bin/env -S 'node' --no-warnings", "Cannot validate ACP adapter's env interpreter line"), + ("#!/usr/bin/env -i node", "Cannot validate ACP adapter's env interpreter line"), + ("#!relative/interpreter", "must declare an absolute interpreter path"), + ("#!/usr/bin/env node\r", "invalid or overlong interpreter line"), + ("#!/usr/bin/env node " + " " * 4096, "invalid or overlong interpreter line"), + ], +) +def test_acp_unsupported_or_invalid_shebang_reports_actionable_runtime_failure( + tmp_path, monkeypatch, shebang, expected_reason +): + _, trusted, _ = _acp_script_fixture(tmp_path, monkeypatch, shebang) + node = trusted / "node" + node.write_text("#!/bin/sh\nexit 1\n") + node.chmod(0o700) + doc = rd.run_readiness(_config(), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + assert row["level"] == rd.LEVEL_BLOCKED + assert expected_reason in row["detail"] + assert not any(row["id"].endswith(".unavailable") for row in doc["checks"]) + + +@pytest.mark.parametrize("interpreter_exists", [False, True]) +def test_acp_absolute_shebang_interpreter_is_checked(tmp_path, monkeypatch, interpreter_exists): + interpreter = tmp_path / "runtime" + if interpreter_exists: + interpreter.write_text("#!/bin/sh\nexit 1\n") + interpreter.chmod(0o700) + _acp_script_fixture(tmp_path, monkeypatch, f"#!{interpreter}") + doc = rd.run_readiness(_config(), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + assert row["level"] == (rd.LEVEL_OK if interpreter_exists else rd.LEVEL_BLOCKED) + if not interpreter_exists: + assert "/runtime'" in row["detail"] + + +def test_acp_unreadable_adapter_reports_runtime_permissions_remedy(tmp_path, monkeypatch): + import notebook_intelligence.acp_runtime as runtime + + _, _, adapter = _acp_script_fixture(tmp_path, monkeypatch) + original_open = runtime.os.open + + def deny_adapter(path, *args, **kwargs): + if str(path) == str(adapter): + raise PermissionError("fixture denies reading") + return original_open(path, *args, **kwargs) + + monkeypatch.setattr(runtime.os, "open", deny_adapter) + doc = rd.run_readiness(_config(), _manager(mode="acp")) + row = _by_id(doc)["acp.runtime"] + assert row["level"] == rd.LEVEL_BLOCKED + assert "read permissions" in row["detail"] + assert not any(row["id"].endswith(".unavailable") for row in doc["checks"]) + + # --------------------------------------------------------------------------- # Live completion (opt-in) # ---------------------------------------------------------------------------