cpex-hosts-python 0.3.1

// Location: ./crates/cpex-hosts-python/src/plugin.rs
// Copyright 2026
// SPDX-License-Identifier: Apache-2.0
// Authors: Ted Habeck
//
// `IsolatedPythonPlugin` — the plugin object the factory produces, and
// `PythonHookAdapter` — the per-hook handler that drives one invocation.
//
// The plugin owns the venv manager and the worker client and implements the
// lifecycle half of the contract (initialize / shutdown). The adapter
// implements `AnyHookHandler` directly rather than going through
// `TypedHandlerAdapter`: that adapter downcasts to one fixed payload type,
// but this host forwards whatever payload arrives, serialized, to a worker
// that reconstructs it on the Python side. The concrete type is not known at
// Rust compile time, so there is nothing to downcast to.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use async_trait::async_trait;
use ppe::praxis_policy_core::{
    context::PluginContext,
    error::PluginError,
    extensions::Extensions,
    hooks::payload::PluginPayload,
    plugin::{Plugin, PluginConfig},
    registry::AnyHookHandler,
};
use serde::Deserialize;

use crate::error::HostError;
use crate::legacy::cmf_shim;
use crate::venv::VenvManager;
use crate::worker::WorkerClient;
use crate::{conversion, credential_payloads, credentials};

/// Default cap on the serialized size of one outbound task, in bytes.
///
/// Mirrors `client.py`'s `max_content_size` default. The bound exists so a
/// pathological payload fails fast on this side rather than after the worker
/// has already buffered it.
const DEFAULT_MAX_CONTENT_SIZE: usize = 10_000_000;

/// Default per-invocation timeout, in seconds.
///
/// Mirrors `venv_comm.py`'s `send_task` default. A plugin entry may override
/// it; absent both, the executor's global `plugin_settings.plugin_timeout`
/// still bounds the call from outside.
const DEFAULT_TIMEOUT_SECS: u64 = 30;

/// Path to the worker script, relative to the venv's installed `cpex`.
///
/// The worker ships inside the venv via the plugin's `cpex` dependency, so
/// the host does not carry a copy of it.
const DEFAULT_SCRIPT_PATH: &str = "cpex/framework/isolated/worker.py";

/// How many times a dead worker may be respawned before the plugin gives up.
///
/// Bounded rather than unlimited: a worker that dies on every launch is a
/// broken plugin (a bad dependency pin, a syntax error, an import that raises),
/// and retrying it forever converts a fast, legible failure into an
/// indefinitely wedged hook that hammers the machine. Three attempts covers the
/// case respawn exists for — a worker killed by an OOM reaper, a crash inside
/// one plugin invocation — without papering over a plugin that cannot start.
const MAX_RESPAWN_ATTEMPTS: u32 = 3;

/// Base delay before the first respawn attempt.
///
/// Doubled per attempt (100ms, 200ms, 400ms). The backoff exists so a worker
/// that dies immediately on launch cannot spin the host in a tight
/// spawn/crash loop; the values stay small because a live request is waiting
/// behind them and the executor's own timeout is still running.
const RESPAWN_BACKOFF_BASE_MS: u64 = 100;

/// Interval after which a plugin that has exhausted its respawn budget is
/// allowed to try again.
///
/// Without this, one bad patch of the worker's environment would permanently
/// poison the plugin for the process's lifetime, even after the underlying
/// cause is fixed. With it, a plugin that has been failing settles into one
/// retry per minute rather than either hammering or staying dead forever.
const RESPAWN_BUDGET_RESET_SECS: u64 = 60;

/// Directory, relative to the project root, holding installed Python plugins.
///
/// This is the sole source of the worker's `sys.path` entries and of the
/// directory the venv is built under. It is *not* configurable per plugin: a
/// `plugin_dirs:` key inside a plugin's `config:` block is ignored (see
/// `VenvConfig`). A top-level `plugin_dirs:` YAML key is rejected by the
/// config loader.
///
/// # Why a fixed default rather than a config key
///
/// `cpex plugin install` writes the plugin into `<project root>/plugins/` and
/// builds its venv there. Both the Python CLI and this host therefore already
/// agree on the location; making it configurable only creates a way for the
/// two to disagree, and the failure mode is an import error inside the worker
/// at invoke time, far from the config that caused it.
pub const DEFAULT_PLUGIN_DIR: &str = "plugins";

/// The project root the `plugins` directory is resolved against.
///
/// The process working directory. That is not an arbitrary choice: the
/// worker's own `ALLOWED_PLUGIN_DIRS` allowlist accepts `os.getcwd()`, and the
/// worker inherits this host's cwd, so a directory resolved this way is
/// importable by construction. Anchoring anywhere else would produce paths the
/// worker rejects.
fn project_root() -> PathBuf {
    // A cwd that cannot be read is not recoverable here, and returning a
    // relative path keeps the failure legible: the venv layout error names
    // `plugins`, which is what an operator would look for.
    std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
}

/// The plugin directory list the worker and venv layout both use.
///
/// One entry: `<project root>/plugins`. Absolute, so the value does not shift
/// if something later changes the process cwd between `initialize()` and an
/// invocation.
fn default_plugin_dirs() -> Vec<String> {
    vec![
        project_root()
            .join(DEFAULT_PLUGIN_DIR)
            .display()
            .to_string(),
    ]
}

/// Global-state key a host can set to propagate its own request id into the
/// worker's `GlobalContext.request_id`.
///
/// The Rust `PluginContext` has no dedicated request-id field, but the Python
/// `GlobalContext` requires one. Reading it from global state lets a gateway
/// that tracks request ids keep them consistent across the boundary.
pub const GLOBAL_REQUEST_ID_KEY: &str = "request_id";

/// Capability names the Python `Capability` enum models
/// (`cpex/framework/extensions/tiers.py`).
///
/// This is deliberately a mirror of the *other* side's vocabulary rather than
/// this host's own. `PluginConfig`'s `capabilities` validator raises on any name
/// it does not recognize, and that happens during config construction inside the
/// worker, so an unrecognized name is a dead hook rather than a narrower view.
/// See [`IsolatedPythonPlugin::worker_capabilities`].
///
/// The host's vocabulary is the superset: `read_all`, `read_client`,
/// `read_workload`, `read_inbound_credentials`, and `read_delegated_tokens` have
/// no Python counterpart yet and are intentionally absent here. When the Python
/// enum grows one, adding it here is what lets it cross the boundary.
const WORKER_KNOWN_CAPABILITIES: &[&str] = &[
    "read_subject",
    "read_roles",
    "read_teams",
    "read_claims",
    "read_permissions",
    "read_agent",
    "read_headers",
    "write_headers",
    "read_labels",
    "append_labels",
    "read_delegation",
    "append_delegation",
];

/// Venv-relevant settings, parsed from a plugin entry's opaque `config` map.
///
/// `PluginConfig` has no fields for a class name, a content-size cap, or a
/// per-plugin timeout — those are host-specific, so they live in the
/// free-form `config` value and are parsed here.
///
/// # `plugin_dirs` is deliberately absent
///
/// A `plugin_dirs:` key in this block is **ignored**; a top-level
/// `plugin_dirs:` YAML key fails the config load. Plugin directories are not configurable: the host
/// always uses `<project root>/plugins`, which is where `cpex plugin install`
/// puts them. See `DEFAULT_PLUGIN_DIR`. Serde ignores unknown fields, so an
/// existing config that still carries the key parses without error — the value
/// simply has no effect.
#[derive(Debug, Clone, Deserialize)]
pub struct VenvConfig {
    /// Fully-qualified Python class implementing the plugin, e.g.
    /// `my_pkg.filters.PiiFilter`. Required: the worker needs it to import
    /// and instantiate the plugin, and the venv cache is keyed on it.
    pub class_name: String,

    /// Requirements file to install into the venv, relative to the package
    /// root. Optional — a plugin installed by FQN conversion gets its
    /// dependencies from the install channel, not a requirements file, and
    /// then the manifest version plus hash is the sole cache signal.
    #[serde(default)]
    pub requirements_file: Option<String>,

    /// Worker script path inside the venv. Defaults to the framework's
    /// `worker.py`; overridable for tests and for a host that vendors its own.
    #[serde(default = "default_script_path")]
    pub script_path: String,

    /// Cap on one serialized outbound task, in bytes.
    #[serde(default = "default_max_content_size")]
    pub max_content_size: usize,

    /// Per-invocation timeout, in seconds. Falls back to the framework
    /// default when absent.
    #[serde(default = "default_timeout_secs")]
    pub timeout_secs: u64,
}

fn default_script_path() -> String {
    DEFAULT_SCRIPT_PATH.to_string()
}

fn default_max_content_size() -> usize {
    DEFAULT_MAX_CONTENT_SIZE
}

fn default_timeout_secs() -> u64 {
    DEFAULT_TIMEOUT_SECS
}

impl VenvConfig {
    /// Parse the venv settings out of a plugin entry's `config` map.
    ///
    /// A missing or malformed `class_name` is a config error, not a panic —
    /// the engine surfaces it against the plugin name at load time.
    pub fn from_plugin_config(config: &PluginConfig) -> Result<Self, Box<PluginError>> {
        let raw = config
            .config
            .clone()
            .unwrap_or_else(|| serde_json::json!({}));

        serde_json::from_value(raw).map_err(|e| {
            PluginError::Config {
                message: format!(
                    "plugin '{}' (isolated_venv): invalid `config:` block — {}. \
                     `class_name` is required (the fully-qualified Python class, \
                     e.g. my_pkg.filters.PiiFilter)",
                    config.name, e
                ),
            }
            .boxed()
        })
    }
}

/// A Python plugin running out-of-process in its own virtualenv.
pub struct IsolatedPythonPlugin {
    /// The authoritative config entry, kept for `Plugin::config()`.
    config: PluginConfig,

    /// Venv settings parsed from the entry's `config` map.
    venv_config: VenvConfig,

    /// Directories the worker prepends to `sys.path`, and under whose first
    /// entry the venv is built.
    ///
    /// Host-owned rather than config-derived: `<project root>/plugins`, per
    /// `DEFAULT_PLUGIN_DIR`. Held here rather than recomputed per invocation so
    /// the value stays fixed for the plugin's lifetime even if the process cwd
    /// changes after construction.
    plugin_dirs: Vec<String>,

    /// Builds and caches the venv. `None` only when the layout cannot be
    /// resolved (an empty dir list, or a `class_name` with no package
    /// segment), which is valid solely for a `worker_override` test setup.
    venv: Option<VenvManager>,

    /// The live worker, populated by `initialize()`.
    worker: tokio::sync::RwLock<Option<Arc<WorkerClient>>>,

    /// Absolute worker-script path that bypasses venv resolution. Tests point
    /// this at a stub; production leaves it unset and resolves from the venv.
    worker_override: Option<PathBuf>,

    /// Working directory for the worker process.
    ///
    /// Defaults to the host's own cwd, matching `venv_comm.py`'s
    /// `cwd=os.getcwd()`. This is load-bearing twice over: the worker's
    /// `ALLOWED_PLUGIN_DIRS` allowlist accepts `os.getcwd()`, so a plugin dir
    /// outside it is rejected outright; and a plugin's relative-path side
    /// effects land here.
    worker_cwd: Option<PathBuf>,

    /// Respawn bookkeeping for a worker that dies after a successful
    /// `initialize()`.
    respawn: tokio::sync::Mutex<RespawnState>,

    /// Set by `shutdown()` before it empties the worker slot, and cleared by
    /// `initialize()`. A respawn already under way when the plugin shuts down
    /// checks it, so it cannot launch and install a worker nothing will stop.
    shut_down: std::sync::atomic::AtomicBool,
}

/// How many respawns this plugin has spent, and when the budget last reset.
///
/// Held behind its own mutex rather than folded into the worker lock so the
/// accounting survives the worker slot being replaced, and so two concurrent
/// invocations that both notice the same death cooperate: the first through
/// respawns, the second finds a live worker and spends nothing.
#[derive(Debug)]
struct RespawnState {
    /// Respawn attempts spent since the last budget reset.
    attempts: u32,
    /// When the budget was last reset, for the [`RESPAWN_BUDGET_RESET_SECS`]
    /// window. `None` until the first respawn.
    window_started: Option<std::time::Instant>,
}

impl RespawnState {
    fn new() -> Self {
        Self {
            attempts: 0,
            window_started: None,
        }
    }

    /// Reserve one respawn attempt, or report the budget is spent.
    ///
    /// Resets the counter first when the window has elapsed, so a plugin that
    /// failed hard an hour ago is not still being punished for it.
    fn try_reserve(&mut self) -> Option<u32> {
        let now = std::time::Instant::now();
        let expired = self
            .window_started
            .is_some_and(|s| now.duration_since(s).as_secs() >= RESPAWN_BUDGET_RESET_SECS);

        if expired {
            self.attempts = 0;
            self.window_started = None;
        }

        if self.attempts >= MAX_RESPAWN_ATTEMPTS {
            return None;
        }

        self.attempts += 1;
        self.window_started.get_or_insert(now);
        Some(self.attempts)
    }

    /// Return the budget to full after a respawn that produced a live worker.
    fn record_success(&mut self) {
        self.attempts = 0;
        self.window_started = None;
    }
}

impl IsolatedPythonPlugin {
    /// Build a plugin from its config entry, parsing the venv settings.
    pub fn from_config(config: &PluginConfig) -> Result<Self, Box<PluginError>> {
        let venv_config = VenvConfig::from_plugin_config(config)?;

        // Not read from the config block — see `DEFAULT_PLUGIN_DIR`. A
        // `plugin_dirs:` key there (or at the YAML top level) is ignored;
        // warning about one is the factory's job, since it can name the plugin.
        let plugin_dirs = default_plugin_dirs();

        // A layout that will not resolve leaves the venv absent. That is a
        // config error in production, but the failure belongs in
        // `initialize()` rather than here: the factory should still hand back a
        // named plugin the engine can report against.
        let venv = VenvManager::new(
            &plugin_dirs,
            &venv_config.class_name,
            venv_config.requirements_file.as_deref(),
            config.version.as_deref(),
        )
        .ok();

        Ok(Self {
            config: config.clone(),
            venv_config,
            plugin_dirs,
            venv,
            worker: tokio::sync::RwLock::new(None),
            worker_override: None,
            worker_cwd: None,
            respawn: tokio::sync::Mutex::new(RespawnState::new()),
            shut_down: std::sync::atomic::AtomicBool::new(false),
        })
    }

