supercode-harness 0.5.45

//! P5-12 (COMPOSABLE-HARNESS-DESIGN.md §2 module 18 `plugins`, D7 "in-process
//! extension API, packaging/marketplaces, custom tools from files, provider
//! injection, extension UI, plugin/package installation"; §2.1 D-10:
//! "config-borne code execution without a trust gate is an injection hole").
//!
//! # The ABI decision: out-of-process, trust-gated, manifest-declared
//!
//! A plugin is **not** an in-process dynamically-linked library or FFI —
//! that would be memory-unsafe in Rust, a versioning nightmare across
//! plugin/host builds, and would bypass the trust/sandbox boundary this
//! module exists to enforce. Instead, a plugin is a **directory** containing
//! a **manifest** (`plugin.toml`, `RawManifest`) that DECLARES what it
//! contributes — the manifest is DATA; the plugin's own code runs ONLY as a
//! subprocess this crate spawns, never linked into supercode's address
//! space. This keeps a plugin memory-safe to load (a malformed/hostile
//! manifest can't corrupt this process, only fail to parse), language-
//! agnostic (a plugin can be any executable), sandboxable via
//! [`crate::sandbox`]/`crate::tools::build_sandboxed_sh` exactly like
//! `bash`, and cleanly trust-gated (below) for D-10.
//!
//! ## The manifest schema (the ABI contract)
//! ```toml
//! name = "my-plugin"      # optional — the plugin's directory name is the
//! version = "0.1.0"       # fallback/authoritative namespace either way
//!
//! [[tools]]
//! name = "greet"                 # required, non-empty
//! command = "python3"            # required, non-empty — the executable
//! args = ["greet.py"]            # optional, fixed argv (config-borne, trusted)
//! description = "Say hello"      # optional
//! params = { type = "object", properties = { name = { type = "string" } } }
//! # ^ optional JSON Schema for the tool's input; defaults to
//! #   {"type": "object"} (an MCP-style server would be the natural growth
//! #   path for a richer tool surface — see "Honest gaps" below).
//!
//! [[hooks]]
//! event = "post_tool"            # a lifecycle event name (cli::hooks::HookEvent)
//! command = "notify.sh"          # required, non-empty
//! ```
//! A plugin registers its `[[tools]]` entries into the model-visible
//! [`crate::tools::ToolRegistry`] (namespaced `plugin__<plugin>__<tool>`,
//! mirroring [`crate::mcp::McpServerHandle`]'s `mcp__<server>__<tool>`
//! convention) via [`register_into`]. `[[hooks]]` entries are PARSED,
//! VALIDATED, and carried on [`LoadedPlugins::hooks`] — see "Honest gaps"
//! below for why their lifecycle EMISSION is not yet wired in this build,
//! the exact same "registerable now, emission deferred" shape
//! `crates/cli/src/hooks.rs`'s own `subagent_start`/`pre_compact` events
//! already use (that module's doc comment, P5-7).
//!
//! ## Discovery
//! [`discover_manifests`] scans a list of directories, each expected to
//! contain `<plugin-name>/plugin.toml` subdirectories — the ALWAYS-scanned
//! `$SUPERCODE_HOME/plugins` (mirroring `crate::agent::global_instructions_dir`,
//! the same "trusted user/global tier" location every other user-level
//! resource in this crate lives under) plus any extra
//! `[capabilities.plugins] dirs = [...]` entries. Since `[capabilities.plugins]`
//! (`dirs` included) is wholesale project-forbidden (see "Trust model"
//! below), `dirs` can only ever be user/global-layer or preset data — never
//! attacker-controlled project config.
//!
//! ## Subprocess execution model
//! A registered [`PluginTool::execute`] spawns the manifest's fixed
//! `command`/`args` (never the model's own arguments — see below) through
//! `crate::tools::build_sandboxed_sh`, the SAME sandboxed-spawn builder
//! `crate::tools::BashTool`/`crate::agent::Agent::background_exec` use — so
//! a plugin tool's subprocess gets the identical P5-10 OS sandbox
//! (Landlock/seatbelt)/env-policy/network-policy posture a `bash` call
//! would, not a second, weaker path. Like `background_exec`
//! (`crate::agent`'s own P5-6 precedent), the child is placed in its own
//! process group (`Command::process_group(0)`, unix) and unconditionally
//! group-killed after the call completes (success, error, OR timeout) —
//! see `kill_group` — so a plugin that spawns a persistent worker
//! grandchild (the exact P5-11 LSP-review class this mirrors) never
//! orphans one.
//!
//! **The model's own tool-call arguments are never shell-spliced.** They
//! are serialized to JSON and written to the child's STDIN — never appended
//! to the (fixed, manifest-sourced) command string `build_sandboxed_sh`
//! wraps in `sh -c`. Since the model-controlled content never touches that
//! string at all, there is nothing for it to break out of.
//!
//! Output (stdout/stderr, captured separately) is bounded at
//! [`PLUGIN_TOOL_MAX_OUTPUT_BYTES`] each — reading never stops at the cap
//! (so a flooding child can't wedge on a full OS pipe), only what's
//! RETAINED is bounded, the same "reading never stops, retention does"
//! contract `supercode_runtime::background::CapturedOutput` documents for itself. The
//! whole call is bounded by [`DEFAULT_PLUGIN_TOOL_TIMEOUT_SECS`]; on
//! timeout the process group is killed and a clear timeout error is
//! returned — never a hang.
//!
//! ## Trust model (D-10 — the cardinal requirement)
//! A plugin is arbitrary code execution, so nothing here ever loads OR RUNS
//! one without an affirmative trust decision:
//! - [`crate::Config::plugins_enabled`] (`[capabilities.plugins] enabled`)
//!   is the feature's own master gate — `false` (the default) means
//!   [`discover_and_load`] returns [`PluginLoadOutcome::Disabled`] without
//!   ever touching the filesystem (no directory read, no manifest parse, no
//!   subprocess) — byte-identical to before this module existed.
//! - [`is_trusted`] is the SEPARATE workspace-trust gate
//!   (`[capabilities.trust]`): even with `plugins_enabled = true`, a
//!   workspace whose [`TrustDecision`] isn't [`TrustDecision::Always`] gets
//!   [`PluginLoadOutcome::BlockedPendingTrust`] — loud (a caller-visible,
//!   non-silent outcome; see [`register_into`]'s one-line stderr notice),
//!   never a silent partial load. BP-10: `ask` now has a real consumer —
//!   [`crate::trust`] puts the "trust this workspace?" question to the
//!   `Config::trust_handler` door (the same
//!   `crate::permissions::PermissionsApprovalHandler` every other `Ask` in
//!   this crate uses) and records the answer per project. With NO door
//!   installed, plugin code is still refused: [`is_trusted`] asks about
//!   `crate::trust::TrustSurface::Plugins`, whose undecided answer is `false`
//!   — quarantine-by-default, unchanged, and never the unsafe direction.
//! - The RESOLVER's own hard dependency (`configfile::validate_modules`'s
//!   pre-existing D-10 check, `plugins → trust`) refuses to resolve a config
//!   with `plugins` on and `trust` off at all — this module's own
//!   [`is_trusted`] check is a SECOND, finer-grained gate on top (trust
//!   *enabled* is necessary but not sufficient; it must also have decided
//!   `always`).
//! - `[capabilities.plugins]` (the whole table: `enabled`, `dirs`, any
//!   future contribution key) is wholesale PROJECT-FORBIDDEN — stripped by
//!   both `crate::configfile::sanitize_for_project` and
//!   `crates/cli/src/userconfig.rs`'s own copy, exactly like `hooks`/
//!   `mcp.servers`/`server` (config-borne code execution). A hostile
//!   `.supercode.toml` cannot enable plugins, add a plugin directory, or
//!   loosen the trust decision at all — only the user/global layer (or a
//!   preset extended from it) can.
//!
//! ## Honest, deliberate gaps (build brief: "no declared-but-dead key")
//! - **No pi-TS-extension compatibility.** pi's in-process TypeScript
//!   `ExtensionAPI` (jiti-loaded modules, ~40 events, `registerProvider`/
//!   `setEditorComponent`/overlay UI) cannot and does not run under this
//!   ABI — supercode's `plugins` module has its OWN ABI by design
//!   (COMPOSABLE-HARNESS-DESIGN.md line 1080-1081, an already-accepted
//!   recorded deviation), not an emulation of pi's. An existing pi
//!   extension simply does not run here.
//! - **No marketplace / package installation / `npm install`.** Plugins are
//!   discovered from a local, trusted directory only — there is no
//!   `plugin install <name>` command, no registry client, no network fetch
//!   anywhere in this module. Fetching/installing a plugin (from a
//!   marketplace, npm, or otherwise) is the OPERATOR'S job today (place a
//!   directory under `$SUPERCODE_HOME/plugins`), same posture `lsp`/
//!   `formatters` already take for THEIR external tools (§2 module 28's own
//!   "no auto-spawn/auto-download fleet" gap).
//! - **No extension UI / provider injection.** `registerProvider`,
//!   `setEditorComponent`, overlay UI, and any other in-process
//!   extension-surface hook are impossible by construction under an
//!   out-of-process ABI (a subprocess cannot reach into this process's
//!   UI/provider registry) — not a partially-wired knob, simply not offered.
//! - **Hook FIRING is deferred; hook REGISTRATION is not.** A manifest's
//!   `[[hooks]]` entries are parsed, validated, trust-gated exactly like
//!   `[[tools]]`, and carried on [`LoadedPlugins::hooks`] — but no lifecycle
//!   site in `crates/cli` consults them yet (the same "registerable now,
//!   emission deferred" shape `crates/cli/src/hooks.rs` already ships and
//!   documents for `subagent_start`/`subagent_stop`/`pre_compact`/
//!   `post_compact`, P5-7). [`register_into`] prints a one-time-per-call
//!   warning when a loaded, trusted plugin declares a hook, so this is a
//!   visible, honest gap — never a silent no-op.
//! - **No hash-trust / manifest-change re-prompt (cx§7 "quarantine +
//!   hash-trust").** The weakest form re-evaluates [`is_trusted`] (a
//!   workspace-level decision) on every load, but does not fingerprint an
//!   individual manifest's content to force a re-decision when it changes —
//!   tracked, not hidden: a workspace already at `TrustDecision::Always`
//!   trusts every manifest under its scanned directories, including one
//!   edited after the fact. The `enabled`/`default`/`dirs` knobs this
//!   module DOES expose are all real and wired; this is a scope gap on top
//!   of them, not a dead key.

