supercode_harness/

plugins.rs

//! 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 `crate::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. **Honest gap:** this build has no
//!   interactive "trust this workspace?" prompt UI wired up anywhere (no
//!   consumer of `TrustDecision::Ask` exists yet, matching
//!   `crate::mcp::HeadlessElicitationHandler`'s own "deny-default, pending a
//!   real interactive handler" precedent) — so `ask` (pi's own own default)
//!   and `never` both cleanly refuse to load in this build; only an
//!   operator explicitly setting `default = "always"` in their trusted
//!   user/global config unlocks plugin loading. This narrows what pi's own
//!   `defaultProjectTrust = "ask"` would otherwise interactively allow, in
//!   the safe direction (quarantine-by-default), never the unsafe one.
//! - 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 {
    config.trust_enabled && config.trust_default == TrustDecision::Always
}

/// 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),
        }
    }

    /// Test-only: override the per-call timeout so timeout/bounded-ness
    /// tests don't need to wait out the real
    /// [`DEFAULT_PLUGIN_TOOL_TIMEOUT_SECS`].
    #[cfg(test)]
    fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }
}

#[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);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn tmp(tag: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!(
            "supercode-plugins-test-{tag}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    fn write_manifest(dir: &Path, name: &str, toml: &str) -> PathBuf {
        let plugin_dir = dir.join(name);
        std::fs::create_dir_all(&plugin_dir).unwrap();
        let manifest = plugin_dir.join("plugin.toml");
        std::fs::write(&manifest, toml).unwrap();
        manifest
    }

    // ---- TrustDecision / is_trusted ----------------------------------

    #[test]
    fn trust_decision_parse_round_trips_known_values() {
        assert_eq!(TrustDecision::parse("ask"), Some(TrustDecision::Ask));
        assert_eq!(TrustDecision::parse("always"), Some(TrustDecision::Always));
        assert_eq!(TrustDecision::parse("never"), Some(TrustDecision::Never));
        assert_eq!(TrustDecision::parse("bogus"), None);
    }

    #[test]
    fn is_trusted_requires_both_trust_enabled_and_default_always() {
        let base = crate::Config::builder().model("m").build();
        assert!(!is_trusted(&base), "trust disabled by default");

        let enabled_ask = crate::Config::builder()
            .model("m")
            .trust_enabled(true)
            .trust_default(TrustDecision::Ask)
            .build();
        assert!(
            !is_trusted(&enabled_ask),
            "trust enabled but default=ask must NOT be trusted (no interactive upgrade wired)"
        );

        let enabled_never = crate::Config::builder()
            .model("m")
            .trust_enabled(true)
            .trust_default(TrustDecision::Never)
            .build();
        assert!(!is_trusted(&enabled_never));

        let enabled_always = crate::Config::builder()
            .model("m")
            .trust_enabled(true)
            .trust_default(TrustDecision::Always)
            .build();
        assert!(is_trusted(&enabled_always));

        let disabled_always = crate::Config::builder()
            .model("m")
            .trust_enabled(false)
            .trust_default(TrustDecision::Always)
            .build();
        assert!(
            !is_trusted(&disabled_always),
            "trust_enabled=false must gate regardless of trust_default"
        );
    }

    // ---- manifest parsing ----------------------------------------------

    #[test]
    fn parse_manifest_str_parses_tools_and_hooks() {
        let toml = r#"
name = "demo"
version = "1.2.3"

[[tools]]
name = "greet"
command = "echo"
args = ["hi"]
description = "says hi"
params = { type = "object" }

[[hooks]]
event = "post_tool"
command = "notify.sh"
"#;
        let m = parse_manifest_str(toml, "fallback").unwrap();
        assert_eq!(m.name, "demo");
        assert_eq!(m.version, "1.2.3");
        assert_eq!(m.tools.len(), 1);
        assert_eq!(m.tools[0].0, "greet");
        assert_eq!(m.tools[0].1.command, "echo");
        assert_eq!(m.tools[0].1.args, vec!["hi".to_string()]);
        assert_eq!(m.hooks.len(), 1);
        assert_eq!(m.hooks[0].event, "post_tool");
        assert_eq!(m.hooks[0].command, "notify.sh");
    }

    #[test]
    fn parse_manifest_str_falls_back_to_directory_name_when_name_absent() {
        let m = parse_manifest_str("version = \"0.1.0\"", "my-dir-name").unwrap();
        assert_eq!(m.name, "my-dir-name");
    }

    #[test]
    fn parse_manifest_str_defaults_version_and_params_when_absent() {
        let toml = r#"
[[tools]]
name = "t"
command = "echo"
"#;
        let m = parse_manifest_str(toml, "p").unwrap();
        assert_eq!(m.version, "0.0.0");
        assert_eq!(m.tools[0].1.params, serde_json::json!({"type": "object"}));
    }

    #[test]
    fn parse_manifest_str_skips_malformed_entries_without_failing_the_manifest() {
        let toml = r#"
[[tools]]
name = ""
command = "echo"

[[tools]]
name = "ok"
command = ""

[[tools]]
name = "good"
command = "echo"

[[hooks]]
event = ""
command = "x"
"#;
        let m = parse_manifest_str(toml, "p").unwrap();
        assert_eq!(m.tools.len(), 1, "only the fully-valid tool survives");
        assert_eq!(m.tools[0].0, "good");
        assert!(m.hooks.is_empty());
    }

    #[test]
    fn parse_manifest_str_rejects_malformed_toml() {
        assert!(parse_manifest_str("not valid toml [[[", "p").is_err());
    }

    // ---- discovery -------------------------------------------------------

    #[test]
    fn discover_manifests_finds_plugin_toml_under_immediate_subdirs() {
        let dir = tmp("discover");
        write_manifest(&dir, "alpha", "name = \"alpha\"\n");
        write_manifest(&dir, "beta", "name = \"beta\"\n");
        // A subdirectory with no plugin.toml must be ignored.
        std::fs::create_dir_all(dir.join("not-a-plugin")).unwrap();

        let found = discover_manifests(std::slice::from_ref(&dir));
        let names: Vec<&str> = found.iter().map(|(n, _)| n.as_str()).collect();
        assert_eq!(names, vec!["alpha", "beta"], "sorted by name");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn discover_manifests_missing_dir_is_silently_skipped() {
        let missing = tmp("missing-parent").join("does-not-exist");
        let found = discover_manifests(&[missing]);
        assert!(found.is_empty());
    }

    #[test]
    fn discover_manifests_later_dir_wins_on_name_collision() {
        let dir_a = tmp("collide-a");
        let dir_b = tmp("collide-b");
        write_manifest(&dir_a, "dup", "version = \"1.0.0\"\n");
        write_manifest(&dir_b, "dup", "version = \"2.0.0\"\n");
        let found = discover_manifests(&[dir_a.clone(), dir_b.clone()]);
        assert_eq!(found.len(), 1);
        let (_, path) = &found[0];
        assert!(path.starts_with(&dir_b), "later dir must win");
        std::fs::remove_dir_all(&dir_a).ok();
        std::fs::remove_dir_all(&dir_b).ok();
    }

    // ---- discover_and_load / register_into gating -----------------------

    #[test]
    fn discover_and_load_is_disabled_when_plugins_off_default_off_byte_identity() {
        let config = crate::Config::builder().model("m").build();
        assert!(!config.plugins_enabled);
        assert!(matches!(
            discover_and_load(&config),
            PluginLoadOutcome::Disabled
        ));
    }

    #[test]
    fn discover_and_load_is_blocked_pending_trust_when_untrusted() {
        let config = crate::Config::builder()
            .model("m")
            .plugins_enabled(true)
            .trust_enabled(true)
            .trust_default(TrustDecision::Ask)
            .build();
        assert!(matches!(
            discover_and_load(&config),
            PluginLoadOutcome::BlockedPendingTrust
        ));
    }

    #[test]
    fn discover_and_load_is_blocked_when_trust_module_itself_is_off() {
        let config = crate::Config::builder()
            .model("m")
            .plugins_enabled(true)
            .trust_enabled(false)
            .build();
        assert!(matches!(
            discover_and_load(&config),
            PluginLoadOutcome::BlockedPendingTrust
        ));
    }

    #[test]
    fn discover_and_load_loads_tools_from_a_trusted_configured_dir() {
        let dir = tmp("load-trusted");
        write_manifest(
            &dir,
            "demo",
            "name = \"demo\"\n\n[[tools]]\nname = \"echo_it\"\ncommand = \"echo\"\n",
        );
        let config = crate::Config::builder()
            .model("m")
            .plugins_enabled(true)
            .plugins_dirs(vec![dir.clone()])
            .trust_enabled(true)
            .trust_default(TrustDecision::Always)
            .build();
        match discover_and_load(&config) {
            PluginLoadOutcome::Loaded(loaded) => {
                assert_eq!(loaded.loaded_plugin_names, vec!["demo".to_string()]);
                assert_eq!(loaded.tools.len(), 1);
                assert_eq!(loaded.tools[0].name(), "plugin__demo__echo_it");
            }
            other => panic!("expected Loaded, got {other:?}"),
        }
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn discover_and_load_records_a_warning_for_an_unparseable_manifest_without_failing_others() {
        let dir = tmp("load-warn");
        write_manifest(&dir, "bad", "not valid toml [[[");
        write_manifest(
            &dir,
            "good",
            "name = \"good\"\n\n[[tools]]\nname = \"t\"\ncommand = \"echo\"\n",
        );
        let config = crate::Config::builder()
            .model("m")
            .plugins_enabled(true)
            .plugins_dirs(vec![dir.clone()])
            .trust_enabled(true)
            .trust_default(TrustDecision::Always)
            .build();
        match discover_and_load(&config) {
            PluginLoadOutcome::Loaded(loaded) => {
                assert_eq!(loaded.tools.len(), 1, "the good plugin still loads");
                assert_eq!(loaded.warnings.len(), 1);
                assert!(loaded.warnings[0].contains("bad"));
            }
            other => panic!("expected Loaded, got {other:?}"),
        }
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn register_into_is_a_true_noop_when_plugins_disabled() {
        let config = crate::Config::builder().model("m").build();
        let mut registry = crate::tools::ToolRegistry::new();
        register_into(&config, &mut registry);
        assert_eq!(registry.len(), 0);
    }

    #[test]
    fn register_into_registers_nothing_when_untrusted() {
        let dir = tmp("register-untrusted");
        write_manifest(
            &dir,
            "demo",
            "name = \"demo\"\n\n[[tools]]\nname = \"t\"\ncommand = \"echo\"\n",
        );
        let config = crate::Config::builder()
            .model("m")
            .plugins_enabled(true)
            .plugins_dirs(vec![dir.clone()])
            .trust_enabled(true)
            .trust_default(TrustDecision::Never)
            .build();
        let mut registry = crate::tools::ToolRegistry::new();
        register_into(&config, &mut registry);
        assert_eq!(registry.len(), 0, "untrusted plugin must never register");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn register_into_registers_trusted_tools() {
        let dir = tmp("register-trusted");
        write_manifest(
            &dir,
            "demo",
            "name = \"demo\"\n\n[[tools]]\nname = \"t\"\ncommand = \"echo\"\n",
        );
        let config = crate::Config::builder()
            .model("m")
            .plugins_enabled(true)
            .plugins_dirs(vec![dir.clone()])
            .trust_enabled(true)
            .trust_default(TrustDecision::Always)
            .build();
        let mut registry = crate::tools::ToolRegistry::new();
        register_into(&config, &mut registry);
        assert_eq!(registry.len(), 1);
        assert!(registry.get("plugin__demo__t").is_some());
        std::fs::remove_dir_all(&dir).ok();
    }

    // ---- PluginTool::execute: subprocess model ---------------------------

    fn ctx(cwd: PathBuf) -> ToolContext {
        ToolContext::new(cwd)
    }

    #[tokio::test]
    async fn plugin_tool_executes_and_returns_stdout() {
        let dir = tmp("exec-basic");
        let spec = PluginToolSpec {
            command: "echo".to_string(),
            args: vec!["hello-plugin".to_string()],
            description: String::new(),
            params: serde_json::json!({"type": "object"}),
        };
        let tool = PluginTool::new("demo", "say", &spec);
        let out = tool
            .execute(serde_json::json!({}), &ctx(dir.clone()))
            .await
            .unwrap();
        assert!(out.contains("hello-plugin"), "{out}");
        std::fs::remove_dir_all(&dir).ok();
    }

    #[tokio::test]
    async fn plugin_tool_args_are_never_shell_spliced() {
        // A tool-call argument containing shell metacharacters and a
        // `touch` payload must be INERT — it travels over stdin, never
        // concatenated into the sh -c command string.
        let dir = tmp("exec-no-splice");
        let marker = dir.join("PWNED");
        let spec = PluginToolSpec {
            command: "cat".to_string(),
            args: vec![],
            description: String::new(),
            params: serde_json::json!({"type": "object"}),
        };
        let tool = PluginTool::new("demo", "cat_args", &spec);
        let evil = format!("$(touch {})", marker.display());
        let out = tool
            .execute(serde_json::json!({"payload": evil}), &ctx(dir.clone()))
            .await
            .unwrap();
        assert!(
            out.contains("touch"),
            "cat should echo the literal, unevaluated JSON back: {out}"
        );
        assert!(
            !marker.exists(),
            "shell metacharacters in tool-call args must never be evaluated"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    #[tokio::test]
    async fn plugin_tool_bounds_output_and_marks_it_truncated() {
        let dir = tmp("exec-bounded");
        // `yes` floods stdout forever — this proves the read completes
        // (never hangs/OOMs) and is retained only up to the cap.
        let spec = PluginToolSpec {
            command: "sh".to_string(),
            args: vec![
                "-c".to_string(),
                format!(
                    "head -c {} /dev/zero | tr '\\0' 'a'",
                    PLUGIN_TOOL_MAX_OUTPUT_BYTES * 2
                ),
            ],
            description: String::new(),
            params: serde_json::json!({"type": "object"}),
        };
        let tool = PluginTool::new("demo", "flood", &spec);
        let out = tool
            .execute(serde_json::json!({}), &ctx(dir.clone()))
            .await
            .unwrap();
        assert!(out.contains("truncated"), "{}", &out[..out.len().min(200)]);
        assert!(
            out.len() < PLUGIN_TOOL_MAX_OUTPUT_BYTES * 2,
            "retained output must be bounded well below what the child wrote"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    #[tokio::test]
    async fn plugin_tool_timeout_is_bounded_and_reported() {
        let dir = tmp("exec-timeout");
        let spec = PluginToolSpec {
            command: "sleep".to_string(),
            args: vec!["3600".to_string()],
            description: String::new(),
            params: serde_json::json!({"type": "object"}),
        };
        let tool = PluginTool::new("demo", "hang", &spec).with_timeout(Duration::from_millis(300));
        let started = std::time::Instant::now();
        let result = tokio::time::timeout(
            Duration::from_secs(10),
            tool.execute(serde_json::json!({}), &ctx(dir.clone())),
        )
        .await
        .expect("must not hang past the plugin tool's own timeout");
        assert!(result.is_err(), "a hanging plugin tool must error out");
        assert!(
            result.unwrap_err().to_string().contains("timed out"),
            "error should say it timed out"
        );
        assert!(
            started.elapsed() < Duration::from_secs(5),
            "took {:?}, expected to bail out near the configured timeout",
            started.elapsed()
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    /// No-orphan proof (mirrors `crate::lsp`'s P5-11 grandchild test): a
    /// plugin tool that spawns its own persistent worker grandchild must
    /// not leave it running after the tool call completes.
    #[cfg(unix)]
    #[tokio::test]
    async fn plugin_tool_reaps_grandchild_worker_processes() {
        let dir = tmp("exec-grandchild");
        let pidfile = dir.join("worker.pid");
        let script = dir.join("spawn_worker.sh");
        std::fs::write(
            &script,
            format!(
                "#!/bin/sh\nsleep 3600 &\necho $! > {}\nwait\n",
                pidfile.display()
            ),
        )
        .unwrap();
        let spec = PluginToolSpec {
            command: "sh".to_string(),
            args: vec![script.to_string_lossy().into_owned()],
            description: String::new(),
            params: serde_json::json!({"type": "object"}),
        };
        let tool = PluginTool::new("demo", "spawns_worker", &spec)
            .with_timeout(Duration::from_millis(300));

        // Race the tool call against a short timeout via a background task
        // so we can inspect the grandchild pid while the parent is still
        // "running" (the script's own `wait` blocks until the plugin
        // subprocess's whole group is killed).
        let handle = tokio::spawn({
            let dir = dir.clone();
            async move { tool.execute(serde_json::json!({}), &ctx(dir)).await }
        });

        let mut grandchild_pid: Option<i32> = None;
        for _ in 0..150 {
            if let Ok(s) = std::fs::read_to_string(&pidfile) {
                if let Ok(pid) = s.trim().parse::<i32>() {
                    grandchild_pid = Some(pid);
                    break;
                }
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let grandchild_pid = grandchild_pid.expect("worker must have recorded its pid");
        assert!(
            unsafe { libc::kill(grandchild_pid, 0) == 0 },
            "grandchild worker must be alive before the plugin tool call completes"
        );

        // The script's own `sh` never exits on its own (it `wait`s on the
        // backgrounded sleep) — the ONLY thing that ends this call is our
        // own timeout's group-kill, which is exactly the no-orphan path
        // under test.
        let _ = tokio::time::timeout(Duration::from_secs(10), handle).await;

        let mut still_alive = true;
        for _ in 0..150 {
            let alive = unsafe { libc::kill(grandchild_pid, 0) == 0 };
            if !alive {
                still_alive = false;
                break;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert!(
            !still_alive,
            "grandchild worker pid {grandchild_pid} must be dead — it must not orphan"
        );
        std::fs::remove_dir_all(&dir).ok();
    }
}