    fn is_shut_down(&self) -> bool {
        self.shut_down.load(std::sync::atomic::Ordering::SeqCst)
    }

    fn shut_down_error() -> HostError {
        HostError::WorkerStart {
            message: "the plugin has been shut down, so its worker is not respawned".into(),
        }
    }

    /// The resolved plugin directories — `<project root>/plugins`.
    pub fn plugin_dirs(&self) -> &[String] {
        &self.plugin_dirs
    }

    /// Run the worker with an explicit working directory.
    ///
    /// Production leaves this unset and inherits the host's cwd. Tests set it
    /// so their scratch plugin dir falls inside the worker's
    /// `ALLOWED_PLUGIN_DIRS` (which accepts `os.getcwd()`), and so a plugin's
    /// marker files land somewhere the test can find and clean up.
    #[cfg(any(test, feature = "testing"))]
    pub fn with_worker_cwd(mut self, cwd: PathBuf) -> Self {
        self.worker_cwd = Some(cwd);
        self
    }

    /// Point the plugin at explicit plugin directories, replacing the
    /// `<project root>/plugins` default.
    ///
    /// Test-only seam. Production has no override by design — the directory is
    /// fixed so the host and `cpex plugin install` cannot disagree. The e2e
    /// tests scaffold a plugin into a temp dir and need the venv built there
    /// rather than in the developer's real `plugins/`, which a test must not
    /// touch.
    ///
    /// Rebuilds the venv manager, since the layout is derived from these dirs.
    #[cfg(any(test, feature = "testing"))]
    pub fn with_plugin_dirs(mut self, dirs: Vec<String>) -> Self {
        self.venv = VenvManager::new(
            &dirs,
            &self.venv_config.class_name,
            self.venv_config.requirements_file.as_deref(),
            self.config.version.as_deref(),
        )
        .ok();
        self.plugin_dirs = dirs;
        self
    }

    /// Point the plugin at an explicit worker script, skipping venv build and
    /// script resolution.
    ///
    /// Test-only seam: the adapter tests need to exercise dispatch against a
    /// stub worker without paying for a real venv, and the credential tests
    /// need a worker that reports exactly what it received.
    #[cfg(any(test, feature = "testing"))]
    pub fn with_worker_override(mut self, script: PathBuf) -> Self {
        self.worker_override = Some(script);
        self
    }

    /// The parsed venv settings.
    pub fn venv_config(&self) -> &VenvConfig {
        &self.venv_config
    }

    /// The live worker's retained stderr, for leak assertions in tests.
    ///
    /// Returns an empty vector when no worker is running.
    #[cfg(any(test, feature = "testing"))]
    pub async fn worker_stderr(&self) -> Vec<String> {
        match self.worker.read().await.as_ref() {
            Some(worker) => worker.stderr_lines().await,
            None => Vec::new(),
        }
    }

    /// Launch a worker process for this plugin.
    ///
    /// The single path both `initialize()` and the respawn take, which is what
    /// makes a respawned worker indistinguishable from the original: same
    /// interpreter, same script, same cwd, same content cap, same timeout. The
    /// security posture is preserved by construction rather than by a second
    /// copy of the setup that could drift — the capability and credential
    /// handshake is applied per invocation in `PythonHookAdapter::invoke`
    /// against `self.config.capabilities`, so a fresh worker is gated by exactly
    /// the same declared capabilities as the one it replaced. There is no
    /// "degraded" or "retry" mode that relaxes it.
    async fn launch_worker(&self) -> Result<WorkerClient, HostError> {
        // With an override the worker script is given outright, so neither a
        // venv nor script resolution is needed — but the interpreter still is.
        let (python, script) = match self.worker_override.as_ref() {
            Some(script) => (which_python3()?, script.clone()),
            None => {
                let venv = self.venv.as_ref().ok_or_else(|| HostError::Config {
                    message: format!(
                        "could not resolve a venv layout under {} — check that `class_name` \
                         ('{}') begins with a package segment",
                        self.plugin_dirs.join(", "),
                        self.venv_config.class_name,
                    ),
                })?;

                venv.ensure().await?;
                let script = crate::worker::resolve_worker_script(
                    &venv.layout().venv_path,
                    &self.venv_config.script_path,
                )?;

                (venv.python_executable(), script)
            },
        };

        // Matching `venv_comm.py`'s `cwd=os.getcwd()`. The worker's
        // ALLOWED_PLUGIN_DIRS allowlist accepts its own cwd, so inheriting the
        // host's is what lets a plugin dir declared relative to the gateway be
        // importable at all.
        WorkerClient::spawn(
            &python,
            &script,
            self.worker_cwd.as_deref(),
            self.venv_config.max_content_size,
            self.venv_config.timeout_secs,
        )
        .await
    }

    /// The live worker, respawning it when the previous one has died.
    ///
    /// A worker can die between invocations — an OOM kill, a segfault in a
    /// native dependency, a plugin that calls `sys.exit`. Before this, the
    /// plugin stayed permanently broken: every later invocation found the dead
    /// `WorkerClient` still in the slot and failed against it, so one crash
    /// disabled the plugin for the lifetime of the process.
    ///
    /// Respawn is bounded ([`MAX_RESPAWN_ATTEMPTS`]) and backed off, and a
    /// plugin that exhausts its budget returns a typed `WorkerStart` error
    /// rather than retrying forever.
    async fn worker(&self) -> Result<Arc<WorkerClient>, Box<PluginError>> {
        let to_plugin_error = |e: HostError| e.into_plugin_error(&self.config.name);

        // The common path: a worker exists and is alive. No lock beyond the
        // read, so a healthy invocation pays nothing for the respawn machinery.
        if let Some(worker) = self.worker.read().await.clone() {
            if worker.is_alive() {
                return Ok(worker);
            }
        } else if self.is_shut_down() {
            return Err(to_plugin_error(Self::shut_down_error()));
        } else {
            // Never initialized is not a respawn case — respawning here would
            // paper over a lifecycle bug (or an initialize() that failed and
            // was rolled back) by silently starting a worker the engine does
            // not know about.
            return Err(to_plugin_error(HostError::WorkerStart {
                message: "plugin has no worker — initialize() did not run or it failed".into(),
            }));
        }

        self.respawn_worker().await
    }

    /// Replace a dead worker, under the respawn lock.
    ///
    /// Split out so the happy path in `worker()` stays lock-free. Holding the
    /// respawn mutex across the relaunch serializes concurrent invocations that
    /// all noticed the same death: the first respawns, the rest re-check and
    /// find a live worker without spending budget.
    async fn respawn_worker(&self) -> Result<Arc<WorkerClient>, Box<PluginError>> {
        let to_plugin_error = |e: HostError| e.into_plugin_error(&self.config.name);
        let mut respawn = self.respawn.lock().await;

        // Re-check under the lock. A concurrent invocation may already have
        // done the work while this one waited, or the plugin may have shut
        // down. An empty slot means the latter: a failed respawn leaves the
        // dead worker in place, so only `shutdown()` empties it.
        let dead = match self.worker.read().await.clone() {
            Some(worker) if worker.is_alive() => return Ok(worker),
            Some(_) if self.is_shut_down() => {
                return Err(to_plugin_error(Self::shut_down_error()));
            },
            Some(worker) => worker,
            None => return Err(to_plugin_error(Self::shut_down_error())),
        };

        let Some(attempt) = respawn.try_reserve() else {
            tracing::error!(
                plugin = %self.config.name,
                max_attempts = MAX_RESPAWN_ATTEMPTS,
                "worker died and the respawn budget is exhausted; giving up"
            );
            return Err(to_plugin_error(HostError::WorkerStart {
                message: format!(
                    "worker died and could not be respawned after {MAX_RESPAWN_ATTEMPTS} attempts \
                     — the plugin's worker is failing to start (check its venv and dependencies)"
                ),
            }));
        };

        // Exponential: 100ms, 200ms, 400ms. Applied *before* the spawn so a
        // worker that dies instantly on launch cannot spin a tight loop.
        let backoff = std::time::Duration::from_millis(
            RESPAWN_BACKOFF_BASE_MS.saturating_mul(1 << (attempt - 1)),
        );
        tracing::warn!(
            plugin = %self.config.name,
            attempt,
            max_attempts = MAX_RESPAWN_ATTEMPTS,
            backoff_ms = backoff.as_millis(),
            "worker is dead; respawning"
        );
        tokio::time::sleep(backoff).await;

        // The plugin may have shut down during the backoff.
        if self.is_shut_down() {
            return Err(to_plugin_error(Self::shut_down_error()));
        }

        // Reap the corpse before replacing it, so the old process and its I/O
        // tasks are not left behind on every respawn. It stays in the slot
        // until a replacement is installed: if the launch below fails, the
        // next invocation finds a dead worker and comes back here, within the
        // same budget, instead of reading an empty slot as "never initialized"
        // and giving up on the plugin for good.
        if let Err(e) = dead.shutdown().await {
            tracing::debug!(plugin = %self.config.name, "reaping the dead worker: {e}");
        }

        // Same launch path as `initialize()` — see `launch_worker`.
        let worker = Arc::new(self.launch_worker().await.map_err(to_plugin_error)?);

        // Re-verified rather than assumed from initialize(): the replacement is a
        // fresh process resolving `worker.py` from the venv again, and the venv
        // can have been rebuilt or downgraded since. Skipping this would make
        // respawn the one path that installs an unverified worker.
        if let Err(e) = self.verify_worker_understands(&worker).await {
            stop_unused_worker(&self.config.name, &worker).await;
            return Err(to_plugin_error(e));
        }

        // Checked under the slot's write lock, which `shutdown()` takes after
        // setting the flag: either this sees the flag, or `shutdown()` waits
        // here and then stops the worker installed below.
        {
            let mut slot = self.worker.write().await;
            if self.is_shut_down() {
                drop(slot);
                stop_unused_worker(&self.config.name, &worker).await;
                return Err(to_plugin_error(Self::shut_down_error()));
            }
            *slot = Some(Arc::clone(&worker));
        }

        // Deliberately *not* `record_success()` here.
        //
        // `is_alive()` reflects the reader task's last observation, and a worker
        // that dies at import time has not necessarily been observed yet — it
        // reports alive for the moment right after spawn. Resetting the budget
        // on that would let a crash-looping worker refill its allowance on every
        // attempt and retry forever, which is exactly what the bound exists to
        // prevent. The budget is instead reset by the passage of time
        // (`RESPAWN_BUDGET_RESET_SECS`), so recovery is still possible once the
        // underlying cause is fixed, and by a *successful invocation* through
        // `note_worker_healthy`, which is proof the worker answered rather than
        // merely proof it launched.
        tracing::info!(
            plugin = %self.config.name,
            attempt,
            "worker respawned; awaiting proof it can serve a request"
        );

        Ok(worker)
    }

    /// Mark the worker healthy after it has actually answered a request.
    ///
    /// Called from the adapter on a successful round trip. A completed
    /// invocation is the only evidence that distinguishes a worker that started
    /// from a worker that works, so it — not a successful spawn — is what
    /// returns the respawn budget to full.
    async fn note_worker_healthy(&self) {
        let mut respawn = self.respawn.lock().await;
        if respawn.attempts > 0 {
            respawn.record_success();
        }
    }

    /// The declared capabilities, reduced to the ones the worker's
    /// `PluginConfig` model will accept.
    ///
    /// The Python model's `capabilities` validator *raises* on a name its
    /// `Capability` enum does not know, and Pydantic validation happens while
    /// the worker is constructing the config — before the plugin is ever
    /// called. This host's capability vocabulary is the larger of the two
    /// (`read_client`, `read_workload`, `read_all`, and the two
    /// `raw_credentials` caps have no Python counterpart yet), so forwarding
    /// the set verbatim would take down every hook of any plugin declaring one
    /// of them.
    ///
    /// Dropping the unknown names is safe in the direction that matters:
    /// the extensions the worker receives were already filtered by the
    /// executor, so a dropped capability can only make the worker's
    /// re-filter *more* restrictive, never less. A slot this host chose not to
    /// send cannot be recovered by a capability string. The reverse — keeping an
    /// unknown name — trades a possibly-narrower view for a dead worker.
    fn worker_capabilities(&self) -> Vec<String> {
        let (known, unknown): (Vec<String>, Vec<String>) = self
            .config
            .capabilities
            .iter()
            .cloned()
            .partition(|c| WORKER_KNOWN_CAPABILITIES.contains(&c.as_str()));

        if !unknown.is_empty() {
            // Worth a line: it means this host's vocabulary has grown past the
            // installed framework's, and a plugin declaring one of these sees a
            // narrower view out-of-process than it would in-process.
            tracing::warn!(
                plugin = %self.config.name,
                dropped = ?unknown,
                "the installed cpex framework does not model these capabilities; \
                 omitting them from the worker's config so it can still validate"
            );
        }

        known
    }