use std::path::{Path, PathBuf};
use std::time::Duration;

use async_trait::async_trait;
use serde_json::Value;

use crate::error::{Error, Result};
use crate::tools::{Tool, ToolContext};

/// Default wall-clock bound on a single plugin tool invocation — generous
/// for a real script while bounding how long a hanging/misbehaving plugin
/// can stall the agent loop (mirrors `crate::mcp::DEFAULT_MCP_TIMEOUT`'s
/// rationale for the same "config-borne subprocess" trust class).
pub const DEFAULT_PLUGIN_TOOL_TIMEOUT_SECS: u64 = 30;

/// Hardening cap (mirrors `crate::mcp::MCP_MAX_RESPONSE_BYTES`'s rationale,
/// scaled down: a plugin tool result is model-context-bound, not a raw
/// resource fetch): the maximum bytes of stdout (and, separately, stderr)
/// a plugin tool invocation retains — reading never stops at this cap (see
/// the module doc comment), only retention does, so a flooding child can't
/// wedge on a full OS pipe either.
pub const PLUGIN_TOOL_MAX_OUTPUT_BYTES: usize = 1024 * 1024;

/// §2 module 14 `trust`'s `[capabilities.trust] default = "ask" | "always" |
/// "never"` decision (§3.1 schema; every preset that turns trust on sets
/// `default = "ask"`, pi's own `defaultProjectTrust` default). See the
/// module doc comment's "Trust model" section for why, absent an
/// interactive upgrade path in this build, only [`TrustDecision::Always`]
/// actually unlocks plugin loading — `Ask`/`Never` both cleanly refuse
/// rather than silently granting or hanging on a prompt nothing answers.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum TrustDecision {
    /// Prompt before trusting — pi's own default. No interactive handler is
    /// wired in this build (honest gap, see the module doc comment), so
    /// this behaves like [`TrustDecision::Never`] for [`is_trusted`].
    #[default]
    Ask,
    /// Always trusted — the only value [`is_trusted`] accepts today.
    Always,
    /// Never trusted, regardless of anything else.
    Never,
}

impl TrustDecision {
    /// Parse the `"ask"` / `"always"` / `"never"` config strings (§3.1).
    /// Unrecognized text is never silently trusted — the resolver treats an
    /// unparseable value the same as "not `always`" (see [`is_trusted`]),
    /// so an operator typo fails closed, not open.
    pub fn parse(s: &str) -> Option<TrustDecision> {
        match s {
            "ask" => Some(TrustDecision::Ask),
            "always" => Some(TrustDecision::Always),
            "never" => Some(TrustDecision::Never),
            _ => None,
        }
    }
}

/// §2 module 14 `trust` + D-10: is this workspace trusted to load/run
/// config-declared plugin code? See the module doc comment's "Trust model"
/// section. `false` whenever [`crate::Config::trust_enabled`] is `false`
/// (the master gate — matches every OTHER module's "disabled means the
/// setting underneath is never consulted" contract) OR
/// [`crate::Config::trust_default`] isn't exactly [`TrustDecision::Always`].
pub fn is_trusted(config: &crate::Config) -> bool {
    // BP-10 (catalog row "Project/workspace trust gate"): `ask` is no
    // longer a synonym for `never` — `crate::trust` is the consumer of
    // `TrustDecision::Ask` this module's doc comment named as the honest
    // gap. Plugin code is `TrustSurface::Code`, so with NO trust door
    // installed the answer is still `false`, byte-identical to the
    // pre-BP-10 behavior this function had.
    crate::trust::is_trusted(config, crate::trust::TrustSurface::Plugins)
}