    /// Fail closed unless the worker implements every wire feature this
    /// plugin's declaration commits the host to using.
    ///
    /// # Why a version check would not do
    ///
    /// The host cannot tell an old `worker.py` from a current one by version:
    /// two framework builds shipped as 0.1.2 with different worker protocols. An
    /// old worker also does not *reject* a field it does not know — it ignores
    /// it. So a plugin declaring `read_inbound_credentials` against such a
    /// worker ran with no credential at all, silently, which is the exact case
    /// fail-closed rule 2 exists to prevent. The worker is therefore asked
    /// directly, and a worker that cannot answer is assumed to support nothing.
    ///
    /// Only features this plugin actually needs are required. A plugin that
    /// declares no gated capability and runs no credential hook is served fine
    /// by an old worker, and refusing to start it would be a regression for
    /// every existing plugin.
    async fn verify_worker_understands(&self, worker: &WorkerClient) -> Result<(), HostError> {
        let required = self.required_worker_features();

        // Tool, prompt and resource `cmf.*` hooks also need the report to
        // choose native dispatch or the legacy shim. When they require no wire
        // feature, an inconclusive probe leaves the report unset and the
        // adapter falls back to the shim. LLM hooks require `cmf_hooks` above.
        let declares_cmf_hook = self.config.hooks.iter().any(|h| cmf_shim::is_cmf_hook(h));
        if required.is_empty() {
            if declares_cmf_hook {
                match worker.capabilities().await {
                    Ok(reported) => self.log_cmf_dispatch(&reported),
                    Err(e) => tracing::warn!(
                        plugin = %self.config.name,
                        "worker capabilities probe failed ({e}); serving cmf hooks through the \
                         legacy payload shim"
                    ),
                }
            }
            return Ok(());
        }

        let reported = worker.capabilities().await.map_err(|e| {
            if required.contains(&crate::worker::FEATURE_CMF_HOOKS) {
                HostError::Config {
                    message: format!(
                        "plugin '{}' declares an LLM hook, but the worker's `cmf_hooks` support \
                         could not be verified ({e}); upgrade the cpex package in the plugin's venv",
                        self.config.name
                    ),
                }
            } else {
                e
            }
        })?;
        if declares_cmf_hook {
            self.log_cmf_dispatch(&reported);
        }
        let missing: Vec<&str> = required
            .iter()
            .copied()
            .filter(|f| !reported.has(f))
            .collect();

        if missing.is_empty() {
            tracing::debug!(
                plugin = %self.config.name,
                protocol_version = reported.protocol_version,
                cpex_version = ?reported.cpex_version,
                "worker handshake satisfied"
            );
            return Ok(());
        }

        // Sorted so the diagnostic is stable across runs: `capabilities` is a
        // HashSet, and an operator comparing two startup failures should not
        // have to notice that only the ordering changed.
        let mut declared: Vec<&str> = self
            .config
            .capabilities
            .iter()
            .map(String::as_str)
            .collect();
        declared.sort_unstable();

        Err(HostError::Credential {
            message: format!(
                "the installed cpex worker does not implement {} — which plugin '{}' needs given \
                 its declared capabilities ({}) and hooks ({}). The worker reported protocol \
                 version {} and features [{}] (cpex {}). Running anyway would drop those fields \
                 silently, so this plugin is failed closed at startup; upgrade the cpex package \
                 in the plugin's venv.",
                missing.join(", "),
                self.config.name,
                declared.join(", "),
                self.config.hooks.join(", "),
                reported.protocol_version,
                reported
                    .features
                    .iter()
                    .map(String::as_str)
                    .collect::<Vec<_>>()
                    .join(", "),
                reported
                    .cpex_version
                    .as_deref()
                    .unwrap_or("version unknown"),
            ),
        })
    }

    /// Say once, at startup, how this plugin's `cmf.*` hooks will be served.
    fn log_cmf_dispatch(&self, reported: &crate::worker::WorkerCapabilities) {
        if reported.has(crate::worker::FEATURE_CMF_HOOKS) {
            tracing::debug!(plugin = %self.config.name, "worker serves cmf hooks natively");
        } else {
            tracing::info!(
                plugin = %self.config.name,
                "worker does not claim `{}`; serving cmf tool, prompt and resource hooks \
                 through the legacy payload shim",
                crate::worker::FEATURE_CMF_HOOKS
            );
        }
    }

    /// The wire features this plugin's declaration commits the host to using.
    ///
    /// Derived from the declaration rather than from what a given request
    /// happens to carry, so the check is decidable at startup and does not vary
    /// per invocation.
    fn required_worker_features(&self) -> Vec<&'static str> {
        let mut required = Vec::new();

        // LLM messages have no legacy worker payload. A worker without native
        // CMF dispatch cannot serve these hooks even if it needs no capability.
        if self
            .config
            .hooks
            .iter()
            .any(|h| matches!(h.as_str(), "cmf.llm_input" | "cmf.llm_output"))
        {
            required.push(crate::worker::FEATURE_CMF_HOOKS);
        }

        // A credential hook only receives material when the matching capability
        // is declared, so both halves are required for the field to travel —
        // which makes the capability the precise trigger.
        let declares_credential_cap = self.config.capabilities.iter().any(|c| {
            c == crate::credentials::CAP_READ_INBOUND || c == crate::credentials::CAP_READ_DELEGATED
        });
        if declares_credential_cap {
            required.push(crate::worker::FEATURE_CREDENTIAL);
        }

        // Any capability the worker models is a capability whose slot this host
        // sends in the `extensions` field. A worker that drops the field leaves
        // the hook with `extensions=None`, so the plugin silently sees nothing
        // of what it declared.
        let declares_extension_cap = self
            .config
            .capabilities
            .iter()
            .any(|c| WORKER_KNOWN_CAPABILITIES.contains(&c.as_str()));
        if declares_extension_cap {
            required.push(crate::worker::FEATURE_EXTENSIONS);
        }

        required
    }

    /// Build the `load_and_run_hook` task for one invocation.
    ///
    /// The field names and encodings are the protocol contract with
    /// `worker.py`: `config` is a JSON *string* the worker `json.loads`es into
    /// `PluginConfig(**config_raw)`, and `context` is the worker's own
    /// `{state, global_context, metadata}` shape rather than the Rust
    /// `PluginContext` layout.
    ///
    /// `legacy_cmf` selects the legacy payload shim's hook names for the
    /// `cmf.*` hooks: true for a worker that does not claim `cmf_hooks`.
    fn build_task(
        &self,
        hook_name: &str,
        legacy_cmf: bool,
        payload: serde_json::Value,
        ctx: &PluginContext,
    ) -> Result<serde_json::Value, HostError> {
        // The worker constructs a Python `PluginConfig` from this, so it must
        // carry that model's field names — not this host's venv settings.
        //
        // `capabilities` has to travel even though this host already applies the
        // executor's filtered view: the Python manager re-filters against
        // `plugin_ref.capabilities` before it calls a 3-arg hook
        // (`cpex/framework/manager.py`, `_execute_with_timeout`). Omitting the
        // field left that set empty, so the second filter stripped every gated
        // slot the first one had just allowed — a 3-arg plugin saw
        // `security.labels` empty and `http` as `None` no matter what it
        // declared. The two filters are now given the same input, so the second
        // one is idempotent rather than destructive.
        // Hook names travel in the spelling the worker registers, which differs
        // from PPE's for the identity, delegation and CMF prompt hooks, and for
        // every shimmed CMF hook.
        let worker_hooks: Vec<&str> = self
            .config
            .hooks
            .iter()
            .map(|hook| worker_hook_for(hook, legacy_cmf))
            .collect();
        let config_json = serde_json::json!({
            "name": self.config.name,
            "kind": self.config.kind,
            "hooks": worker_hooks,
            "capabilities": self.worker_capabilities(),
            "config": self.config.config.clone().unwrap_or(serde_json::Value::Null),
        });
        let config_string =
            serde_json::to_string(&config_json).map_err(|e| HostError::Protocol {
                message: format!("could not serialize the plugin config for the worker: {e}"),
            })?;

        // Prefer a request id the pipeline already put in global state, so a
        // plugin's logs correlate with the gateway's. `PluginContext` has no
        // dedicated field for one, so fall back to a fresh UUID rather than
        // sending a placeholder that would collide across requests.
        let request_id = ctx
            .get_global(GLOBAL_REQUEST_ID_KEY)
            .and_then(serde_json::Value::as_str)
            .map(str::to_string)
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());

        Ok(serde_json::json!({
            "task_type": crate::worker::TASK_LOAD_AND_RUN_HOOK,
            "plugin_dirs": self.plugin_dirs,
            "class_name": self.venv_config.class_name,
            "config": config_string,
            "hook_type": worker_hook_for(hook_name, legacy_cmf),
            "plugin_name": self.config.name,
            "payload": payload,
            "context": {
                // `local_state` is this plugin's private slice; `global_state`
                // is the pipeline-wide map the worker exposes as
                // `global_context`.
                "state": ctx.local_state,
                "global_context": {
                    // `request_id` is the one *required* field on the Python
                    // `GlobalContext`; omitting it fails Pydantic validation
                    // inside the worker before the plugin is ever called. The
                    // Rust `PluginContext` carries no request id, so one is
                    // synthesized per invocation — a plugin that correlates
                    // logs by request id still gets a stable, unique value for
                    // the call.
                    "request_id": request_id,
                    "state": ctx.global_state,
                },
                "metadata": {},
            },
        }))
    }
}

#[async_trait]
impl Plugin for IsolatedPythonPlugin {
    fn config(&self) -> &PluginConfig {
        &self.config
    }

    /// Build the venv (or reuse a cached one) and launch the worker.
    ///
    /// Heavy setup lives here rather than on the invoke path: the engine
    /// awaits this once per plugin, with rollback on failure.
    async fn initialize(&self) -> Result<(), Box<PluginError>> {
        self.shut_down
            .store(false, std::sync::atomic::Ordering::SeqCst);
        let worker = self
            .launch_worker()
            .await
            .map_err(|e| e.into_plugin_error(&self.config.name))?;

        // Verify the worker understands what this plugin's declaration commits
        // us to sending, before it is installed as the live worker. A mismatch
        // is a startup failure, not a per-request one: the plugin cannot be
        // served safely at all, and leaving it registered would mean discovering
        // that on a live request instead.
        self.verify_worker_understands(&worker)
            .await
            .map_err(|e| e.into_plugin_error(&self.config.name))?;

        *self.worker.write().await = Some(Arc::new(worker));
        Ok(())
    }

    /// Stop the worker, killing it if it will not exit on its own.
    async fn shutdown(&self) -> Result<(), Box<PluginError>> {
        // Before the slot is emptied, so a respawn in flight sees it. See
        // `respawn_worker`.
        self.shut_down
            .store(true, std::sync::atomic::Ordering::SeqCst);
        let worker = self.worker.write().await.take();
        let result = match worker {
            Some(worker) => worker
                .shutdown()
                .await
                .map_err(|e| e.into_plugin_error(&self.config.name)),
            None => Ok(()),
        };

        // Give up this plugin's claim on the (possibly shared) venv, so a later
        // rebuild is not permanently downgraded to an in-place update by a stale
        // owner entry. Best-effort: a registry write failure must not turn a
        // clean shutdown into an error, since the venv itself is fine.
        if let Some(venv) = self.venv.as_ref()
            && let Err(e) = venv.release_owner().await
        {
            tracing::warn!(
                plugin = %self.config.name,
                "could not release the venv owner claim: {e}"
            );
        }

        result
    }
}

/// Stop a worker that was launched but will not be installed.
async fn stop_unused_worker(plugin: &str, worker: &WorkerClient) {
    if let Err(e) = worker.shutdown().await {
        tracing::debug!(plugin = %plugin, "stopping an unused worker: {e}");
    }
}

/// The hook name the worker knows `hook_name` by, given how `cmf.*` hooks are
/// being served.
fn worker_hook_for(hook_name: &str, legacy_cmf: bool) -> &str {
    if legacy_cmf && let Some(legacy) = cmf_shim::legacy_hook_name(hook_name) {
        return legacy;
    }
    conversion::worker_hook_name(hook_name)
}

/// Whether `worker` must be served `cmf.*` hooks through the legacy payload
/// shim: true unless its handshake claimed `cmf_hooks`. A worker that was never
/// asked, or could not answer, gets the shim.
fn needs_legacy_cmf(worker: &WorkerClient) -> bool {
    !worker
        .reported_capabilities()
        .is_some_and(|c| c.has(crate::worker::FEATURE_CMF_HOOKS))
}

/// Resolve an absolute `python3` from PATH.
///
/// `WorkerClient::spawn` takes an interpreter path rather than doing a PATH
/// lookup, so the override path needs one. Production goes through the venv's
/// own interpreter and never calls this.
fn which_python3() -> Result<PathBuf, HostError> {
    let candidates = [
        "/usr/bin/python3",
        "/usr/local/bin/python3",
        "/opt/homebrew/bin/python3",
    ];
    if let Some(found) = candidates.iter().map(Path::new).find(|p| p.exists()) {
        return Ok(found.to_path_buf());
    }

    // Fall back to asking the shell, which honors PATH (pyenv, conda, nix).
    let output = std::process::Command::new("sh")
        .args(["-c", "command -v python3"])
        .output()
        .map_err(|e| HostError::WorkerStart {
            message: format!("could not look up python3: {e}"),
        })?;

    let path = String::from_utf8_lossy(&output.stdout).trim().to_string();
    if path.is_empty() {
        return Err(HostError::WorkerStart {
            message: "python3 was not found on PATH".into(),
        });
    }
    Ok(PathBuf::from(path))
}

/// Handler for one hook of one Python plugin.
///
/// Holds the hook name because a single plugin object serves every hook it
/// declared, and the worker needs to be told which one to run.
pub struct PythonHookAdapter {
    plugin: Arc<IsolatedPythonPlugin>,
    hook_name: &'static str,
}

impl PythonHookAdapter {
    pub fn new(plugin: Arc<IsolatedPythonPlugin>, hook_name: &'static str) -> Self {
        Self { plugin, hook_name }
    }
}