/// One `[[tools]]` entry from a `plugin.toml` manifest — see the module doc
/// comment's "Manifest schema" section.
#[derive(Debug, Clone, PartialEq)]
pub struct PluginToolSpec {
    /// The executable to spawn (searched on `PATH`, like any `Command::new`)
    /// — fixed, manifest-sourced data; never the model's own input.
    pub command: String,
    /// Fixed extra arguments to `command` — same trust class as `command`.
    pub args: Vec<String>,
    /// Human description surfaced to the model as the tool's description.
    pub description: String,
    /// JSON Schema for the tool's input object; `{"type": "object"}` when
    /// the manifest doesn't declare one (same default
    /// [`crate::mcp::McpToolDef::input_schema`] uses).
    pub params: Value,
}

/// One `[[hooks]]` entry from a `plugin.toml` manifest — parsed and
/// trust-gated, but not yet wired to firing (see the module doc comment's
/// "Honest, deliberate gaps" section).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PluginHookSpec {
    /// The lifecycle event name (e.g. `"post_tool"`,
    /// `"session_start"` — `crates/cli/src/hooks.rs::HookEvent::as_str`'s
    /// string form).
    pub event: String,
    /// The command to run — same trust class as a tool's `command`
    /// (config-borne, from an already trust-gated manifest).
    pub command: String,
}

/// A parsed, trust-gated-pending `plugin.toml` manifest — see the module
/// doc comment's "Manifest schema" section for the ABI contract this
/// mirrors.
#[derive(Debug, Clone, PartialEq)]
pub struct PluginManifest {
    /// The plugin's own declared name, or its directory name when the
    /// manifest omits `name` (see [`parse_manifest_str`]).
    pub name: String,
    /// Free-form version string (`"0.0.0"` when omitted) — descriptive
    /// only; this module does not interpret or compare versions.
    pub version: String,
    /// `(tool short name, spec)` pairs from `[[tools]]`, in manifest order.
    /// An entry with an empty `name` or `command` is skipped (malformed,
    /// not a crash — same "skip the bad entry" precedent
    /// `crate::configfile::lsp_servers_from_settings` documents for
    /// itself).
    pub tools: Vec<(String, PluginToolSpec)>,
    /// `[[hooks]]` entries, in manifest order. An entry with an empty
    /// `event` or `command` is skipped, same precedent as `tools`.
    pub hooks: Vec<PluginHookSpec>,
}

#[derive(Debug, Default, serde::Deserialize)]
struct RawManifest {
    name: Option<String>,
    #[serde(default)]
    version: String,
    #[serde(default)]
    tools: Vec<RawTool>,
    #[serde(default)]
    hooks: Vec<RawHook>,
}

#[derive(Debug, Default, serde::Deserialize)]
struct RawTool {
    #[serde(default)]
    name: String,
    #[serde(default)]
    command: String,
    #[serde(default)]
    args: Vec<String>,
    #[serde(default)]
    description: String,
    params: Option<Value>,
}

#[derive(Debug, Default, serde::Deserialize)]
struct RawHook {
    #[serde(default)]
    event: String,
    #[serde(default)]
    command: String,
}

/// Parse `text` (a `plugin.toml`'s contents) into a [`PluginManifest`],
/// using `fallback_name` (the plugin's directory name) when the manifest
/// itself doesn't declare `name`. A malformed TOML document is a clean
/// `Err`, never a panic; a malformed INDIVIDUAL `[[tools]]`/`[[hooks]]`
/// entry (empty `name`/`command`/`event`) is silently skipped rather than
/// failing the whole manifest (see [`PluginManifest::tools`]'s doc
/// comment).
pub fn parse_manifest_str(text: &str, fallback_name: &str) -> Result<PluginManifest> {
    let raw: RawManifest = toml::from_str(text)
        .map_err(|e| Error::tool("plugins", format!("parsing manifest: {e}")))?;
    let name = raw
        .name
        .filter(|n| !n.trim().is_empty())
        .unwrap_or_else(|| fallback_name.to_string());
    let version = if raw.version.trim().is_empty() {
        "0.0.0".to_string()
    } else {
        raw.version
    };
    let tools = raw
        .tools
        .into_iter()
        .filter(|t| !t.name.trim().is_empty() && !t.command.trim().is_empty())
        .map(|t| {
            (
                t.name,
                PluginToolSpec {
                    command: t.command,
                    args: t.args,
                    description: t.description,
                    params: t
                        .params
                        .unwrap_or_else(|| serde_json::json!({"type": "object"})),
                },
            )
        })
        .collect();
    let hooks = raw
        .hooks
        .into_iter()
        .filter(|h| !h.event.trim().is_empty() && !h.command.trim().is_empty())
        .map(|h| PluginHookSpec {
            event: h.event,
            command: h.command,
        })
        .collect();
    Ok(PluginManifest {
        name,
        version,
        tools,
        hooks,
    })
}

/// Read and parse `path` (a `plugin.toml` file) — see [`parse_manifest_str`].
/// `fallback_name` is the containing directory's name.
pub fn parse_manifest(path: &Path, fallback_name: &str) -> Result<PluginManifest> {
    let text = std::fs::read_to_string(path)
        .map_err(|e| Error::tool("plugins", format!("reading {}: {e}", path.display())))?;
    parse_manifest_str(&text, fallback_name)
}

/// The always-scanned trusted plugins location:
/// `$SUPERCODE_HOME/plugins` (mirrors `crate::agent::global_instructions_dir`
/// — the same user/global tier every other ambient resource in this crate
/// lives under).
pub fn default_plugins_dir() -> PathBuf {
    crate::agent::global_instructions_dir().join("plugins")
}

/// Scan `dirs` for `<plugin-name>/plugin.toml` manifests — each entry of
/// `dirs` is expected to be a directory whose immediate subdirectories are
/// plugin roots (the same shape `default_plugins_dir()` itself has). Returns
/// `(plugin name, manifest path)` pairs, sorted by name; a name that
/// appears under more than one scanned directory keeps the LAST directory's
/// entry (later/more-specific wins — same precedent
/// `crates/cli/src/main.rs::attach_mcp`'s "same-named entries here WIN"
/// documents for `capabilities.mcp.servers` over `mcp.json`). A `dirs`
/// entry that doesn't exist or isn't readable is silently skipped (not
/// every configured location need exist).
pub fn discover_manifests(dirs: &[PathBuf]) -> Vec<(String, PathBuf)> {
    let mut found: std::collections::BTreeMap<String, PathBuf> = std::collections::BTreeMap::new();
    for dir in dirs {
        let Ok(entries) = std::fs::read_dir(dir) else {
            continue;
        };
        for entry in entries.flatten() {
            let path = entry.path();
            if !path.is_dir() {
                continue;
            }
            let manifest = path.join("plugin.toml");
            if !manifest.is_file() {
                continue;
            }
            let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
                continue;
            };
            found.insert(name.to_string(), manifest);
        }
    }
    found.into_iter().collect()
}