#[async_trait]
impl AnyHookHandler for PythonHookAdapter {
    /// Serialize, send, and convert one hook invocation.
    ///
    /// Every failure returns a `PluginError` and stops there — the *executor*
    /// applies the plugin's configured `on_error` policy (fail / ignore /
    /// disable). This host deliberately does not interpret that policy itself;
    /// doing so would double-apply it.
    async fn invoke(
        &self,
        payload: &dyn PluginPayload,
        extensions: &Extensions,
        ctx: &mut PluginContext,
    ) -> Result<Box<dyn std::any::Any + Send + Sync>, Box<PluginError>> {
        let plugin_name = &self.plugin.config.name;
        let to_plugin_error = |e: HostError| e.into_plugin_error(plugin_name);

        // The worker first: whether it serves `cmf.*` hooks natively decides
        // what the task carries, and a respawn inside `worker()` may replace it
        // with one that answers differently.
        let worker = self.plugin.worker().await?;

        // A probe that failed at startup leaves no report, which would pin
        // this worker to the shim for its whole life. Ask once more, the first
        // time a `cmf.*` hook needs the answer.
        if cmf_shim::is_cmf_hook(self.hook_name)
            && worker.reported_capabilities().is_none()
            && worker.claim_probe_retry()
        {
            match worker.capabilities().await {
                Ok(reported) => self.plugin.log_cmf_dispatch(&reported),
                Err(e) => tracing::warn!(
                    plugin = %plugin_name,
                    "worker capabilities probe failed again ({e}); serving cmf hooks through \
                     the legacy payload shim for this worker"
                ),
            }
        }
        let legacy_cmf = needs_legacy_cmf(&worker);

        // A worker that predates `cmf.*` hooks gets the legacy typed payload
        // under the legacy hook name. A message the legacy payload cannot
        // represent, and the LLM hooks, fail closed here.
        let shim = if legacy_cmf && cmf_shim::is_cmf_hook(self.hook_name) {
            Some(cmf_shim::to_legacy(self.hook_name, payload).map_err(to_plugin_error)?)
        } else {
            None
        };

        let payload_json = match shim.as_ref() {
            Some(shim) => shim.payload().clone(),
            None => conversion::serialize_payload(payload).map_err(to_plugin_error)?,
        };
        let mut task = self
            .plugin
            .build_task(self.hook_name, legacy_cmf, payload_json, ctx)
            .map_err(to_plugin_error)?;

        // Raw credentials, for the two hooks that carry them and only for a
        // plugin that declared the matching capability. Fails closed.
        credentials::attach_credential_for_payload(
            &mut task,
            self.hook_name,
            payload,
            extensions,
            &self.plugin.config.capabilities,
        )
        .map_err(to_plugin_error)?;

        // The capability-filtered extensions, so a 3-arg
        // `(payload, context, extensions)` hook sees out-of-process what it
        // would see in-process. Attaches nothing when no slot is visible.
        crate::extensions::attach_extensions(&mut task, extensions).map_err(to_plugin_error)?;

        let mut response = worker.send_task(task).await.map_err(to_plugin_error)?;

        // A completed round trip is the only proof a (possibly respawned) worker
        // actually serves requests, so it is what returns the respawn budget.
        self.plugin.note_worker_healthy().await;

        // State the worker sends back is merged into the caller's context
        // before the result is returned, so a plugin's context writes survive.
        apply_context_deltas(&response, ctx);

        // Under the shim the modified payload is a legacy one, so it is taken
        // out before the generic conversion (which would parse it as a CMF
        // message) and mapped back onto the original message afterwards.
        // Allow, deny and the violation are converted unchanged.
        //
        // PPE's identity and delegation payloads take the same route: the
        // plugin answers with the worker's result model, which is mapped onto
        // a clone of the dispatched payload rather than parsed into a type
        // PPE's own downcasts would not find.
        let translated = shim.is_some() || credential_payloads::is_ppe_credential_payload(payload);
        let returned_modified = if translated {
            response
                .as_object_mut()
                .and_then(|r| r.remove("modified_payload"))
        } else {
            None
        };

        // `extensions` is passed back in so the returned-extensions path can
        // reuse the inbound `Arc`s for immutable slots and read the write
        // tokens the executor issued. The executor's copy-on-write merge then
        // validates the result against the mutability tiers.
        let mut fields = conversion::response_to_result(self.hook_name, response, extensions)
            .map_err(to_plugin_error)?;

        if let Some(modified) = returned_modified.filter(|m| !m.is_null()) {
            apply_translated_modification(shim.as_ref(), payload, modified, &mut fields)
                .map_err(to_plugin_error)?;
        }
        Ok(Box::new(fields))
    }

    /// The hook family the declared name belongs to, so the registry's family
    /// check accepts this handler. The adapter forwards whatever payload the
    /// hook carries, so it serves the family of any name it is declared on. A
    /// name with no family in the hook metadata reports itself, which the
    /// registry accepts.
    fn hook_type_name(&self) -> &'static str {
        ppe::praxis_policy_core::hooks::lookup_hook_metadata(self.hook_name)
            .and_then(|meta| meta.family)
            .unwrap_or(self.hook_name)
    }
}

/// Map a modified payload the worker returned in its own model back onto the
/// payload PPE dispatched, for the shimmed `cmf.*` hooks and for PPE's
/// identity and delegation payloads.
fn apply_translated_modification(
    shim: Option<&cmf_shim::ShimmedPayload>,
    payload: &dyn PluginPayload,
    modified: serde_json::Value,
    fields: &mut ppe::praxis_policy_core::executor::ErasedResultFields,
) -> Result<(), HostError> {
    use ppe::praxis_policy_core::delegation::DelegationPayload;
    use ppe::praxis_policy_core::identity::IdentityPayload;

    // A deny modifies nothing, so a result the plugin attached to one is not
    // worth failing the deny over.
    if !fields.continue_processing {
        fields.modified_payload = None;
        return Ok(());
    }
    if let Some(shim) = shim {
        fields.modified_payload = Some(Box::new(shim.apply_modified(modified)?));
        return Ok(());
    }
    let any = payload.as_any();
    if let Some(identity) = any.downcast_ref::<IdentityPayload>() {
        match credential_payloads::identity_from_worker(identity, modified)? {
            credential_payloads::IdentityOutcome::Resolved(resolved) => {
                fields.modified_payload = Some(resolved);
            },
            credential_payloads::IdentityOutcome::Rejected(violation) => {
                // A rejection is a deny, and a deny modifies nothing.
                fields.modified_payload = None;
                fields.continue_processing = false;
                fields.violation = Some(violation);
            },
        }
    } else if let Some(delegation) = any.downcast_ref::<DelegationPayload>() {
        fields.modified_payload = Some(Box::new(credential_payloads::delegation_from_worker(
            delegation, modified,
        )?));
    }
    Ok(())
}