/// SIGKILL an entire process group — reused verbatim from
/// `crate::lsp::kill_process_group` (P5-11's grandchild-orphan fix), the
/// exact same primitive for the exact same reason: a plugin's declared
/// command commonly spawns its OWN worker subprocess, and plain
/// `Child::start_kill` only ever signals the one directly-tracked pid.
#[cfg(unix)]
fn kill_group(pid: u32) {
    crate::lsp::kill_process_group(pid);
}

#[cfg(not(unix))]
fn kill_group(_pid: u32) {}

/// POSIX single-quote a string for safe inclusion in a `sh -c` command —
/// used ONLY for the manifest's OWN fixed `command`/`args` (trusted,
/// config-borne data), never for the model's tool-call arguments, which
/// travel over stdin instead (see the module doc comment's "Subprocess
/// execution model" section). Wrapping in single quotes and escaping any
/// embedded single quote (`'` -> `'\''`) is safe regardless of what
/// characters the string contains.
fn shell_quote(s: &str) -> String {
    format!("'{}'", s.replace('\'', "'\\''"))
}

/// Bounded-read one child pipe to completion — reading never stops at
/// `cap` (so the child can't wedge on a full OS pipe by writing past it),
/// only what's RETAINED does; returns `(text, truncated)`.
async fn drain_capped<R>(mut reader: R, cap: usize) -> (String, bool)
where
    R: tokio::io::AsyncRead + Unpin,
{
    use tokio::io::AsyncReadExt;
    let mut buf: Vec<u8> = Vec::new();
    let mut truncated = false;
    let mut chunk = [0u8; 8192];
    loop {
        match reader.read(&mut chunk).await {
            Ok(0) => break,
            Ok(n) => {
                if buf.len() < cap {
                    let room = cap - buf.len();
                    let take = room.min(n);
                    buf.extend_from_slice(&chunk[..take]);
                    if take < n {
                        truncated = true;
                    }
                } else {
                    truncated = true;
                }
            }
            Err(_) => break,
        }
    }
    (String::from_utf8_lossy(&buf).into_owned(), truncated)
}

/// A model-callable tool backed by one plugin's declared `[[tools]]` entry
/// — see the module doc comment's "Subprocess execution model" section for
/// the full spawn/sandbox/bound/no-orphan contract [`Tool::execute`] below
/// implements.
#[derive(Debug, Clone)]
pub struct PluginTool {
    name: String,
    description: String,
    params: Value,
    command: String,
    args: Vec<String>,
    timeout: Duration,
}

impl PluginTool {
    /// Build the namespaced (`plugin__<plugin>__<tool>`) tool for one
    /// manifest `[[tools]]` entry — mirrors
    /// `crate::mcp::McpServerHandle::tools`'s `mcp__<server>__<tool>`
    /// convention. Uses [`DEFAULT_PLUGIN_TOOL_TIMEOUT_SECS`]; see
    /// `Self::with_timeout` to override (test-only — this module exposes
    /// no config knob for it, matching weakest-form scope).
    pub fn new(plugin_name: &str, tool_name: &str, spec: &PluginToolSpec) -> Self {
        PluginTool {
            name: format!("plugin__{plugin_name}__{tool_name}"),
            description: spec.description.clone(),
            params: spec.params.clone(),
            command: spec.command.clone(),
            args: spec.args.clone(),
            timeout: Duration::from_secs(DEFAULT_PLUGIN_TOOL_TIMEOUT_SECS),
        }
    }
}

#[async_trait]
impl Tool for PluginTool {
    fn name(&self) -> &str {
        &self.name
    }

    fn description(&self) -> &str {
        &self.description
    }

    fn parameters(&self) -> Value {
        self.params.clone()
    }