/// Merge worker-returned context state back into the mutable plugin context.
///
/// The worker returns the context under the same `{state, global_context}`
/// shape it received. Merging rather than replacing is deliberate: a plugin
/// that touched one key must not blank out the rest of the pipeline's state.
fn apply_context_deltas(response: &serde_json::Value, ctx: &mut PluginContext) {
    let Some(context) = response.get("context") else {
        return;
    };

    if let Some(state) = context.get("state").and_then(|v| v.as_object()) {
        for (key, value) in state {
            ctx.local_state.insert(key.clone(), value.clone());
        }
    }

    if let Some(global) = context
        .get("global_context")
        .and_then(|g| g.get("state"))
        .and_then(|v| v.as_object())
    {
        for (key, value) in global {
            ctx.global_state.insert(key.clone(), value.clone());
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::testing::TempDir;

    use super::*;

    fn config_with(value: serde_json::Value) -> PluginConfig {
        PluginConfig {
            name: "py-plugin".into(),
            kind: crate::factory::KIND.into(),
            hooks: vec!["tool_pre_invoke".into()],
            config: Some(value),
            ..Default::default()
        }
    }

    #[test]
    fn full_config_block_deserializes_every_field() {
        let cfg = VenvConfig::from_plugin_config(&config_with(serde_json::json!({
            "class_name": "my_pkg.filters.PiiFilter",
            "requirements_file": "requirements.txt",
            "plugin_dirs": ["./plugins", "./vendor"],
            "script_path": "custom/worker.py",
            "max_content_size": 2048,
            "timeout_secs": 5,
        })))
        .expect("a fully populated block parses");

        assert_eq!(cfg.class_name, "my_pkg.filters.PiiFilter");
        assert_eq!(cfg.requirements_file.as_deref(), Some("requirements.txt"));
        assert_eq!(cfg.script_path, "custom/worker.py");
        assert_eq!(cfg.max_content_size, 2048);
        assert_eq!(cfg.timeout_secs, 5);
        // The block above also carries `plugin_dirs`, which is no longer a
        // field — an unknown key must parse harmlessly rather than error, so
        // an existing config keeps loading.
    }

    #[test]
    fn absent_optionals_fall_back_to_documented_defaults() {
        let cfg = VenvConfig::from_plugin_config(&config_with(serde_json::json!({
            "class_name": "my_pkg.Minimal",
        })))
        .expect("class_name alone is a valid block");

        // These three defaults are the contract with `client.py` /
        // `venv_comm.py` — a drift here changes behavior silently.
        assert_eq!(
            cfg.max_content_size, 10_000_000,
            "matches client.py's max_content_size"
        );
        assert_eq!(
            cfg.timeout_secs, 30,
            "matches venv_comm.py's send_task timeout"
        );
        assert_eq!(cfg.script_path, "cpex/framework/isolated/worker.py");

        assert!(
            cfg.requirements_file.is_none(),
            "no requirements file is valid"
        );
    }

    #[test]
    fn missing_class_name_errors() {
        let err = VenvConfig::from_plugin_config(&config_with(serde_json::json!({
            "plugin_dirs": ["./plugins"],
        })))
        .expect_err("class_name is required");

        assert!(matches!(*err, PluginError::Config { .. }));
        assert!(err.to_string().contains("class_name"));
    }

    #[test]
    fn absent_config_block_errors_rather_than_panicking() {
        let cfg = PluginConfig {
            name: "no-config".into(),
            kind: crate::factory::KIND.into(),
            hooks: vec!["tool_pre_invoke".into()],
            config: None,
            ..Default::default()
        };

        let err = VenvConfig::from_plugin_config(&cfg)
            .expect_err("an absent block cannot supply class_name");
        assert!(matches!(*err, PluginError::Config { .. }));
    }

    /// The plugin dir a config with no usable `plugin_dirs` anywhere resolves
    /// to: `<cwd>/plugins`.
    fn expected_default_dir() -> String {
        std::env::current_dir()
            .unwrap()
            .join("plugins")
            .display()
            .to_string()
    }

    #[test]
    fn plugin_dirs_default_to_plugins_at_the_project_root() {
        // Neither key is set anywhere, which is the shape the host now expects.
        let plugin = IsolatedPythonPlugin::from_config(&config_with(serde_json::json!({
            "class_name": "my_pkg.Minimal",
        })))
        .expect("class_name alone is a valid block");

        assert_eq!(
            plugin.plugin_dirs(),
            [expected_default_dir()],
            "the host supplies the plugin dir; the config no longer can"
        );
        // Absolute, so a later cwd change cannot move the venv out from under
        // a running plugin.
        assert!(
            Path::new(&plugin.plugin_dirs()[0]).is_absolute(),
            "the resolved dir must be absolute: {:?}",
            plugin.plugin_dirs()
        );
    }

    #[test]
    fn a_plugin_dirs_key_in_the_config_block_is_ignored() {
        // The key an existing config still carries. It must not steer the host,
        // and it must not fail the parse either — an operator upgrading should
        // see the warning (emitted by the factory) rather than a load error.
        let plugin = IsolatedPythonPlugin::from_config(&config_with(serde_json::json!({
            "class_name": "my_pkg.Minimal",
            "plugin_dirs": ["/somewhere/else", "/and/here"],
        })))
        .expect("an ignored key must not fail the parse");

        assert_eq!(
            plugin.plugin_dirs(),
            [expected_default_dir()],
            "`plugin_dirs` in the config block must be ignored"
        );
    }

    #[test]
    fn ppe_hook_names_reach_the_worker_in_its_own_spelling() {
        // The Python framework keys payload models, hook methods and credential
        // handling on the older names, so the task's `hook_type` and the
        // config's `hooks` list both carry the worker's spelling.
        let mut config = config_with(serde_json::json!({"class_name": "my_pkg.Minimal"}));
        config.hooks = vec![
            "identity.resolve".into(),
            "token.delegate".into(),
            "cmf.prompt_pre_invoke".into(),
            "cmf.tool_pre_invoke".into(),
        ];
        let plugin = IsolatedPythonPlugin::from_config(&config).expect("parses");

        for (dispatched, worker) in [
            ("identity.resolve", "identity_resolve"),
            ("token.delegate", "token_delegate"),
            ("cmf.prompt_pre_invoke", "cmf.prompt_pre_fetch"),
            ("cmf.tool_pre_invoke", "cmf.tool_pre_invoke"),
        ] {
            let task = plugin
                .build_task(
                    dispatched,
                    false,
                    serde_json::json!({}),
                    &PluginContext::new(),
                )
                .expect("task builds");
            assert_eq!(task["hook_type"], worker, "hook_type for {dispatched}");
        }

        let task = plugin
            .build_task(
                "identity.resolve",
                false,
                serde_json::json!({}),
                &PluginContext::new(),
            )
            .expect("task builds");
        assert_eq!(
            worker_config(&task)["hooks"],
            serde_json::json!([
                "identity_resolve",
                "token_delegate",
                "cmf.prompt_pre_fetch",
                "cmf.tool_pre_invoke"
            ])
        );
    }

    #[test]
    fn under_the_shim_cmf_hooks_reach_the_worker_under_legacy_names() {
        let mut config = config_with(serde_json::json!({"class_name": "my_pkg.Minimal"}));
        config.hooks = vec![
            "cmf.tool_pre_invoke".into(),
            "cmf.prompt_post_invoke".into(),
            "cmf.resource_pre_fetch".into(),
            "cmf.llm_input".into(),
            "identity.resolve".into(),
        ];
        let plugin = IsolatedPythonPlugin::from_config(&config).expect("parses");

        let task = plugin
            .build_task(
                "cmf.prompt_post_invoke",
                true,
                serde_json::json!({}),
                &PluginContext::new(),
            )
            .expect("task builds");
        assert_eq!(task["hook_type"], "prompt_post_fetch");
        assert_eq!(
            worker_config(&task)["hooks"],
            serde_json::json!([
                "tool_pre_invoke",
                "prompt_post_fetch",
                "resource_pre_fetch",
                // No legacy name; the adapter refuses it before any task is built.
                "cmf.llm_input",
                "identity_resolve"
            ])
        );
    }

    #[test]
    fn the_resolved_plugin_dir_is_what_the_worker_receives() {
        // The dir has to reach the worker's `sys.path`, not just the venv
        // layout — otherwise the venv is built in the right place and the
        // import still fails.
        let plugin = IsolatedPythonPlugin::from_config(&config_with(serde_json::json!({
            "class_name": "my_pkg.Minimal",
            "plugin_dirs": ["/ignored"],
        })))
        .expect("parses");

        let task = plugin
            .build_task(
                "tool_pre_invoke",
                false,
                serde_json::json!({"name": "t"}),
                &PluginContext::new(),
            )
            .expect("task builds");

        assert_eq!(
            task["plugin_dirs"],
            serde_json::json!([expected_default_dir()]),
            "the worker must be told the resolved dir, not the ignored one"
        );
    }

    /// Parse the `config` field, which travels as a JSON *string*.
    fn worker_config(task: &serde_json::Value) -> serde_json::Value {
        serde_json::from_str(task["config"].as_str().expect("config is a string"))
            .expect("the worker's config parses")
    }

    /// Build a task for a plugin declaring `capabilities`.
    fn task_for_capabilities(capabilities: &[&str]) -> serde_json::Value {
        let mut config = config_with(serde_json::json!({"class_name": "my_pkg.Minimal"}));
        config.capabilities = capabilities.iter().map(|c| (*c).to_string()).collect();

        IsolatedPythonPlugin::from_config(&config)
            .expect("parses")
            .build_task(
                "tool_pre_invoke",
                false,
                serde_json::json!({"name": "t"}),
                &PluginContext::new(),
            )
            .expect("task builds")
    }

    #[test]
    fn declared_capabilities_reach_the_workers_config() {
        // The Python manager re-filters extensions against the capabilities in
        // *its* config before calling a 3-arg hook. Send none and that filter
        // runs against an empty set, stripping every gated slot the executor
        // just allowed — which is how a plugin declaring `read_labels` ended up
        // seeing an empty label set out-of-process.
        let config = worker_config(&task_for_capabilities(&["read_labels", "append_labels"]));

        let mut sent: Vec<&str> = config["capabilities"]
            .as_array()
            .expect("capabilities is an array")
            .iter()
            .map(|c| c.as_str().expect("a string"))
            .collect();
        sent.sort_unstable();

        assert_eq!(
            sent,
            ["append_labels", "read_labels"],
            "the worker must be told what the plugin declared"
        );
    }

    #[test]
    fn a_capability_the_python_model_rejects_is_not_forwarded() {
        // `PluginConfig`'s validator raises on an unknown capability, and it
        // does so while the worker builds the config — before the plugin runs.
        // So forwarding a host-only name costs every hook of that plugin, not
        // just the slot it gates. Dropping it can only narrow the view, which
        // the executor's own filter has already done.
        let config = worker_config(&task_for_capabilities(&["read_labels", "read_client"]));

        assert_eq!(
            config["capabilities"],
            serde_json::json!(["read_labels"]),
            "a name the installed framework cannot validate must be dropped, \
             not sent and not escalated into a task failure"
        );
    }

    #[test]
    fn a_plugin_declaring_nothing_sends_an_empty_capability_list() {
        // The field is always present so the worker's config shape does not
        // depend on whether anything was declared; an empty list and an absent
        // field mean the same thing to Pydantic, and the explicit one is easier
        // to read off the wire when diagnosing a filtering question.
        let config = worker_config(&task_for_capabilities(&[]));

        assert_eq!(
            config["capabilities"],
            serde_json::json!([]),
            "no capabilities is an empty list, not a missing field"
        );
    }

    // --- adapter dispatch ---------------------------------------------------

    /// A stub worker that answers `load_and_run_hook` with a canned result and
    /// records the task it received.
    ///
    /// Behavior is driven by a marker file so a single script covers every
    /// case: the test writes `mode` next to the script before invoking.
    const ADAPTER_STUB: &str = r#"
import json, os, sys

here = os.path.dirname(os.path.abspath(__file__))

def mode_file(name, default=""):
    try:
        with open(os.path.join(here, name)) as f:
            return f.read().strip()
    except FileNotFoundError:
        return default

def mode():
    return mode_file("mode", "allow")

def append(name, text):
    with open(os.path.join(here, name), "a") as f:
        f.write(text + "\n")

def count(name):
    try:
        with open(os.path.join(here, name)) as f:
            return len(f.read().splitlines())
    except FileNotFoundError:
        return 0

# One line per launched process, so a test can tell whether a worker started.
append("spawns", str(os.getpid()))

while True:
    line = sys.stdin.readline()
    if not line:
        break
    line = line.strip()
    if not line:
        continue
    task = json.loads(line)
    rid = task.get("request_id", "unknown")

    if task.get("task_type") == "shutdown":
        print(json.dumps({"status": "success", "request_id": rid}), flush=True)
        break

    # Answered before the recording below, so the spawn-time handshake does not
    # overwrite the `last_task.json` the tests assert their hook task against.
    # This stub stands in for a *current* worker, so it claims the features the
    # real one implements; `handshake_mode` covers the old-worker case.
    if task.get("task_type") == "capabilities":
        hs = mode_file("handshake_mode")
        append("probes", hs)
        # An over-limit reply fails the probe the way a transient transport
        # fault does, without the delay a timeout would cost. `flaky_cmf` fails
        # only the first probe; `broken` fails every one.
        if hs == "broken" or (hs == "flaky_cmf" and count("probes") == 1):
            print(json.dumps({"request_id": rid, "pad": "x" * 8192}), flush=True)
            continue
        if hs == "unsupported":
            # Exactly what a worker predating the handshake replies.
            print(json.dumps({"status": "error", "message": "task type not supported.", "request_id": rid}), flush=True)
            continue
        features = [] if hs == "no_features" else ["credential", "extensions", "modified_extensions"]
        if hs in ("cmf", "flaky_cmf"):
            features.append("cmf_hooks")
        print(json.dumps({
            "status": "success", "protocol_version": 1,
            "features": features, "cpex_version": "0.1.2", "request_id": rid,
        }), flush=True)
        continue

    # Record the task so the test can assert on what the host actually sent.
    with open(os.path.join(here, "last_task.json"), "w") as f:
        json.dump(task, f)

    # ...and which process served it, so a respawn test can prove the request
    # was answered by a *new* worker rather than by the original recovering.
    with open(os.path.join(here, "last_pid"), "w") as f:
        f.write(str(os.getpid()))

    m = mode()

    # `die` exits without replying, which is how a test kills the worker so the
    # *next* invocation has a dead one to respawn.
    if m == "die":
        sys.exit(9)

    if m in ("deny", "deny_unmappable"):
        response = {
            "continue_processing": False,
            "violation": {"code": "PII_DETECTED", "reason": "email in args", "details": {"field": "q"}},
        }
        if m == "deny_unmappable":
            response["modified_payload"] = {"headers": {"authorization": "invalid"}}
    elif m == "error":
        response = {"status": "error", "message": "the plugin raised ValueError"}
    elif m == "context":
        response = {
            "continue_processing": True,
            "context": {"state": {"seen": 1}, "global_context": {"state": {"shared": "yes"}}},
        }
    elif m == "modify":
        response = {
            "continue_processing": True,
            "modified_payload": {"name": "search", "args": {"q": "[REDACTED]"}},
        }
    elif m == "identity_result":
        # What the worker dumps for a Python IdentityResult.
        response = {
            "continue_processing": True,
            "modified_payload": {
                "subject": {"id": "alice", "type": "user", "roles": ["hr"],
                            "permissions": [], "teams": [], "claims": {}},
                "delegation": None, "rejected": False, "reject_status": 401,
                "reject_reason": "", "raw_claims": {"sub": "alice"},
            },
        }
    elif m == "identity_rejected":
        response = {
            "continue_processing": True,
            "modified_payload": {"subject": None, "rejected": True,
                                 "reject_status": 401, "reject_reason": "token expired"},
        }
    elif m == "delegation_result":
        response = {
            "continue_processing": True,
            "modified_payload": {
                "delegated_token": "STUB-MINTED-TOKEN",
                "delegation_update": {"chain": [], "depth": 0, "origin_subject_id": None,
                                      "actor_subject_id": None, "delegated": False,
                                      "age_seconds": 0.0},
                "forwarded_headers": {}, "cache_key": None, "cache_ttl": None,
            },
        }
    else:
        response = {"continue_processing": True}

    response["request_id"] = rid
    print(json.dumps(response), flush=True)
"#;

    /// Build a plugin wired to the adapter stub, plus its scratch dir.
    async fn adapter_plugin(
        hooks: &[&str],
        capabilities: &[&str],
    ) -> (TempDir, Arc<IsolatedPythonPlugin>) {
        let (dir, plugin) = uninitialized_adapter_plugin(hooks, capabilities);
        plugin.initialize().await.expect("the stub worker launches");
        (dir, plugin)
    }

    fn set_mode(dir: &TempDir, mode: &str) {
        std::fs::write(dir.path().join("mode"), mode).unwrap();
    }

    /// How the stub should answer the spawn-time `capabilities` handshake.
    ///
    /// `"unsupported"` reproduces a worker predating the handshake;
    /// `"no_features"` a worker that knows the task but implements nothing.
    fn set_handshake_mode(dir: &TempDir, mode: &str) {
        std::fs::write(dir.path().join("handshake_mode"), mode).unwrap();
    }

    /// Build a plugin wired to the adapter stub *without* initializing it.
    ///
    /// Splits the scratch dir from `initialize()` so a test can set the stub's
    /// handshake mode first and then assert on the startup result — which
    /// `adapter_plugin` cannot, since it unwraps.
    fn uninitialized_adapter_plugin(
        hooks: &[&str],
        capabilities: &[&str],
    ) -> (TempDir, Arc<IsolatedPythonPlugin>) {
        uninitialized_adapter_plugin_with(
            hooks,
            capabilities,
            serde_json::json!({ "class_name": "my_pkg.Plugin" }),
        )
    }

    /// [`uninitialized_adapter_plugin`] with an explicit `config:` block.
    fn uninitialized_adapter_plugin_with(
        hooks: &[&str],
        capabilities: &[&str],
        config_block: serde_json::Value,
    ) -> (TempDir, Arc<IsolatedPythonPlugin>) {
        let dir = TempDir::new();
        let script = dir.path().join("adapter_stub.py");
        std::fs::write(&script, ADAPTER_STUB).unwrap();

        let config = PluginConfig {
            name: "py-plugin".into(),
            kind: crate::factory::KIND.into(),
            hooks: hooks.iter().map(|h| (*h).to_string()).collect(),
            capabilities: capabilities.iter().map(|c| (*c).to_string()).collect(),
            config: Some(config_block),
            ..Default::default()
        };

        let plugin = Arc::new(
            IsolatedPythonPlugin::from_config(&config)
                .expect("config parses")
                .with_worker_override(script),
        );
        (dir, plugin)
    }

    fn last_task(dir: &TempDir) -> serde_json::Value {
        let raw = std::fs::read_to_string(dir.path().join("last_task.json"))
            .expect("the stub recorded a task");
        serde_json::from_str(&raw).expect("recorded task is JSON")
    }

    /// Invoke a hook through the adapter and return the erased result fields.
    async fn invoke(
        plugin: &Arc<IsolatedPythonPlugin>,
        hook: &'static str,
        payload: &dyn PluginPayload,
        ctx: &mut PluginContext,
    ) -> Result<ppe::praxis_policy_core::executor::ErasedResultFields, Box<PluginError>> {
        let adapter = PythonHookAdapter::new(Arc::clone(plugin), hook);
        let erased = adapter.invoke(payload, &Extensions::default(), ctx).await?;
        Ok(ppe::praxis_policy_core::executor::extract_erased(erased)
            .expect("the adapter returns ErasedResultFields"))
    }

    #[tokio::test]
    async fn an_invoke_round_trips_and_returns_allow() {
        if crate::testing::skip_without_python3("an_invoke_round_trips_and_returns_allow") {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &[]).await;
        set_mode(&dir, "allow");

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            args: Some(std::collections::HashMap::from([(
                "q".into(),
                serde_json::json!("rust"),
            )])),
            headers: None,
        };
        let mut ctx = PluginContext::new();
        let fields = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .expect("the round trip succeeds");

        assert!(fields.continue_processing);
        assert!(fields.violation.is_none());

        // The task the worker received must carry the protocol contract.
        let task = last_task(&dir);
        assert_eq!(task["task_type"], "load_and_run_hook");
        assert_eq!(task["hook_type"], "tool_pre_invoke");
        assert_eq!(task["plugin_name"], "py-plugin");
        assert_eq!(task["class_name"], "my_pkg.Plugin");
        assert_eq!(task["payload"]["name"], "search");
        assert_eq!(task["payload"]["args"]["q"], "rust");

        // `config` is a JSON *string* the worker json.loads()es — not an object.
        let config_str = task["config"]
            .as_str()
            .expect("config must be a JSON string");
        let config: serde_json::Value = serde_json::from_str(config_str).unwrap();
        assert_eq!(config["name"], "py-plugin");

        plugin.shutdown().await.expect("shuts down");
    }

    #[tokio::test]
    async fn a_deny_response_becomes_a_deny_result_with_its_violation() {
        if crate::testing::skip_without_python3(
            "a_deny_response_becomes_a_deny_result_with_its_violation",
        ) {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &[]).await;
        set_mode(&dir, "deny");

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        let fields = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .expect("a deny is a successful invocation that returns a deny");

        assert!(!fields.continue_processing);
        let violation = fields.violation.expect("a deny must carry its violation");
        assert_eq!(violation.code, "PII_DETECTED");
        assert_eq!(violation.reason, "email in args");
        assert_eq!(violation.details["field"], "q");

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_worker_error_surfaces_as_a_plugin_error_for_the_executor_to_police() {
        // The host returns the error and stops. Applying the on_error policy is
        // the executor's job — doing it here would double-apply it.
        if crate::testing::skip_without_python3(
            "a_worker_error_surfaces_as_a_plugin_error_for_the_executor_to_police",
        ) {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &[]).await;
        set_mode(&dir, "error");

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        let Err(err) = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await else {
            panic!("a worker error response must not read as success");
        };

        match *err {
            PluginError::Execution {
                ref plugin_name,
                ref code,
                ref message,
                ..
            } => {
                assert_eq!(plugin_name, "py-plugin");
                assert_eq!(code.as_deref(), Some("worker_error"));
                assert!(
                    message.contains("ValueError"),
                    "the plugin's message should survive: {message}"
                );
            },
            ref other => panic!("expected an Execution error, got {other:?}"),
        }

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn context_deltas_are_written_back_into_the_plugin_context() {
        if crate::testing::skip_without_python3(
            "context_deltas_are_written_back_into_the_plugin_context",
        ) {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &[]).await;
        set_mode(&dir, "context");

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        ctx.set_local("preexisting", serde_json::json!("kept"));

        invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .unwrap();

        assert_eq!(ctx.get_local("seen"), Some(&serde_json::json!(1)));
        assert_eq!(ctx.get_global("shared"), Some(&serde_json::json!("yes")));
        assert_eq!(
            ctx.get_local("preexisting"),
            Some(&serde_json::json!("kept")),
            "deltas merge — a plugin touching one key must not blank the rest"
        );

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_modified_payload_comes_back_as_the_hooks_typed_payload() {
        if crate::testing::skip_without_python3(
            "a_modified_payload_comes_back_as_the_hooks_typed_payload",
        ) {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &[]).await;
        set_mode(&dir, "modify");

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        let fields = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .unwrap();

        let modified = fields.modified_payload.expect("the modification survives");
        let typed = modified
            .as_any()
            .downcast_ref::<crate::legacy::ToolPreInvokePayload>()
            .expect("must come back as the tool payload or the executor's downcast fails");
        assert_eq!(typed.args.as_ref().unwrap()["q"], "[REDACTED]");

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_cmf_hook_sends_the_message_payload_shape_to_a_cmf_worker() {
        if crate::testing::skip_without_python3(
            "a_cmf_hook_sends_the_message_payload_shape_to_a_cmf_worker",
        ) {
            return;
        }
        let (dir, plugin) = uninitialized_adapter_plugin(&["cmf.tool_pre_invoke"], &[]);
        // A worker claiming `cmf_hooks` gets the CMF message, not the shim.
        set_handshake_mode(&dir, "cmf");
        plugin.initialize().await.expect("the stub worker launches");
        set_mode(&dir, "allow");

        let payload = ppe::praxis_policy_core::cmf::MessagePayload {
            message: ppe::praxis_policy_core::cmf::Message::text(
                ppe::praxis_policy_core::cmf::Role::User,
                "hello",
            ),
        };
        let mut ctx = PluginContext::new();
        invoke(&plugin, "cmf.tool_pre_invoke", &payload, &mut ctx)
            .await
            .unwrap();

        let task = last_task(&dir);
        assert_eq!(task["hook_type"], "cmf.tool_pre_invoke");
        assert_eq!(
            task["payload"]["message"]["role"], "user",
            "a cmf hook must send the CMF message payload, not the generic wrapper"
        );

        plugin.shutdown().await.unwrap();
    }

    /// A CMF tool call after an empty text part, for the shim dispatch tests.
    /// The shim keeps the empty sibling part as it was.
    fn cmf_tool_call(q: &str) -> ppe::praxis_policy_core::cmf::MessagePayload {
        use ppe::praxis_policy_core::cmf::{ContentPart, Message, Role, ToolCall};
        ppe::praxis_policy_core::cmf::MessagePayload {
            message: Message::with_content(
                Role::Assistant,
                vec![
                    ContentPart::Text {
                        text: String::new(),
                    },
                    ContentPart::ToolCall {
                        content: ToolCall {
                            tool_call_id: "call-7".into(),
                            name: "search".into(),
                            arguments: std::collections::HashMap::from([(
                                "q".into(),
                                serde_json::json!(q),
                            )]),
                            namespace: None,
                        },
                    },
                ],
            ),
        }
    }

    #[tokio::test]
    async fn a_cmf_hook_reaches_an_old_worker_as_the_legacy_payload() {
        if crate::testing::skip_without_python3(
            "a_cmf_hook_reaches_an_old_worker_as_the_legacy_payload",
        ) {
            return;
        }
        // The default stub handshake claims no `cmf_hooks`, like a 0.1.x worker.
        let (dir, plugin) = adapter_plugin(&["cmf.tool_pre_invoke"], &[]).await;
        set_mode(&dir, "allow");

        let mut ctx = PluginContext::new();
        let fields = invoke(
            &plugin,
            "cmf.tool_pre_invoke",
            &cmf_tool_call("rust"),
            &mut ctx,
        )
        .await
        .expect("the shimmed round trip succeeds");
        assert!(fields.continue_processing);
        assert!(fields.modified_payload.is_none());

        let task = last_task(&dir);
        assert_eq!(task["hook_type"], "tool_pre_invoke");
        assert_eq!(
            task["payload"],
            serde_json::json!({"name": "search", "args": {"q": "rust"}})
        );
        assert_eq!(
            worker_config(&task)["hooks"],
            serde_json::json!(["tool_pre_invoke"])
        );

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_legacy_modification_maps_back_onto_the_cmf_message() {
        if crate::testing::skip_without_python3(
            "a_legacy_modification_maps_back_onto_the_cmf_message",
        ) {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["cmf.tool_pre_invoke"], &[]).await;
        // Returns `{"name": "search", "args": {"q": "[REDACTED]"}}`.
        set_mode(&dir, "modify");

        let mut ctx = PluginContext::new();
        let fields = invoke(
            &plugin,
            "cmf.tool_pre_invoke",
            &cmf_tool_call("alice@example.com"),
            &mut ctx,
        )
        .await
        .unwrap();

        let modified = fields.modified_payload.expect("the modification survives");
        let message = modified
            .as_any()
            .downcast_ref::<ppe::praxis_policy_core::cmf::MessagePayload>()
            .expect("a cmf hook's modification must come back as a MessagePayload");
        let json = serde_json::to_value(message).unwrap();
        assert_eq!(json["message"]["content"][0]["text"], "");
        assert_eq!(
            json["message"]["content"][1]["content"]["arguments"]["q"],
            "[REDACTED]"
        );
        assert_eq!(
            json["message"]["content"][1]["content"]["tool_call_id"],
            "call-7"
        );

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_legacy_deny_passes_through_the_shim_unchanged() {
        if crate::testing::skip_without_python3("a_legacy_deny_passes_through_the_shim_unchanged") {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["cmf.tool_pre_invoke"], &[]).await;
        for mode in ["deny", "deny_unmappable"] {
            set_mode(&dir, mode);
            let mut ctx = PluginContext::new();
            let fields = invoke(
                &plugin,
                "cmf.tool_pre_invoke",
                &cmf_tool_call("x"),
                &mut ctx,
            )
            .await
            .expect("an unusable modification must not replace a denial with an error");

            assert!(!fields.continue_processing);
            assert!(fields.modified_payload.is_none());
            let violation = fields.violation.expect("the deny keeps its violation");
            assert_eq!(violation.code, "PII_DETECTED");
            assert_eq!(violation.details["field"], "q");
        }

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn an_unrepresentable_message_fails_closed_before_reaching_the_worker() {
        if crate::testing::skip_without_python3(
            "an_unrepresentable_message_fails_closed_before_reaching_the_worker",
        ) {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["cmf.tool_pre_invoke"], &[]).await;
        set_mode(&dir, "allow");

        let no_tool_call = ppe::praxis_policy_core::cmf::MessagePayload {
            message: ppe::praxis_policy_core::cmf::Message::text(
                ppe::praxis_policy_core::cmf::Role::User,
                "hello",
            ),
        };
        for hook in ["cmf.tool_pre_invoke"] {
            let mut ctx = PluginContext::new();
            let Err(err) = invoke(&plugin, hook, &no_tool_call, &mut ctx).await else {
                panic!("'{hook}' must fail closed rather than allow");
            };
            let message = err.to_string();
            assert!(message.contains(hook), "names the hook: {message}");
            assert!(message.contains("py-plugin"), "names the plugin: {message}");
            match *err {
                PluginError::Execution { ref code, .. } => {
                    assert_eq!(code.as_deref(), Some("legacy_shim_unsupported"));
                },
                ref other => panic!("expected an Execution error, got {other:?}"),
            }
        }
        assert!(
            !dir.path().join("last_task.json").exists(),
            "nothing may reach the worker for a message the shim refused"
        );

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn invoking_before_initialize_is_a_plugin_error_not_a_panic() {
        let config = PluginConfig {
            name: "uninitialized".into(),
            kind: crate::factory::KIND.into(),
            hooks: vec!["tool_pre_invoke".into()],
            config: Some(serde_json::json!({ "class_name": "my_pkg.Plugin" })),
            ..Default::default()
        };
        let plugin = Arc::new(IsolatedPythonPlugin::from_config(&config).unwrap());

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        let Err(err) = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await else {
            panic!("dispatch without a worker cannot succeed");
        };
        assert!(matches!(*err, PluginError::Execution { .. }));
    }

    #[tokio::test]
    async fn a_credential_capability_attaches_the_dto_on_an_identity_hook() {
        // Ties the capability gate to real dispatch: the DTO must reach the
        // worker's task, on the contract's field name.
        if crate::testing::skip_without_python3(
            "a_credential_capability_attaches_the_dto_on_an_identity_hook",
        ) {
            return;
        }
        use ppe::praxis_policy_core::extensions::raw_credentials::{
            RawCredentialsExtension, RawInboundToken, TokenKind, TokenRole,
        };

        let (dir, plugin) =
            adapter_plugin(&["identity.resolve"], &["read_inbound_credentials"]).await;
        set_mode(&dir, "allow");

        let mut raw = RawCredentialsExtension::default();
        raw.inbound_tokens.insert(
            TokenRole::User,
            RawInboundToken::new("E2E-SECRET-TOKEN", "Authorization", TokenKind::Jwt),
        );
        let extensions = Extensions {
            raw_credentials: Some(Arc::new(raw)),
            ..Default::default()
        };

        let payload = crate::legacy::IdentityResolvePayload::default();
        let mut ctx = PluginContext::new();
        let adapter = PythonHookAdapter::new(Arc::clone(&plugin), "identity.resolve");
        adapter
            .invoke(&payload, &extensions, &mut ctx)
            .await
            .expect("dispatch succeeds");

        let task = last_task(&dir);
        assert_eq!(task["credential"]["inbound"]["token"], "E2E-SECRET-TOKEN");
        assert_eq!(task["credential"]["inbound"]["kind"], "jwt");
        assert_eq!(
            task["hook_type"], "identity_resolve",
            "the worker registers the identity hook under its own spelling"
        );

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_credential_capability_attaches_the_dto_on_a_delegation_hook() {
        if crate::testing::skip_without_python3(
            "a_credential_capability_attaches_the_dto_on_a_delegation_hook",
        ) {
            return;
        }
        use ppe::praxis_policy_core::extensions::raw_credentials::{
            DelegationKey, DelegationMode, RawCredentialsExtension, RawDelegatedToken,
        };

        let (dir, plugin) = adapter_plugin(&["token.delegate"], &["read_delegated_tokens"]).await;
        set_mode(&dir, "allow");

        let audience = "https://billing.example.com";
        let mut raw = RawCredentialsExtension::default();
        raw.delegated_tokens.insert(
            DelegationKey::new(DelegationMode::OnBehalfOfUser, audience, vec![]),
            RawDelegatedToken::new(
                "E2E-DELEGATED-TOKEN",
                "Authorization",
                audience,
                vec![],
                chrono::Utc::now(),
            ),
        );
        let extensions = Extensions {
            raw_credentials: Some(Arc::new(raw)),
            ..Default::default()
        };

        let payload = crate::legacy::TokenDelegatePayload {
            target_name: "billing".into(),
            target_audience: Some(audience.into()),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        let adapter = PythonHookAdapter::new(Arc::clone(&plugin), "token.delegate");
        adapter
            .invoke(&payload, &extensions, &mut ctx)
            .await
            .expect("dispatch succeeds");

        let task = last_task(&dir);
        assert_eq!(
            task["credential"]["delegated"]["token"],
            "E2E-DELEGATED-TOKEN"
        );
        assert_eq!(task["hook_type"], "token_delegate");

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn no_capability_means_no_credential_reaches_the_worker() {
        if crate::testing::skip_without_python3(
            "no_capability_means_no_credential_reaches_the_worker",
        ) {
            return;
        }
        use ppe::praxis_policy_core::extensions::raw_credentials::{
            RawCredentialsExtension, RawInboundToken, TokenKind, TokenRole,
        };

        // Same hook, same request — this plugin simply declared nothing.
        let (dir, plugin) = adapter_plugin(&["identity.resolve"], &[]).await;
        set_mode(&dir, "allow");

        let mut raw = RawCredentialsExtension::default();
        raw.inbound_tokens.insert(
            TokenRole::User,
            RawInboundToken::new("E2E-SECRET-TOKEN", "Authorization", TokenKind::Jwt),
        );
        let extensions = Extensions {
            raw_credentials: Some(Arc::new(raw)),
            ..Default::default()
        };

        let payload = crate::legacy::IdentityResolvePayload::default();
        let mut ctx = PluginContext::new();
        let adapter = PythonHookAdapter::new(Arc::clone(&plugin), "identity.resolve");
        adapter
            .invoke(&payload, &extensions, &mut ctx)
            .await
            .unwrap();

        let task = last_task(&dir);
        assert!(task.get("credential").is_none(), "no credential field");
        assert!(
            !serde_json::to_string(&task)
                .unwrap()
                .contains("E2E-SECRET-TOKEN"),
            "no token bytes anywhere in the task the worker received"
        );

        plugin.shutdown().await.unwrap();
    }

    // --- dead-worker respawn ------------------------------------------------

    fn last_pid(dir: &TempDir) -> u32 {
        std::fs::read_to_string(dir.path().join("last_pid"))
            .expect("the stub recorded its pid")
            .trim()
            .parse()
            .expect("pid is numeric")
    }

    /// Wait until the client has observed the worker's death.
    ///
    /// `is_alive()` flips when the reader task sees EOF, which is asynchronous
    /// with respect to the process exiting. Polling for it keeps the respawn
    /// tests deterministic without a blind sleep.
    async fn await_worker_death(plugin: &Arc<IsolatedPythonPlugin>) {
        for _ in 0..200 {
            let alive = plugin
                .worker
                .read()
                .await
                .as_ref()
                .is_some_and(|w| w.is_alive());
            if !alive {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        panic!("the worker never registered as dead");
    }

    #[tokio::test]
    async fn a_dead_worker_is_respawned_on_the_next_invocation() {
        // Pre-fix, one crash disabled the plugin for the life of the process:
        // the dead WorkerClient stayed in the slot and every later invocation
        // failed against it. This asserts recovery, and asserts it came from a
        // genuinely new process rather than the old one somehow answering.
        if crate::testing::skip_without_python3("a_dead_worker_is_respawned_on_the_next_invocation")
        {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &[]).await;
        set_mode(&dir, "allow");

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();

        // A healthy round trip establishes the original worker's pid.
        invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .expect("the first invocation succeeds");
        let original_pid = last_pid(&dir);

        // Kill it: the stub exits without replying.
        set_mode(&dir, "die");
        let _ = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await;
        await_worker_death(&plugin).await;

        // The next invocation must respawn and succeed rather than reporting a
        // permanently broken plugin.
        set_mode(&dir, "allow");
        let fields = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .expect("a dead worker must be replaced, not left broken");
        assert!(fields.continue_processing);

        let new_pid = last_pid(&dir);
        assert_ne!(
            original_pid, new_pid,
            "the request must be served by a respawned process"
        );

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_worker_that_cannot_honor_a_declared_credential_fails_closed_at_startup() {
        // Fail-closed rule 2. An older worker.py does not *reject* the credential
        // field, it ignores it — so before the handshake, a plugin declaring
        // read_inbound_credentials started clean and then ran with no credential
        // at all, silently. That is reachable in production because two framework
        // builds shipped as 0.1.2 with different worker protocols.
        if crate::testing::skip_without_python3(
            "a_worker_that_cannot_honor_a_declared_credential_fails_closed_at_startup",
        ) {
            return;
        }
        let (dir, plugin) =
            uninitialized_adapter_plugin(&["identity.resolve"], &["read_inbound_credentials"]);
        // A worker predating the handshake: replies "task type not supported."
        set_handshake_mode(&dir, "unsupported");

        let err = plugin
            .initialize()
            .await
            .expect_err("a worker that would drop the credential must not be installed");

        let message = err.to_string();
        assert!(
            message.contains("does not implement credential"),
            "the failure must name the missing feature: {message}"
        );
        assert!(
            message.contains("upgrade the cpex package"),
            "the failure must tell an operator what to do: {message}"
        );

        // And the plugin must be left with no worker rather than a half-installed
        // one a later invocation could still reach.
        assert!(
            plugin.worker.read().await.is_none(),
            "a failed handshake must not leave a worker in the slot"
        );
    }

    #[tokio::test]
    async fn llm_hooks_require_a_worker_with_native_cmf_support() {
        if crate::testing::skip_without_python3(
            "llm_hooks_require_a_worker_with_native_cmf_support",
        ) {
            return;
        }
        for hook in ["cmf.llm_input", "cmf.llm_output"] {
            for handshake in ["no_features", "unsupported"] {
                let (dir, plugin) = uninitialized_adapter_plugin(&[hook], &[]);
                set_handshake_mode(&dir, handshake);
                let err = plugin
                    .initialize()
                    .await
                    .expect_err("the shim cannot serve an LLM hook");
                assert!(err.to_string().contains("cmf_hooks"), "{err}");
                assert!(plugin.worker.read().await.is_none());
            }
            let (dir, plugin) = uninitialized_adapter_plugin(&[hook], &[]);
            set_handshake_mode(&dir, "cmf");
            plugin
                .initialize()
                .await
                .expect("a native CMF worker can serve an LLM hook");
            plugin.shutdown().await.unwrap();
        }
    }

    #[tokio::test]
    async fn a_plugin_declaring_nothing_gated_still_runs_on_an_old_worker() {
        // The other half of the rule: the handshake must not become a blanket
        // version gate. A plugin that declares no gated capability is served
        // correctly by an old worker, and refusing to start it would break every
        // existing deployment for no safety gain.
        if crate::testing::skip_without_python3(
            "a_plugin_declaring_nothing_gated_still_runs_on_an_old_worker",
        ) {
            return;
        }
        let (dir, plugin) = uninitialized_adapter_plugin(&["tool_pre_invoke"], &[]);
        set_handshake_mode(&dir, "unsupported");

        plugin
            .initialize()
            .await
            .expect("a plugin needing no gated field must still start on an old worker");

        // ...and it must actually work, not merely start.
        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        let fields = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .expect("the hook still dispatches");
        assert!(fields.continue_processing);

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_worker_answering_the_handshake_with_no_features_also_fails_closed() {
        // A worker that knows the task but reports an empty feature set is not
        // the same case as one that cannot answer, and must be treated the same:
        // what matters is whether the feature is claimed, never whether the
        // worker was new enough to be asked.
        if crate::testing::skip_without_python3(
            "a_worker_answering_the_handshake_with_no_features_also_fails_closed",
        ) {
            return;
        }
        let (dir, plugin) =
            uninitialized_adapter_plugin(&["tool_pre_invoke"], &["read_labels", "append_labels"]);
        set_handshake_mode(&dir, "no_features");

        let err = plugin
            .initialize()
            .await
            .expect_err("an extensions-declaring plugin needs a worker that delivers the field");

        let message = err.to_string();
        assert!(
            message.contains("does not implement extensions"),
            "an extensions capability requires the extensions feature: {message}"
        );
    }

    #[tokio::test]
    async fn a_respawned_worker_receives_the_same_capability_handshake() {
        // The security rail. A respawned worker must be gated by exactly the
        // capabilities the original was — never a weaker set, and never a
        // "recovery mode" that skips the gate. Both directions are asserted
        // against one respawn: the declared capability still delivers its
        // credential, and an undeclared slot still delivers nothing.
        if crate::testing::skip_without_python3(
            "a_respawned_worker_receives_the_same_capability_handshake",
        ) {
            return;
        }
        use ppe::praxis_policy_core::extensions::raw_credentials::{
            RawCredentialsExtension, RawInboundToken, TokenKind, TokenRole,
        };

        let (dir, plugin) =
            adapter_plugin(&["identity.resolve"], &["read_inbound_credentials"]).await;
        set_mode(&dir, "allow");

        let mut raw = RawCredentialsExtension::default();
        raw.inbound_tokens.insert(
            TokenRole::User,
            RawInboundToken::new("RESPAWN-SECRET", "Authorization", TokenKind::Jwt),
        );
        let extensions = Extensions {
            raw_credentials: Some(Arc::new(raw)),
            ..Default::default()
        };

        let payload = crate::legacy::IdentityResolvePayload::default();
        let adapter = PythonHookAdapter::new(Arc::clone(&plugin), "identity.resolve");

        let mut ctx = PluginContext::new();
        adapter
            .invoke(&payload, &extensions, &mut ctx)
            .await
            .expect("the first dispatch succeeds");
        let original_pid = last_pid(&dir);
        assert_eq!(
            last_task(&dir)["credential"]["inbound"]["token"],
            "RESPAWN-SECRET"
        );

        set_mode(&dir, "die");
        let mut ctx = PluginContext::new();
        let _ = adapter.invoke(&payload, &extensions, &mut ctx).await;
        await_worker_death(&plugin).await;

        // Same hook, same declared capability, new process.
        set_mode(&dir, "allow");
        let mut ctx = PluginContext::new();
        adapter
            .invoke(&payload, &extensions, &mut ctx)
            .await
            .expect("dispatch succeeds against the respawned worker");

        assert_ne!(original_pid, last_pid(&dir), "a new process served it");
        let task = last_task(&dir);
        assert_eq!(
            task["credential"]["inbound"]["token"], "RESPAWN-SECRET",
            "the respawned worker must get the same credential handshake"
        );
        assert_eq!(task["credential"]["inbound"]["kind"], "jwt");

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_respawned_worker_gets_no_credential_it_did_not_declare() {
        // The fail-closed half: a plugin with no declared capability must not
        // pick one up by virtue of having been respawned.
        if crate::testing::skip_without_python3(
            "a_respawned_worker_gets_no_credential_it_did_not_declare",
        ) {
            return;
        }
        use ppe::praxis_policy_core::extensions::raw_credentials::{
            RawCredentialsExtension, RawInboundToken, TokenKind, TokenRole,
        };

        let (dir, plugin) = adapter_plugin(&["identity.resolve"], &[]).await;
        set_mode(&dir, "allow");

        let mut raw = RawCredentialsExtension::default();
        raw.inbound_tokens.insert(
            TokenRole::User,
            RawInboundToken::new("MUST-NOT-LEAK", "Authorization", TokenKind::Jwt),
        );
        let extensions = Extensions {
            raw_credentials: Some(Arc::new(raw)),
            ..Default::default()
        };

        let payload = crate::legacy::IdentityResolvePayload::default();
        let adapter = PythonHookAdapter::new(Arc::clone(&plugin), "identity.resolve");

        let mut ctx = PluginContext::new();
        adapter
            .invoke(&payload, &extensions, &mut ctx)
            .await
            .unwrap();

        set_mode(&dir, "die");
        let mut ctx = PluginContext::new();
        let _ = adapter.invoke(&payload, &extensions, &mut ctx).await;
        await_worker_death(&plugin).await;

        set_mode(&dir, "allow");
        let mut ctx = PluginContext::new();
        adapter
            .invoke(&payload, &extensions, &mut ctx)
            .await
            .expect("dispatch succeeds after respawn");

        let task = last_task(&dir);
        assert!(
            task.get("credential").is_none(),
            "respawn must not widen the capability gate"
        );
        assert!(
            !serde_json::to_string(&task)
                .unwrap()
                .contains("MUST-NOT-LEAK"),
            "no token bytes may reach a worker that declared nothing"
        );

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_worker_that_will_not_stay_up_gives_up_with_a_typed_error() {
        // The anti-loop rail. A worker that dies on every launch must surface a
        // typed error after a bounded number of attempts rather than retrying
        // forever. Pointing the override at a script that always exits makes
        // every respawn produce a corpse.
        if crate::testing::skip_without_python3(
            "a_worker_that_will_not_stay_up_gives_up_with_a_typed_error",
        ) {
            return;
        }
        let dir = TempDir::new();
        let script = dir.path().join("always_dies.py");
        // Exits at import, before reading a task — so every spawn dies.
        std::fs::write(
            &script,
            "import sys\nprint('fatal: dependency missing', file=sys.stderr, flush=True)\nsys.exit(1)\n",
        )
        .unwrap();

        let config = PluginConfig {
            name: "crash-loop".into(),
            kind: crate::factory::KIND.into(),
            hooks: vec!["tool_pre_invoke".into()],
            config: Some(serde_json::json!({ "class_name": "my_pkg.Plugin" })),
            ..Default::default()
        };
        let plugin = Arc::new(
            IsolatedPythonPlugin::from_config(&config)
                .expect("config parses")
                .with_worker_override(script),
        );
        // The process *starts* (then dies), so initialize succeeds — this is
        // exactly the shape a bad dependency pin produces.
        plugin.initialize().await.expect("the process launches");

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };

        // Drive invocations until one reports the budget is spent. The bound is
        // what is under test: without it this loop never terminates.
        let started = std::time::Instant::now();
        let mut gave_up = None;
        for _ in 0..(MAX_RESPAWN_ATTEMPTS + 4) {
            let mut ctx = PluginContext::new();
            let Err(err) = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await else {
                panic!("a worker that always dies cannot serve a request");
            };
            let message = err.to_string();
            if message.contains("could not be respawned") {
                gave_up = Some(message);
                break;
            }
        }

        let message = gave_up
            .expect("a permanently failing worker must give up with an error, not retry forever");
        assert!(
            message.contains(&MAX_RESPAWN_ATTEMPTS.to_string()),
            "the error should name the attempt bound: {message}"
        );

        // The backoff is bounded too — giving up must not take minutes.
        assert!(
            started.elapsed() < std::time::Duration::from_secs(20),
            "giving up took too long: {:?}",
            started.elapsed()
        );

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn respawn_is_not_attempted_before_initialize() {
        // Respawn must not paper over a lifecycle bug by starting a worker the
        // manager never asked for: an uninitialized plugin still errors.
        let config = PluginConfig {
            name: "uninitialized".into(),
            kind: crate::factory::KIND.into(),
            hooks: vec!["tool_pre_invoke".into()],
            config: Some(serde_json::json!({ "class_name": "my_pkg.Plugin" })),
            ..Default::default()
        };
        let plugin = Arc::new(IsolatedPythonPlugin::from_config(&config).unwrap());

        let Err(err) = plugin.worker().await else {
            panic!("a plugin with no worker must not silently spawn one");
        };
        let message = err.to_string();
        assert!(
            message.contains("initialize()"),
            "the error should name the lifecycle gap: {message}"
        );
        assert!(
            plugin.worker.read().await.is_none(),
            "no worker was started"
        );
    }

    #[tokio::test]
    async fn concurrent_invocations_share_one_respawn() {
        // Several hooks of one plugin can be in flight at once (the executor
        // dispatches parallel-phase plugins concurrently). They must not each
        // spend respawn budget on the same death and start competing workers.
        if crate::testing::skip_without_python3("concurrent_invocations_share_one_respawn") {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &[]).await;
        set_mode(&dir, "allow");

        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .unwrap();

        set_mode(&dir, "die");
        let _ = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await;
        await_worker_death(&plugin).await;
        set_mode(&dir, "allow");

        // Four concurrent invocations against one dead worker.
        let mut handles = Vec::new();
        for _ in 0..4 {
            let plugin = Arc::clone(&plugin);
            handles.push(tokio::spawn(async move {
                let payload = crate::legacy::ToolPreInvokePayload {
                    name: "search".into(),
                    ..Default::default()
                };
                let mut ctx = PluginContext::new();
                invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
                    .await
                    .map(|f| f.continue_processing)
            }));
        }

        for handle in handles {
            assert!(handle.await.unwrap().expect("each invocation recovers"));
        }

        // One respawn served all four, so the budget is intact — a second
        // independent death must still be recoverable.
        set_mode(&dir, "die");
        let mut ctx = PluginContext::new();
        let _ = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await;
        await_worker_death(&plugin).await;
        set_mode(&dir, "allow");

        let mut ctx = PluginContext::new();
        invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .expect("budget was not exhausted by the concurrent respawn");

        plugin.shutdown().await.unwrap();
    }

    /// Lines in a file the stub appends to, or 0 when it has none.
    fn lines_in(dir: &TempDir, name: &str) -> usize {
        std::fs::read_to_string(dir.path().join(name))
            .map(|s| s.lines().count())
            .unwrap_or(0)
    }

    #[tokio::test]
    async fn a_failed_respawn_is_retried_on_the_next_invocation() {
        // A respawn whose replacement fails verification used to leave the
        // slot empty, which every later call read as "initialize() did not
        // run": one bad respawn disabled the plugin for good.
        if crate::testing::skip_without_python3(
            "a_failed_respawn_is_retried_on_the_next_invocation",
        ) {
            return;
        }
        // `read_labels` makes verification require the `extensions` feature.
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &["read_labels"]).await;
        set_mode(&dir, "allow");
        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .unwrap();
        let original_pid = last_pid(&dir);

        set_mode(&dir, "die");
        let _ = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await;
        await_worker_death(&plugin).await;

        // The replacement launches but fails the handshake.
        set_handshake_mode(&dir, "no_features");
        set_mode(&dir, "allow");
        let Err(err) = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await else {
            panic!("a replacement that fails verification must not serve the request");
        };
        assert!(
            err.to_string().contains("does not implement extensions"),
            "{err}"
        );
        assert!(
            plugin.worker.read().await.is_some(),
            "the dead worker stays in the slot so a later call can respawn again"
        );

        // The cause is fixed; the next call respawns within the same budget.
        std::fs::remove_file(dir.path().join("handshake_mode")).unwrap();
        let fields = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .expect("a later invocation retries the respawn");
        assert!(fields.continue_processing);
        assert_ne!(original_pid, last_pid(&dir), "served by a new process");

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_respawn_racing_shutdown_launches_no_worker() {
        // A request that found the worker dead is in the respawn backoff when
        // the engine shuts the plugin down. The respawn must give up rather
        // than start a worker after shutdown, which nothing would stop.
        if crate::testing::skip_without_python3("a_respawn_racing_shutdown_launches_no_worker") {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["tool_pre_invoke"], &[]).await;
        set_mode(&dir, "allow");
        let payload = crate::legacy::ToolPreInvokePayload {
            name: "search".into(),
            ..Default::default()
        };
        let mut ctx = PluginContext::new();
        invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
            .await
            .unwrap();
        set_mode(&dir, "die");
        let _ = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await;
        await_worker_death(&plugin).await;
        set_mode(&dir, "allow");
        assert_eq!(lines_in(&dir, "spawns"), 1);

        let racing = {
            let plugin = Arc::clone(&plugin);
            tokio::spawn(async move {
                let payload = crate::legacy::ToolPreInvokePayload {
                    name: "search".into(),
                    ..Default::default()
                };
                let mut ctx = PluginContext::new();
                invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx)
                    .await
                    .map(|_| ())
            })
        };
        // Inside the first respawn's 100ms backoff.
        tokio::time::sleep(std::time::Duration::from_millis(30)).await;
        plugin.shutdown().await.unwrap();

        let Err(err) = racing.await.unwrap() else {
            panic!("an invocation racing shutdown must not be served by a new worker");
        };
        assert!(err.to_string().contains("shut down"), "{err}");

        // Give a wrongly launched worker time to show up before asserting none did.
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
        assert_eq!(
            lines_in(&dir, "spawns"),
            1,
            "no worker launched after shutdown"
        );
        assert!(plugin.worker.read().await.is_none(), "the slot stays empty");

        // And later calls report the shutdown rather than respawning.
        let Err(err) = invoke(&plugin, "tool_pre_invoke", &payload, &mut ctx).await else {
            panic!("a shut-down plugin cannot serve a request");
        };
        assert!(err.to_string().contains("shut down"), "{err}");
    }

    #[tokio::test]
    async fn a_transient_probe_failure_is_retried_before_choosing_the_shim() {
        if crate::testing::skip_without_python3(
            "a_transient_probe_failure_is_retried_before_choosing_the_shim",
        ) {
            return;
        }
        // A small frame cap lets the stub fail a probe with an over-limit reply.
        let (dir, plugin) = uninitialized_adapter_plugin_with(
            &["cmf.tool_pre_invoke"],
            &[],
            serde_json::json!({ "class_name": "my_pkg.Plugin", "max_content_size": 4096 }),
        );
        // The startup probe fails; any later one claims `cmf_hooks`.
        set_handshake_mode(&dir, "flaky_cmf");
        plugin
            .initialize()
            .await
            .expect("a failed probe does not stop a plugin that requires nothing");
        set_mode(&dir, "allow");

        let payload = ppe::praxis_policy_core::cmf::MessagePayload {
            message: ppe::praxis_policy_core::cmf::Message::text(
                ppe::praxis_policy_core::cmf::Role::User,
                "hello",
            ),
        };
        let mut ctx = PluginContext::new();
        invoke(&plugin, "cmf.tool_pre_invoke", &payload, &mut ctx)
            .await
            .expect("served natively once the retried probe answers");

        let task = last_task(&dir);
        assert_eq!(
            task["hook_type"], "cmf.tool_pre_invoke",
            "the retried probe's report chooses native dispatch, not the shim"
        );
        assert_eq!(lines_in(&dir, "probes"), 2);

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_probe_that_keeps_failing_is_retried_once_per_worker() {
        if crate::testing::skip_without_python3(
            "a_probe_that_keeps_failing_is_retried_once_per_worker",
        ) {
            return;
        }
        let (dir, plugin) = uninitialized_adapter_plugin_with(
            &["cmf.tool_pre_invoke"],
            &[],
            serde_json::json!({ "class_name": "my_pkg.Plugin", "max_content_size": 4096 }),
        );
        set_handshake_mode(&dir, "broken");
        plugin.initialize().await.unwrap();
        set_mode(&dir, "allow");

        for _ in 0..3 {
            let mut ctx = PluginContext::new();
            invoke(
                &plugin,
                "cmf.tool_pre_invoke",
                &cmf_tool_call("rust"),
                &mut ctx,
            )
            .await
            .expect("the shim still serves the hook");
            assert_eq!(last_task(&dir)["hook_type"], "tool_pre_invoke");
        }
        assert_eq!(
            lines_in(&dir, "probes"),
            2,
            "one startup probe and one retry, not one per call"
        );

        plugin.shutdown().await.unwrap();
    }

    // --- PPE identity and delegation payloads ---------------------------------

    fn ppe_identity() -> ppe::praxis_policy_core::identity::IdentityPayload {
        use ppe::praxis_policy_core::identity::{IdentityPayload, TokenSource};
        IdentityPayload::new("PPE-IDENTITY-SECRET-TOKEN", TokenSource::Bearer).with_headers(
            std::collections::HashMap::from([
                (
                    "authorization".into(),
                    "Bearer PPE-IDENTITY-SECRET-TOKEN".into(),
                ),
                ("cookie".into(), "session=PPE-COOKIE-SECRET".into()),
                ("x-tenant".into(), "acme".into()),
            ]),
        )
    }

    async fn invoke_on(
        plugin: &Arc<IsolatedPythonPlugin>,
        hook: &'static str,
        payload: &dyn PluginPayload,
    ) -> ppe::praxis_policy_core::executor::ErasedResultFields {
        let mut ctx = PluginContext::new();
        invoke(plugin, hook, payload, &mut ctx)
            .await
            .expect("dispatch succeeds")
    }

    #[tokio::test]
    async fn a_ppe_identity_payload_reaches_the_worker_without_its_credential() {
        if crate::testing::skip_without_python3(
            "a_ppe_identity_payload_reaches_the_worker_without_its_credential",
        ) {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["identity.resolve"], &[]).await;
        set_mode(&dir, "allow");
        invoke_on(&plugin, "identity.resolve", &ppe_identity()).await;

        let task = last_task(&dir);
        assert_eq!(task["hook_type"], "identity_resolve");
        assert_eq!(task["payload"]["headers"], serde_json::json!({}));
        assert!(task.get("credential").is_none());
        let wire = serde_json::to_string(&task).unwrap();
        assert!(!wire.contains("PPE-IDENTITY-SECRET-TOKEN"), "{wire}");
        assert!(!wire.contains("PPE-COOKIE-SECRET"), "{wire}");

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_ppe_identity_payload_gives_a_declaring_plugin_the_gated_credential() {
        if crate::testing::skip_without_python3(
            "a_ppe_identity_payload_gives_a_declaring_plugin_the_gated_credential",
        ) {
            return;
        }
        let (dir, plugin) =
            adapter_plugin(&["identity.resolve"], &["read_inbound_credentials"]).await;
        set_mode(&dir, "identity_result");
        let fields = invoke_on(&plugin, "identity.resolve", &ppe_identity()).await;

        let task = last_task(&dir);
        assert_eq!(
            task["credential"]["inbound"]["token"],
            "PPE-IDENTITY-SECRET-TOKEN"
        );
        assert!(
            !serde_json::to_string(&task["payload"])
                .unwrap()
                .contains("PPE-IDENTITY-SECRET-TOKEN"),
            "the token rides only the gated channel"
        );

        // The IdentityResult maps back onto the PPE payload PPE downcasts to.
        let modified = fields
            .modified_payload
            .expect("the result is a modification");
        let identity = modified
            .as_any()
            .downcast_ref::<ppe::praxis_policy_core::identity::IdentityPayload>()
            .expect("identity.resolve must come back as PPE's IdentityPayload");
        assert_eq!(
            identity.subject.as_ref().and_then(|s| s.id.as_deref()),
            Some("alice")
        );
        assert_eq!(identity.raw_token(), "PPE-IDENTITY-SECRET-TOKEN");

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_rejected_identity_result_becomes_a_deny() {
        if crate::testing::skip_without_python3("a_rejected_identity_result_becomes_a_deny") {
            return;
        }
        let (dir, plugin) = adapter_plugin(&["identity.resolve"], &[]).await;
        set_mode(&dir, "identity_rejected");
        let fields = invoke_on(&plugin, "identity.resolve", &ppe_identity()).await;
        assert!(!fields.continue_processing);
        assert!(fields.modified_payload.is_none());
        assert_eq!(fields.violation.unwrap().reason, "token expired");

        plugin.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn a_ppe_delegation_payload_round_trips_through_the_worker() {
        if crate::testing::skip_without_python3(
            "a_ppe_delegation_payload_round_trips_through_the_worker",
        ) {
            return;
        }
        use ppe::praxis_policy_core::delegation::DelegationPayload;

        // PPE's bearer token is the caller's inbound credential.
        let (dir, plugin) =
            adapter_plugin(&["token.delegate"], &["read_inbound_credentials"]).await;
        set_mode(&dir, "delegation_result");
        let payload = DelegationPayload::new("PPE-CALLER-BEARER", "billing")
            .with_target_audience("https://billing.example.com");
        let fields = invoke_on(&plugin, "token.delegate", &payload).await;

        let task = last_task(&dir);
        assert_eq!(task["hook_type"], "token_delegate");
        assert_eq!(task["payload"]["target_name"], "billing");
        assert!(task["payload"].get("bearer_token").is_none());
        assert_eq!(
            task["credential"]["delegated"]["token"],
            "PPE-CALLER-BEARER"
        );

        let modified = fields
            .modified_payload
            .expect("the result is a modification");
        let delegation = modified
            .as_any()
            .downcast_ref::<DelegationPayload>()
            .expect("token.delegate must come back as PPE's DelegationPayload");
        let minted = delegation.delegated_token.as_ref().expect("a minted token");
        assert_eq!(minted.token.as_str(), "STUB-MINTED-TOKEN");
        assert_eq!(minted.audience, "https://billing.example.com");

        plugin.shutdown().await.unwrap();
    }

    #[test]
    fn the_respawn_budget_is_bounded_then_resets_with_time() {
        // Unit-level pin on the accounting, so the bound and the reset window
        // are covered without spawning processes.
        let mut state = RespawnState::new();
        for expected in 1..=MAX_RESPAWN_ATTEMPTS {
            assert_eq!(state.try_reserve(), Some(expected));
        }
        assert_eq!(
            state.try_reserve(),
            None,
            "the budget must be bounded, not unlimited"
        );

        // A successful invocation refills it.
        state.record_success();
        assert_eq!(state.try_reserve(), Some(1));

        // ...and so does the passage of the reset window, so a plugin is not
        // poisoned for the life of the process once its cause is fixed.
        let mut state = RespawnState::new();
        while state.try_reserve().is_some() {}
        state.window_started = Some(
            std::time::Instant::now()
                - std::time::Duration::from_secs(RESPAWN_BUDGET_RESET_SECS + 1),
        );
        assert_eq!(
            state.try_reserve(),
            Some(1),
            "an elapsed window must restore the budget"
        );
    }

    #[test]
    fn unknown_config_keys_are_tolerated() {
        // Plugin `config:` blocks are shared with the Python side, which reads
        // keys this host does not model. Rejecting them would break configs
        // that work under the Python CLI.
        let cfg = VenvConfig::from_plugin_config(&config_with(serde_json::json!({
            "class_name": "my_pkg.Minimal",
            "some_python_only_setting": true,
        })))
        .expect("extra keys belong to the Python plugin, not this host");
        assert_eq!(cfg.class_name, "my_pkg.Minimal");
    }
}