    async fn execute(&self, args: Value, ctx: &ToolContext) -> Result<String> {
        let quoted = format!(
            "{} {}",
            shell_quote(&self.command),
            self.args
                .iter()
                .map(|a| shell_quote(a))
                .collect::<Vec<_>>()
                .join(" ")
        );
        let mut cmd = crate::tools::build_sandboxed_sh(&quoted, ctx)?;
        cmd.current_dir(&ctx.cwd)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .kill_on_drop(true);
        #[cfg(unix)]
        cmd.process_group(0);

        let mut child = cmd
            .spawn()
            .map_err(|e| Error::tool("plugins", format!("spawn `{}`: {e}", self.command)))?;
        let pid = child.id();

        let args_json = serde_json::to_vec(&args)
            .map_err(|e| Error::tool("plugins", format!("encoding tool args: {e}")))?;
        let mut stdin = child
            .stdin
            .take()
            .ok_or_else(|| Error::tool("plugins", "no stdin"))?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| Error::tool("plugins", "no stdout"))?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| Error::tool("plugins", "no stderr"))?;

        let run = async {
            use tokio::io::AsyncWriteExt;
            // Write the model's tool-call arguments as JSON over stdin —
            // NEVER appended to the command string above (see the module
            // doc comment). Best-effort: a plugin that doesn't read stdin
            // at all must not hang this write forever, so this is inside
            // the same outer timeout as everything else in `run`.
            let _ = stdin.write_all(&args_json).await;
            let _ = stdin.flush().await;
            drop(stdin); // EOF, so a plugin blocked on read(stdin) unblocks

            let stdout_task = tokio::spawn(drain_capped(stdout, PLUGIN_TOOL_MAX_OUTPUT_BYTES));
            let stderr_task = tokio::spawn(drain_capped(stderr, PLUGIN_TOOL_MAX_OUTPUT_BYTES));
            let status = child.wait().await;
            let (out, out_truncated) = stdout_task.await.unwrap_or_default();
            let (err, err_truncated) = stderr_task.await.unwrap_or_default();
            (status, out, out_truncated, err, err_truncated)
        };

        let outcome = tokio::time::timeout(self.timeout, run).await;

        // Unconditional group-kill, success OR timeout OR error — belt-
        // and-suspenders against a surviving worker grandchild even on the
        // clean-exit path (see the module doc comment's no-orphan
        // paragraph; `killpg` on an already-exited leader's group still
        // reaches any surviving member, and is a documented no-op — ESRCH
        // — if the whole group is already gone).
        if let Some(pid) = pid {
            kill_group(pid);
        }

        let (status, out, out_truncated, err, err_truncated) = match outcome {
            Ok(result) => result,
            Err(_) => {
                return Err(Error::tool(
                    "plugins",
                    format!(
                        "plugin tool `{}` timed out after {:?}",
                        self.name, self.timeout
                    ),
                ));
            }
        };

        let mut result = out;
        if out_truncated {
            result.push_str(&format!(
                "\n[plugin output truncated at {PLUGIN_TOOL_MAX_OUTPUT_BYTES} bytes]"
            ));
        }
        if !err.trim().is_empty() {
            result.push_str("\n[stderr]\n");
            result.push_str(&err);
            if err_truncated {
                result.push_str(&format!(
                    "\n[plugin stderr truncated at {PLUGIN_TOOL_MAX_OUTPUT_BYTES} bytes]"
                ));
            }
        }
        match status {
            Ok(s) if !s.success() => {
                result.push_str(&format!(
                    "\n[plugin tool `{}` exited {}]",
                    self.name,
                    s.code().map(|c| c.to_string()).unwrap_or_default()
                ));
            }
            Err(e) => {
                return Err(Error::tool(
                    "plugins",
                    format!("plugin tool `{}` wait failed: {e}", self.name),
                ));
            }
            _ => {}
        }
        Ok(result)
    }
}

/// Every trusted, loaded plugin's contributions — [`discover_and_load`]'s
/// success case.
#[derive(Debug, Default)]
pub struct LoadedPlugins {
    /// Ready-to-register tools, in discovery order.
    pub tools: Vec<PluginTool>,
    /// `(plugin name, hook spec)` pairs — see the module doc comment's
    /// "Honest, deliberate gaps" section for why these are carried but not
    /// yet fired.
    pub hooks: Vec<(String, PluginHookSpec)>,
    /// Names of every plugin whose manifest parsed successfully.
    pub loaded_plugin_names: Vec<String>,
    /// Human-readable warnings for manifests that failed to parse — never
    /// fatal to the OTHER plugins' load (one bad manifest doesn't sink the
    /// rest), but never silently swallowed either.
    pub warnings: Vec<String>,
}

/// The result of one [`discover_and_load`] call — see the module doc
/// comment's "Trust model" section for what drives each variant.
#[derive(Debug)]
pub enum PluginLoadOutcome {
    /// `[capabilities.plugins] enabled` is `false` (the default) — nothing
    /// was touched: no directory read, no manifest parsed, no subprocess
    /// spawned.
    Disabled,
    /// `enabled = true`, but [`is_trusted`] said no — nothing was loaded.
    /// Distinct from [`PluginLoadOutcome::Disabled`] so a caller can report
    /// this honestly (quarantined pending trust) rather than looking
    /// identical to the feature being off.
    BlockedPendingTrust,
    /// Trusted and enabled — every discovered manifest was at least
    /// attempted; see [`LoadedPlugins::warnings`] for any that failed.
    Loaded(LoadedPlugins),
}

/// The single entry point: resolve `config`'s plugin gates
/// ([`crate::Config::plugins_enabled`], [`is_trusted`]) and, only if both
/// pass, discover + parse every manifest under the scanned directories.
/// See the module doc comment's "Trust model" section for the full D-10
/// contract this enforces.
pub fn discover_and_load(config: &crate::Config) -> PluginLoadOutcome {
    if !config.plugins_enabled {
        return PluginLoadOutcome::Disabled;
    }
    if !is_trusted(config) {
        return PluginLoadOutcome::BlockedPendingTrust;
    }
    let mut dirs = vec![default_plugins_dir()];
    dirs.extend(config.plugins_dirs.iter().cloned());
    let manifests = discover_manifests(&dirs);

    let mut loaded = LoadedPlugins::default();
    for (name, path) in manifests {
        match parse_manifest(&path, &name) {
            Ok(manifest) => {
                for (tool_name, spec) in &manifest.tools {
                    loaded
                        .tools
                        .push(PluginTool::new(&manifest.name, tool_name, spec));
                }
                for hook in &manifest.hooks {
                    loaded.hooks.push((manifest.name.clone(), hook.clone()));
                }
                loaded.loaded_plugin_names.push(manifest.name);
            }
            Err(e) => {
                loaded
                    .warnings
                    .push(format!("plugin `{name}` ({}): {e}", path.display()));
            }
        }
    }
    PluginLoadOutcome::Loaded(loaded)
}

/// Register every trusted, loaded plugin tool into `registry` — the single
/// production choke point `crate::agent::Agent::with_parts` calls (so every
/// `Agent` construction path gets plugin tools "for free", the same way
/// `crate::lsp`/`crate::formatters`/`crate::checkpoint`'s observers are
/// wired unconditionally from `crate::agent::build_tool_context`). A no-op,
/// with nothing printed, when [`crate::Config::plugins_enabled`] is `false`
/// (default-off byte-identity). When enabled but not yet trusted, prints
/// ONE line to stderr (never silent — see [`PluginLoadOutcome::BlockedPendingTrust`]'s
/// doc comment) and registers nothing. When loaded, registers every tool
/// and prints a one-time-per-call warning for any hook a trusted plugin
/// declared (see the module doc comment's "Honest, deliberate gaps"
/// section) plus any manifest parse warning.
pub fn register_into(config: &crate::Config, registry: &mut crate::tools::ToolRegistry) {
    match discover_and_load(config) {
        PluginLoadOutcome::Disabled => {}
        PluginLoadOutcome::BlockedPendingTrust => {
            eprintln!(
                "warning: [capabilities.plugins] is enabled but this workspace is not trusted \
                 ([capabilities.trust] default must be \"always\") — no plugin was loaded"
            );
        }
        PluginLoadOutcome::Loaded(loaded) => {
            for warning in &loaded.warnings {
                eprintln!("warning: {warning}");
            }
            for (plugin, hook) in &loaded.hooks {
                eprintln!(
                    "warning: plugin `{plugin}`'s `{}` hook is registered but is not yet \
                     emitted in this build (no-op) — see crate::plugins's module doc comment",
                    hook.event
                );
            }
            for tool in loaded.tools {
                registry.register(tool);
            }
        }
    }
}