agent_engine/extensions/runtime/

process.rs

//! Process-based extension runtime — JSON-RPC 2.0 over stdio.
//!
//! Spawns the extension as a child process. Communication uses
//! Content-Length framing (LSP-style) over stdin/stdout.

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStdin, ChildStdout, Command};
use tokio::sync::{mpsc, oneshot, Mutex, RwLock};
use tokio::task::JoinHandle;

use super::{ExtensionHandler, ExtensionHealth, ExtensionLifecycle, RestartPolicy};
use crate::extensions::hooks::events::{HookEvent, HookResult};
use crate::extensions::manifest::CURRENT_EXTENSION_PROTOCOL_VERSION;

#[derive(Serialize)]
struct JsonRpcRequest {
    jsonrpc: &'static str,
    method: String,
    params: Value,
    id: u64,
}

#[derive(Serialize)]
struct InitializeParams {
    synaps_version: &'static str,
    extension_protocol_version: u32,
    plugin_id: String,
    plugin_root: Option<String>,
    config: Value,
}

#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct RegisteredExtensionToolSpec {
    pub name: String,
    pub description: String,
    pub input_schema: Value,
}

#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct RegisteredProviderSpec {
    pub id: String,
    pub display_name: String,
    pub description: String,
    #[serde(default)]
    pub models: Vec<RegisteredProviderModelSpec>,
    #[serde(default)]
    pub config_schema: Option<Value>,
}

#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct RegisteredProviderModelSpec {
    pub id: String,
    #[serde(default)]
    pub display_name: Option<String>,
    #[serde(default)]
    pub capabilities: Value,
    #[serde(default)]
    pub context_window: Option<u64>,
}

#[derive(Debug, Clone, Serialize)]
pub struct ProviderCompleteParams {
    pub provider_id: String,
    pub model_id: String,
    pub model: String,
    pub messages: Vec<crate::SharedMessage>,
    pub system_prompt: Option<String>,
    pub tools: Vec<Value>,
    pub temperature: Option<f32>,
    pub max_tokens: Option<u32>,
    pub thinking_budget: u32,
}

#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct ProviderCompleteResult {
    pub content: Vec<Value>,
    #[serde(default)]
    pub stop_reason: Option<String>,
    #[serde(default)]
    pub usage: Option<Value>,
}

#[derive(Debug, Clone, PartialEq)]
pub struct ProviderToolUse {
    pub id: String,
    pub name: String,
    pub input: Value,
}

/// A single streaming event from a provider extension.
#[derive(Debug, Clone, PartialEq)]
pub enum ProviderStreamEvent {
    /// Incremental assistant text.
    TextDelta { text: String },
    /// Incremental thinking text.
    ThinkingDelta { text: String },
    /// A complete tool-use block (matches ProviderToolUse fields).
    ToolUse {
        id: String,
        name: String,
        input: Value,
    },
    /// Usage metadata (typically near end-of-stream).
    Usage { usage: Value },
    /// Provider-side error (non-fatal stream notification — caller decides).
    Error { message: String },
    /// Optional explicit end-of-stream marker.
    Done,
}

/// Parse a single `provider.stream.event` notification's `params` value into a
/// [`ProviderStreamEvent`]. Returns `Err(String)` on malformed input.
///
/// Accepts both `{"event": {"type": "...", ...}}` and flat `{"type": "...", ...}`
/// shapes.
pub fn parse_provider_stream_event(params: &Value) -> Result<ProviderStreamEvent, String> {
    let inner = match params.get("event") {
        Some(ev) => ev,
        None => params,
    };
    let obj = inner
        .as_object()
        .ok_or_else(|| "provider stream event must be a JSON object".to_string())?;

    let ty = obj
        .get("type")
        .and_then(Value::as_str)
        .ok_or_else(|| "provider stream event missing type".to_string())?;

    match ty {
        "text" => {
            let text = obj
                .get("delta")
                .or_else(|| obj.get("text"))
                .and_then(Value::as_str)
                .ok_or_else(|| {
                    "provider stream text event missing 'delta' or 'text'".to_string()
                })?;
            Ok(ProviderStreamEvent::TextDelta {
                text: text.to_string(),
            })
        }
        "thinking" => {
            let text = obj
                .get("delta")
                .or_else(|| obj.get("text"))
                .and_then(Value::as_str)
                .ok_or_else(|| {
                    "provider stream thinking event missing 'delta' or 'text'".to_string()
                })?;
            Ok(ProviderStreamEvent::ThinkingDelta {
                text: text.to_string(),
            })
        }
        "tool_use" => {
            let id = obj
                .get("id")
                .and_then(Value::as_str)
                .ok_or_else(|| "provider stream tool_use missing id".to_string())?;
            if id.is_empty() {
                return Err("provider stream tool_use id must be non-empty".to_string());
            }
            let name = obj
                .get("name")
                .and_then(Value::as_str)
                .ok_or_else(|| "provider stream tool_use missing name".to_string())?;
            if name.is_empty() {
                return Err("provider stream tool_use name must be non-empty".to_string());
            }
            let input = match obj.get("input") {
                None => Value::Object(Default::default()),
                Some(v) if v.is_object() => v.clone(),
                Some(_) => {
                    return Err("provider stream tool_use input must be a JSON object".to_string());
                }
            };
            Ok(ProviderStreamEvent::ToolUse {
                id: id.to_string(),
                name: name.to_string(),
                input,
            })
        }
        "usage" => {
            let mut clone = obj.clone();
            clone.remove("type");
            Ok(ProviderStreamEvent::Usage {
                usage: Value::Object(clone),
            })
        }
        "error" => {
            let message = obj
                .get("message")
                .and_then(Value::as_str)
                .ok_or_else(|| "provider stream error missing message".to_string())?;
            if message.is_empty() {
                return Err("provider stream error message must be non-empty".to_string());
            }
            Ok(ProviderStreamEvent::Error {
                message: message.to_string(),
            })
        }
        "done" => Ok(ProviderStreamEvent::Done),
        other => Err(format!("unknown provider stream event type: {other}")),
    }
}

pub async fn execute_provider_tool_use(
    registry: &crate::ToolRegistry,
    session_tools: &crate::tools::activation::SessionToolSet,
    hook_bus: &Arc<crate::extensions::hooks::HookBus>,
    tool_use: ProviderToolUse,
    ctx: crate::ToolContext,
    max_tool_output: usize,
) -> Value {
    let tool_id = tool_use.id;
    let tool_name = tool_use.name;
    let input = tool_use.input;

    // ═══ EXECUTION GATE (Task 16, spec §7.1) ═══
    // The extension-provider interior tool loop passes the SAME gate as
    // stream dispatch: resolve the provider-supplied wire name to the exact
    // live ToolId, verify the session set snapshot generation + pinned
    // schema digest, require core/exact-grant status, re-check source
    // trust, and only then acquire the implementation — all against the
    // one consistent registry borrow the caller holds (read-locked for the
    // whole outer provider call, so the catalog cannot drift inside the
    // loop). Denial happens BEFORE input translation, BEFORE hook
    // emission, BEFORE execution, and returns a valid tool_result with
    // typed bounded static content and `is_error: true`.
    let authorized = match crate::tools::activation::ExecutionGate::authorize_wire_call(
        registry,
        session_tools,
        &tool_name,
    ) {
        Ok(authorized) => authorized,
        Err(denial) => {
            return serde_json::json!({
                "type": "tool_result",
                "tool_use_id": tool_id,
                "content": denial.to_string(),
                "is_error": true,
            });
        }
    };
    let tool = authorized.implementation();

    let runtime_name = authorized.runtime_name().to_string();
    let input = registry.translate_input_for_api_tool(&tool_name, input);
    let decision = crate::runtime::resolve_before_tool_call_decision(
        input.clone(),
        crate::runtime::emit_before_tool_call(
            hook_bus,
            &tool_name,
            Some(&runtime_name),
            input.clone(),
            // Extension-originated tool.invoke: no owning conversation on this
            // path (ToolContext carries none) → `null`, as before.
            None,
        )
        .await,
        ctx.capabilities.secret_prompt.as_ref(),
        false,
        // The session's latch, so "Allow all" here arms (and honours) the
        // same per-extension consent as the main tool path.
        ctx.capabilities.session_allow_all.as_deref(),
    )
    .await;

    let crate::runtime::BeforeToolCallDecision::Continue { input } = decision else {
        let crate::runtime::BeforeToolCallDecision::Block { reason } = decision else {
            unreachable!()
        };
        return serde_json::json!({
            "type": "tool_result",
            "tool_use_id": tool_id,
            "content": format!("Tool call blocked by extension: {}", reason),
            "is_error": true,
        });
    };

    let input_for_hook = input.clone();
    let (result, is_error) = match tool.execute(input, ctx).await {
        Ok(output) => (output, false),
        Err(error) => (format!("Tool execution failed: {}", error), true),
    };
    let result = crate::runtime::emit_after_tool_call(
        hook_bus,
        &tool_name,
        Some(&runtime_name),
        input_for_hook,
        result,
        max_tool_output,
        None,
    )
    .await;

    let mut response = serde_json::json!({
        "type": "tool_result",
        "tool_use_id": tool_id,
        "content": crate::truncate_str(&result, max_tool_output).to_string(),
    });
    if is_error {
        response["is_error"] = serde_json::json!(true);
    }
    response
}

/// Interior extension-provider tool loop. `session_tools` is built ONCE by
/// the caller (under the same registry read guard passed as `registry`) and
/// reused for every tool call in every interior round — the registry stays
/// read-locked for the whole outer provider call, so no catalog drift is
/// possible inside this loop and one set snapshot governs all rounds.
#[allow(clippy::too_many_arguments)] // explicit loop policy inputs; grouping would hide registry/session coupling
pub async fn complete_provider_with_tools<F>(
    handler: Arc<dyn ExtensionHandler>,
    mut params: ProviderCompleteParams,
    registry: &crate::ToolRegistry,
    session_tools: &crate::tools::activation::SessionToolSet,
    hook_bus: &Arc<crate::extensions::hooks::HookBus>,
    mut context_factory: F,
    max_tool_output: usize,
    max_iterations: usize,
    // Out-param: total tool calls the provider requested across every
    // interior round. An out-param (rather than a return tuple) so the
    // count is still observable on the error paths below — the audit
    // record must stay honest when the iteration limit is hit.
    tools_requested: &mut u32,
) -> Result<ProviderCompleteResult, String>
where
    F: FnMut() -> crate::ToolContext,
{
    let max_iterations = max_iterations.max(1);
    *tools_requested = 0;
    for iteration in 0..max_iterations {
        let result = handler.provider_complete(params.clone()).await?;
        let tool_uses = extract_provider_tool_uses(&result.content)?;
        if tool_uses.is_empty() {
            return Ok(result);
        }
        // Counted before the limit check: the provider DID request these,
        // even when we refuse to run another round.
        *tools_requested = tools_requested.saturating_add(tool_uses.len() as u32);
        if iteration + 1 == max_iterations {
            return Err(format!(
                "extension provider '{}' exceeded provider tool-use iteration limit ({})",
                handler.id(),
                max_iterations,
            ));
        }

        let assistant_content = result.content.clone();
        params.messages.push(std::sync::Arc::new(serde_json::json!({
            "role": "assistant",
            "content": assistant_content,
        })));

        let mut tool_results = Vec::with_capacity(tool_uses.len());
        for tool_use in tool_uses {
            tool_results.push(
                execute_provider_tool_use(
                    registry,
                    session_tools,
                    hook_bus,
                    tool_use,
                    context_factory(),
                    max_tool_output,
                )
                .await,
            );
        }
        params.messages.push(std::sync::Arc::new(serde_json::json!({
            "role": "user",
            "content": tool_results,
        })));
    }
    Err(format!(
        "extension provider '{}' exceeded provider tool-use iteration limit ({})",
        handler.id(),
        max_iterations,
    ))
}

pub fn extract_provider_tool_uses(content: &[Value]) -> Result<Vec<ProviderToolUse>, String> {
    let mut tool_uses = Vec::new();
    for block in content {
        if block.get("type").and_then(Value::as_str) != Some("tool_use") {
            continue;
        }
        let id = block
            .get("id")
            .and_then(Value::as_str)
            .ok_or_else(|| "provider tool_use missing id".to_string())?;
        let name = block
            .get("name")
            .and_then(Value::as_str)
            .ok_or_else(|| "provider tool_use missing name".to_string())?;
        if id.trim().is_empty() {
            return Err("provider tool_use id is empty".to_string());
        }
        if name.trim().is_empty() {
            return Err("provider tool_use name is empty".to_string());
        }
        let input = block
            .get("input")
            .cloned()
            .unwrap_or_else(|| serde_json::json!({}));
        if !input.is_object() {
            return Err(format!(
                "provider tool_use '{}' input must be a JSON object",
                id
            ));
        }
        tool_uses.push(ProviderToolUse {
            id: id.to_string(),
            name: name.to_string(),
            input,
        });
    }
    Ok(tool_uses)
}

#[derive(Debug, Clone, Default, PartialEq)]
pub struct InitializeCapabilitiesResult {
    pub tools: Vec<RegisteredExtensionToolSpec>,
    pub providers: Vec<RegisteredProviderSpec>,
    /// Generic, plugin-defined capabilities. Each entry's `kind` is a
    /// free-form string the plugin author picks (e.g. `"capture"`,
    /// `"ocr"`, `"agent"`, `"foot_pedal"`); core does not enumerate or
    /// branch on it. Permissions are gated by the declared
    /// [`CapabilityDeclaration::permissions`] list.
    pub capabilities: Vec<CapabilityDeclaration>,
}

/// Generic capability declaration returned by an extension's
/// `initialize` response.
///
/// Core does not interpret `kind` or `params` — those are contracts
/// between the plugin and whatever consumer (chatui glue, another
/// plugin, an external watcher) cares about that capability kind. The
/// host only enforces that the plugin holds every permission it
/// declares it needs.
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
pub struct CapabilityDeclaration {
    /// Free-form capability kind tag chosen by the plugin author.
    /// Examples: `"capture"`, `"ocr"`, `"agent"`, `"clipboard_mirror"`.
    /// Must be non-empty; otherwise opaque to core.
    pub kind: String,
    /// Human-readable display name shown in `/extensions status` and
    /// similar surfaces. Must be non-empty.
    pub name: String,
    /// Permission names this capability needs at runtime, e.g.
    /// `["audio.input"]`. Every entry must parse to a known
    /// [`crate::extensions::permissions::Permission`] *and* be present
    /// in the extension's granted permission set.
    #[serde(default)]
    pub permissions: Vec<String>,
    /// Free-form metadata. Core stores and forwards but never
    /// interprets. Plugins use this for endpoint URLs, mode lists,
    /// version tags, anything that doesn't fit the typed slots.
    #[serde(default, skip_serializing_if = "is_null_value")]
    pub params: serde_json::Value,
}

fn is_null_value(v: &serde_json::Value) -> bool {
    v.is_null()
}

/// Validate a [`CapabilityDeclaration`] against the granted permission
/// set.
///
/// Rules — generic, no enumeration of capability kinds:
/// - `kind` must be non-empty.
/// - `name` must be non-empty.
/// - Every string in `permissions` must parse to a known
///   [`crate::extensions::permissions::Permission`].
/// - Every parsed permission must be present in `granted`.
pub fn validate_capability(
    decl: &CapabilityDeclaration,
    granted: &crate::extensions::permissions::PermissionSet,
) -> Result<(), String> {
    use crate::extensions::permissions::Permission;
    if decl.kind.trim().is_empty() {
        return Err("capability 'kind' must be non-empty".to_string());
    }
    if decl.name.trim().is_empty() {
        return Err("capability 'name' must be non-empty".to_string());
    }
    for perm_name in &decl.permissions {
        let parsed = Permission::parse(perm_name).ok_or_else(|| {
            format!(
                "capability '{}' declares unknown permission '{}'",
                decl.kind, perm_name
            )
        })?;
        if !granted.has(parsed) {
            return Err(format!(
                "capability '{}' requires permission '{}' but it is not granted",
                decl.kind, perm_name
            ));
        }
    }
    Ok(())
}

#[derive(Deserialize)]
struct InitializeResult {
    protocol_version: u32,
    #[serde(default)]
    capabilities: InitializeCapabilities,
}

#[derive(Default, Deserialize)]
struct InitializeCapabilities {
    #[serde(default)]
    tools: Vec<RegisteredExtensionToolSpec>,
    #[serde(default)]
    providers: Vec<RegisteredProviderSpec>,
    /// Canonical generic capability list.
    #[serde(default)]
    capabilities: Vec<CapabilityDeclaration>,
}

/// A JSON-RPC notification frame received from an extension (no `id`).
///
/// Internal API exposed publicly (`#[doc(hidden)]`) so integration tests
/// can subscribe to notifications via [`ProcessExtension::subscribe_notifications`].
/// Extension authors should not depend on this type — it may change without notice.
#[doc(hidden)]
#[derive(Debug, Clone)]
pub struct NotificationFrame {
    pub method: String,
    pub params: Value,
}

/// CP-11 fix-2 (B): bounded notification handoff. Combined with the 4 MiB
/// per-frame reader cap, queue capacity fixes the worst-case bytes parked
/// between the sidecar reader and a subscriber; beyond it the reader
/// AWAITS (backpressure onto the child's stdout pipe) instead of retaining
/// hostile flood frames in host memory.
pub const NOTIFICATION_QUEUE_CAPACITY: usize = 8;
/// CP-11 fix-2 (B): bounded provider stream-event handoff between the
/// `provider_stream` notification loop and the route forwarder.
pub const PROVIDER_EVENT_QUEUE_CAPACITY: usize = 8;

/// Shared mailbox for the background reader task. Holds in-flight request
/// senders (keyed by JSON-RPC id) and an optional notification subscriber.
///
/// Persists across process restarts: a new reader task replaces the old one
/// but the `Inbox` itself is reused. Pending requests are drained with
/// errors when the reader observes EOF or a transport failure.
struct Inbox {
    pending: Mutex<HashMap<u64, oneshot::Sender<Result<Value, String>>>>,
    /// Fan-out notification subscribers. Each subscriber is keyed by a monotonic
    /// id so it can unsubscribe itself without disturbing other concurrent
    /// subscribers (e.g. background watchers that overlap with
    /// `command.invoke` / `provider.stream`). Dead senders (receiver dropped)
    /// are pruned lazily on dispatch and on `unsubscribe_notifications`.
    /// BOUNDED ([`NOTIFICATION_QUEUE_CAPACITY`]): a subscriber that stops
    /// draining backpressures the reader instead of growing host memory.
    notification_sinks: Mutex<Vec<(usize, mpsc::Sender<NotificationFrame>)>>,
    /// Monotonic id allocator for `notification_sinks`. Wraparound is not a
    /// concern in practice — `usize` is enormous relative to subscription
    /// lifetimes — but uniqueness is only required among live subscribers.
    next_sink_id: std::sync::atomic::AtomicUsize,
    /// Set to true when the reader task exits (EOF or error). Used to prevent
    /// callers from registering pending requests that will never be fulfilled.
    closed: std::sync::atomic::AtomicBool,
    /// Child observation is separate from `ProcessExtension::state`: that lock
    /// serializes RPC I/O and may be held for an entire request. The lifecycle
    /// slot is held only to clone/publish a handle, never across process I/O.
    lifecycle: Mutex<Option<Arc<ProcessLifecycle>>>,
    /// Even generations identify live transports; the low bit permanently
    /// invalidates a transport. Allocation fails closed rather than wrapping.
    next_generation: AtomicU64,
    /// Permissions granted to the calling extension. Set after manifest
    /// validation; checked by inbound RPC handlers (e.g. memory.append).
    permissions: RwLock<Option<crate::extensions::permissions::PermissionSet>>,
    /// Stdin handle of the currently-running child process. Used by the
    /// reader task to write JSON-RPC responses for inbound requests.
    /// Replaced on each spawn (initial spawn + restarts).
    inbound_stdin: Mutex<Option<Arc<Mutex<ChildStdin>>>>,
    /// Extension id, used for namespace policy and diagnostics.
    extension_id: String,
    exclusive_memory: bool,
}

impl Inbox {
    fn new(extension_id: String) -> Self {
        Self {
            pending: Mutex::new(HashMap::new()),
            notification_sinks: Mutex::new(Vec::new()),
            next_sink_id: std::sync::atomic::AtomicUsize::new(0),
            closed: std::sync::atomic::AtomicBool::new(false),
            lifecycle: Mutex::new(None),
            next_generation: AtomicU64::new(2),
            permissions: RwLock::new(None),
            inbound_stdin: Mutex::new(None),
            exclusive_memory: crate::memory_backend::MemoryBinding::configured_current()
                .exclusive(),
            extension_id,
        }
    }

    async fn invalidate_lifecycle(&self) {
        if let Some(lifecycle) = self.lifecycle.lock().await.as_ref() {
            lifecycle.invalidate();
        }
    }

    /// Drains all pending request senders, sending `Err(reason)` to each.
    /// Also marks the inbox as closed so no new requests can be registered.
    async fn fail_all_pending(&self, reason: &str) {
        self.invalidate_lifecycle().await;
        self.closed
            .store(true, std::sync::atomic::Ordering::Release);
        let drained: Vec<_> = {
            let mut pending = self.pending.lock().await;
            pending.drain().collect()
        };
        for (_, tx) in drained {
            let _ = tx.send(Err(reason.to_string()));
        }
    }
}

/// One actual child/transport lifetime. Reader tasks retain this exact record,
/// so a late EOF from an old reader can never invalidate a replacement's token.
struct ProcessLifecycle {
    child: Mutex<Child>,
    generation: AtomicU64,
    initialized: AtomicBool,
}

impl ProcessLifecycle {
    fn invalidate(&self) {
        // Idempotent; no wraparound and no window that exposes the old token
        // as healthy while teardown or replacement is already underway.
        self.generation.fetch_or(1, Ordering::AcqRel);
    }

    fn snapshot(&self, recovered: bool) -> Option<ExtensionLifecycle> {
        if self.generation.load(Ordering::Acquire) & 1 == 0 {
            let mut child = self.child.try_lock().ok()?;
            match child.try_wait() {
                Ok(None) => {}
                Ok(Some(_)) | Err(_) => self.invalidate(),
            }
        }
        let generation = self.generation.load(Ordering::Acquire);
        let health = if generation & 1 != 0 {
            ExtensionHealth::Failed
        } else if !self.initialized.load(Ordering::Acquire) {
            ExtensionHealth::Loaded
        } else if recovered {
            ExtensionHealth::Degraded
        } else {
            ExtensionHealth::Running
        };
        Some(ExtensionLifecycle { generation, health })
    }
}

struct ProcessState {
    lifecycle: Arc<ProcessLifecycle>,
    stdin: Arc<Mutex<ChildStdin>>,
    reader_handle: JoinHandle<()>,
}

impl Drop for ProcessState {
    fn drop(&mut self) {
        self.lifecycle.invalidate();
        self.reader_handle.abort();
        // `kill_on_drop` alone is insufficient: the lifecycle observation slot
        // also owns the child. Request cancellation must stop it immediately.
        if let Ok(mut child) = self.lifecycle.child.try_lock() {
            let _ = child.start_kill();
        }
    }
}

/// A running extension process communicating via JSON-RPC 2.0 over stdio.
pub struct ProcessExtension {
    id: String,
    command: String,
    args: Vec<String>,
    cwd: Option<PathBuf>,
    state: Arc<Mutex<Option<ProcessState>>>,
    /// Serializes a full request/response exchange and restart attempts.
    call_lock: Arc<Mutex<()>>,
    next_id: AtomicU64,
    restart_count: AtomicUsize,
    /// Last resolved initialize config. Kept in memory only so a restarted
    /// extension receives the same manifest-resolved values (including secrets).
    /// Never written to disk or emitted in logs.
    initialize_config: Mutex<Value>,
    /// Lifetime restart total — never reset. `restart_count` tracks
    /// *consecutive* failures (reset on successful restart, H-8) for the
    /// exhaustion budget; this one feeds health/reporting so a previously
    /// crashed extension still reads Degraded instead of pretending it
    /// never crashed.
    total_restarts: AtomicUsize,
    /// Set by [`ExtensionHandler::shutdown`] (unload, reload, lease
    /// teardown), never cleared: a stopped handler must not respawn. A call
    /// authorized against it before the stop (a tool call batched with the
    /// `reload-extension` that replaces it) would otherwise take the
    /// empty-state restart path and run a second copy of the extension
    /// beside its replacement.
    stopped: AtomicBool,
    /// Restart policy controlling exponential backoff and budget.
    pub(crate) restart_policy: RestartPolicy,
    /// Shared mailbox between the reader task and request callers. Persists
    /// across process restarts so that any active notification subscriber
    /// survives a restart-on-error.
    inbox: Arc<Inbox>,
}

/// Prefix the reader task gives every JSON-RPC `error` response from the
/// extension. Such an error proves the process is alive and answering, so the
/// call-with-restart path must return it verbatim instead of restarting.
pub(crate) const APPLICATION_ERROR_PREFIX: &str = "Extension error: ";

/// Whether a call error is an application-level JSON-RPC error (the process
/// answered) rather than a transport failure (EOF, write error, dropped
/// response channel, timeout).
pub(crate) fn is_application_error(error: &str) -> bool {
    error.starts_with(APPLICATION_ERROR_PREFIX)
}

impl ProcessExtension {
    /// Abort the child transport immediately. Cancellation uses this rather
    /// than waiting for the per-call permit; draining pending responders
    /// releases every blocked call/producer.
    pub async fn force_shutdown(&self) {
        self.inbox.invalidate_lifecycle().await;
        let mut state_guard = self.state.lock().await;
        if let Some(state) = state_guard.take() {
            state.reader_handle.abort();
            state.lifecycle.invalidate();
            let _ = state.lifecycle.child.lock().await.kill().await;
        }
        drop(state_guard);
        self.inbox.closed.store(true, Ordering::Release);
        self.inbox.notification_sinks.lock().await.clear();
        self.inbox
            .fail_all_pending("transport closed: extension cancelled")
            .await;
    }

    pub async fn spawn(id: &str, command: &str, args: &[String]) -> Result<Self, String> {
        Self::spawn_with_cwd(id, command, args, None).await
    }

    /// Spawn `command` with `args` and optional working directory.
    ///
    /// Child stderr is captured and forwarded to debug tracing with the extension
    /// id so extension authors can inspect diagnostics without corrupting stdout.
    pub async fn spawn_with_cwd(
        id: &str,
        command: &str,
        args: &[String],
        cwd: Option<PathBuf>,
    ) -> Result<Self, String> {
        let exclusive = crate::memory_backend::MemoryBinding::configured_current().exclusive();
        Self::spawn_with_memory_policy(id, command, args, cwd, exclusive).await
    }

    pub(crate) async fn spawn_with_memory_policy(
        id: &str,
        command: &str,
        args: &[String],
        cwd: Option<PathBuf>,
        exclusive: bool,
    ) -> Result<Self, String> {
        if exclusive && id == "axel-memory-manager" {
            return Err(
                "independent Axel plugin is disabled by the host-selected memory backend".into(),
            );
        }
        let mut inbox = Inbox::new(id.to_string());
        inbox.exclusive_memory = exclusive;
        let inbox = Arc::new(inbox);
        let state = Self::spawn_state(id, command, args, cwd.as_ref(), inbox.clone()).await?;
        Ok(Self {
            id: id.to_string(),
            command: command.to_string(),
            args: args.to_vec(),
            cwd,
            state: Arc::new(Mutex::new(Some(state))),
            call_lock: Arc::new(Mutex::new(())),
            next_id: AtomicU64::new(1),
            restart_count: AtomicUsize::new(0),
            initialize_config: Mutex::new(Value::Object(Default::default())),
            total_restarts: AtomicUsize::new(0),
            stopped: AtomicBool::new(false),
            restart_policy: RestartPolicy::default(),
            inbox,
        })
    }

    /// Override the restart policy. Intended for tests.
    pub fn with_restart_policy(mut self, policy: RestartPolicy) -> Self {
        self.restart_policy = policy;
        self
    }

    async fn spawn_state(
        id: &str,
        command: &str,
        args: &[String],
        cwd: Option<&PathBuf>,
        inbox: Arc<Inbox>,
    ) -> Result<ProcessState, String> {
        let mut cmd = Command::new(command);
        cmd.args(args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        if let Some(cwd) = cwd {
            cmd.current_dir(cwd);
        }

        // Clear the inherited environment so extensions cannot read secrets from
        // the parent process (e.g. ANTHROPIC_API_KEY, SSH_AUTH_SOCK, AWS_*).
        // Only forward a minimal set of safe, non-sensitive variables.
        // TODO: forward the extension's declared `secret_env` values here once
        //       secret_env is resolved at spawn time rather than config time.
        cmd.env_clear();
        for var in &["PATH", "HOME", "LANG", "TERM", "XDG_RUNTIME_DIR"] {
            if let Ok(val) = std::env::var(var) {
                cmd.env(var, val);
            }
        }

        cmd.kill_on_drop(true);

        let mut child = cmd
            .spawn()
            .map_err(|e| format!("Failed to spawn extension '{}': {}", id, e))?;

        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| format!("No stdin for extension '{}'", id))?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| format!("No stdout for extension '{}'", id))?;
        if let Some(stderr) = child.stderr.take() {
            let extension_id = id.to_string();
            tokio::spawn(async move {
                // Bounded stderr forwarding (Task 20): extension-controlled
                // bytes. Each line is capped, total forwarded bytes are
                // budgeted; past the budget only byte counts are tracked so
                // a hostile child can never grow memory or flood the log.
                const MAX_STDERR_LINE: usize = 8 * 1024;
                const MAX_STDERR_FORWARD_TOTAL: u64 = 128 * 1024;
                let mut reader = BufReader::new(stderr);
                let mut forwarded: u64 = 0;
                let mut suppressed: u64 = 0;
                loop {
                    let mut buf: Vec<u8> = Vec::new();
                    let mut limited =
                        tokio::io::AsyncReadExt::take(&mut reader, MAX_STDERR_LINE as u64 + 1);
                    match tokio::io::AsyncBufReadExt::read_until(&mut limited, b'\n', &mut buf)
                        .await
                    {
                        Ok(0) => break,
                        Ok(n) => {
                            if forwarded < MAX_STDERR_FORWARD_TOTAL {
                                forwarded = forwarded.saturating_add(n as u64);
                                let truncated = n > MAX_STDERR_LINE;
                                let end = buf.len().min(MAX_STDERR_LINE);
                                let line = String::from_utf8_lossy(&buf[..end]);
                                let line = line.trim_end_matches(['\r', '\n']);
                                tracing::debug!(
                                    extension = %extension_id,
                                    truncated,
                                    stderr = %line,
                                );
                            } else {
                                suppressed = suppressed.saturating_add(n as u64);
                            }
                        }
                        Err(error) => {
                            tracing::debug!(
                                extension = %extension_id,
                                error = %error,
                                "Failed to read extension stderr",
                            );
                            break;
                        }
                    }
                }
                if suppressed > 0 {
                    tracing::debug!(
                        extension = %extension_id,
                        suppressed_bytes = suppressed,
                        "Extension stderr exceeded the forwarding budget; further bytes counted only",
                    );
                }
            });
        }

        let generation = inbox
            .next_generation
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| n.checked_add(2))
            .map_err(|_| "extension lifecycle generation exhausted".to_string())?;
        let lifecycle = Arc::new(ProcessLifecycle {
            child: Mutex::new(child),
            generation: AtomicU64::new(generation),
            initialized: AtomicBool::new(false),
        });
        {
            let mut slot = inbox.lifecycle.lock().await;
            if let Some(old) = slot.as_ref() {
                old.invalidate();
            }
            *slot = Some(lifecycle.clone());
        }
        // Reset BEFORE starting the reader: a fast EOF must never be erased.
        inbox.closed.store(false, Ordering::Release);
        let reader_handle =
            Self::spawn_reader(stdout, inbox.clone(), id.to_string(), lifecycle.clone());

        let stdin_arc = Arc::new(Mutex::new(stdin));
        // Publish current stdin into the inbox so the reader task can write
        // JSON-RPC responses for inbound requests (e.g. memory.append).
        *inbox.inbound_stdin.lock().await = Some(stdin_arc.clone());

        Ok(ProcessState {
            lifecycle,
            stdin: stdin_arc,
            reader_handle,
        })
    }

    /// Spawn the background reader task that owns `stdout`, demultiplexing
    /// JSON-RPC responses (by id) and notifications (no id) into the shared
    /// [`Inbox`]. Returns a `JoinHandle` so callers can `.abort()` it on
    /// restart or shutdown.
    fn spawn_reader(
        stdout: ChildStdout,
        inbox: Arc<Inbox>,
        extension_id: String,
        lifecycle: Arc<ProcessLifecycle>,
    ) -> JoinHandle<()> {
        tokio::spawn(async move {
            let mut reader = BufReader::new(stdout);
            loop {
                match Self::read_one_frame(&mut reader, &extension_id).await {
                    Ok(Some(value)) => {
                        Self::dispatch_frame(value, &inbox, &extension_id).await;
                    }
                    Ok(None) => {
                        lifecycle.invalidate();
                        tracing::debug!(
                            extension = %extension_id,
                            "Extension stdout closed (EOF); failing pending requests",
                        );
                        inbox.fail_all_pending("transport closed: EOF").await;
                        // Drop all notification subscribers on EOF.
                        inbox.notification_sinks.lock().await.clear();
                        return;
                    }
                    Err(error) => {
                        lifecycle.invalidate();
                        tracing::debug!(
                            extension = %extension_id,
                            error = %error,
                            "Extension transport read error",
                        );
                        inbox
                            .fail_all_pending(&format!("transport error: {}", error))
                            .await;
                        inbox.notification_sinks.lock().await.clear();
                        return;
                    }
                }
            }
        })
    }

    /// Read one Content-Length-framed JSON message from `reader`. Returns
    /// `Ok(None)` on a clean EOF *before* any header bytes are read; any
    /// other unexpected EOF is reported as `Err`.
    async fn read_one_frame(
        reader: &mut BufReader<ChildStdout>,
        extension_id: &str,
    ) -> Result<Option<Value>, String> {
        let mut content_length: Option<usize> = None;
        let mut saw_any_header = false;
        loop {
            let mut header_line = String::new();
            // Bounded header read: a hostile child streaming bytes without
            // a newline must fail the 1 KiB header policy, not grow memory.
            let n = {
                let mut limited = tokio::io::AsyncReadExt::take(&mut *reader, 1025);
                limited
                    .read_line(&mut header_line)
                    .await
                    .map_err(|e| format!("Read header error: {}", e))?
            };
            if n == 0 {
                if saw_any_header {
                    return Err("Unexpected EOF while reading response headers".into());
                }
                return Ok(None);
            }
            saw_any_header = true;
            if header_line.len() > 1024 {
                return Err(format!(
                    "Extension '{}' header line too long ({} bytes)",
                    extension_id,
                    header_line.len()
                ));
            }
            let trimmed = header_line.trim();
            if trimmed.is_empty() {
                break;
            }
            if let Some((name, value)) = trimmed.split_once(':') {
                if name.trim().eq_ignore_ascii_case("Content-Length") {
                    content_length = Some(value.trim().parse().map_err(|_| {
                        format!("Invalid Content-Length value: {:?}", value.trim())
                    })?);
                }
            }
        }
        let content_length = content_length.ok_or_else(|| {
            format!(
                "Extension '{}' frame missing Content-Length header",
                extension_id
            )
        })?;
        const MAX_RESPONSE_SIZE: usize = 4 * 1024 * 1024;
        if content_length > MAX_RESPONSE_SIZE {
            return Err(format!(
                "Extension '{}' frame too large: {} bytes (max {})",
                extension_id, content_length, MAX_RESPONSE_SIZE
            ));
        }
        let mut buf = vec![0u8; content_length];
        tokio::io::AsyncReadExt::read_exact(reader, &mut buf)
            .await
            .map_err(|e| format!("Read body error: {}", e))?;
        let value: Value =
            serde_json::from_slice(&buf).map_err(|e| format!("Parse frame error: {}", e))?;
        Ok(Some(value))
    }

    /// Route a parsed JSON-RPC frame to the right consumer:
    /// - response (`id` numeric, no `method`) → matching pending oneshot
    /// - request (`id` numeric and `method`) → inbound request handler
    /// - notification (`method` set, no `id`) → notification subscriber
    /// - anything else → trace and drop
    async fn dispatch_frame(value: Value, inbox: &Arc<Inbox>, extension_id: &str) {
        let id_field = value.get("id");
        let id_is_present = !matches!(id_field, None | Some(Value::Null));
        let method_field = value
            .get("method")
            .and_then(Value::as_str)
            .map(str::to_string);

        if id_is_present && method_field.is_some() {
            // Inbound request from the extension. Spawn a task to handle it
            // so the reader loop is never blocked on memory I/O or other work.
            // JSON-RPC 2.0 §4: `id` MUST be a String, Number, or Null. The
            // extension picks its own request ids and is the only party that
            // correlates them, so echo whatever it sent back verbatim rather
            // than forcing it through u64.
            //
            // This previously required a numeric id and dropped everything
            // else at trace level. `docs/extensions/protocol.md` shows string
            // ids in its own examples, so an extension written against the
            // documentation hung forever with no diagnostic. (#291 defect 3)
            let id = match id_field {
                Some(id @ (Value::String(_) | Value::Number(_))) => id.clone(),
                _ => {
                    tracing::warn!(
                        extension = %extension_id,
                        frame = %value,
                        "Discarding inbound request: JSON-RPC id must be a string or a number",
                    );
                    return;
                }
            };
            let Some(method) = method_field else { return };
            let params = value.get("params").cloned().unwrap_or(Value::Null);
            let inbox = inbox.clone();
            let extension_id = extension_id.to_string();
            tokio::spawn(async move {
                let outcome = Self::handle_inbound_request(&inbox, &method, params).await;
                let payload = match outcome {
                    Ok(result) => serde_json::json!({
                        "jsonrpc": "2.0",
                        "id": id,
                        "result": result,
                    }),
                    Err((code, message)) => serde_json::json!({
                        "jsonrpc": "2.0",
                        "id": id,
                        "error": {"code": code, "message": message},
                    }),
                };
                let stdin_handle = inbox.inbound_stdin.lock().await.clone();
                if let Some(stdin) = stdin_handle {
                    let body = match serde_json::to_string(&payload) {
                        Ok(s) => s,
                        Err(error) => {
                            tracing::warn!(
                                extension = %extension_id,
                                error = %error,
                                "Failed to serialize inbound response",
                            );
                            return;
                        }
                    };
                    let frame = format!("Content-Length: {}\r\n\r\n{}", body.len(), body);
                    let mut stdin = stdin.lock().await;
                    if let Err(error) = stdin.write_all(frame.as_bytes()).await {
                        inbox.invalidate_lifecycle().await;
                        tracing::warn!(
                            extension = %extension_id,
                            error = %error,
                            "Failed to write inbound response",
                        );
                        return;
                    }
                    if let Err(error) = stdin.flush().await {
                        inbox.invalidate_lifecycle().await;
                        tracing::warn!(
                            extension = %extension_id,
                            error = %error,
                            "Failed to flush inbound response",
                        );
                    }
                } else {
                    tracing::warn!(
                        extension = %extension_id,
                        "No stdin available to reply to inbound request",
                    );
                }
            });
            return;
        }

        if id_is_present {
            // Response frames are different: the id must be one WE issued, and
            // the runtime only ever issues numeric ids. A non-numeric id here
            // cannot correlate to anything, so it is genuinely a protocol
            // error on the extension's side — say so at warn, not trace.
            let id = match id_field.and_then(Value::as_u64) {
                Some(id) => id,
                None => {
                    tracing::warn!(
                        extension = %extension_id,
                        frame = %value,
                        "Discarding response frame: id does not match any request \
                         the runtime issued (runtime request ids are always numeric)",
                    );
                    return;
                }
            };
            let sender = inbox.pending.lock().await.remove(&id);
            match sender {
                Some(tx) => {
                    let payload = if let Some(err) = value.get("error") {
                        let message = err
                            .get("message")
                            .and_then(Value::as_str)
                            .unwrap_or("unknown extension error")
                            .to_string();
                        Err(format!("{APPLICATION_ERROR_PREFIX}{}", message))
                    } else {
                        Ok(value.get("result").cloned().unwrap_or(Value::Null))
                    };
                    let _ = tx.send(payload);
                }
                None => {
                    tracing::trace!(
                        extension = %extension_id,
                        id = id,
                        "Response with unknown id (no pending request); dropping",
                    );
                }
            }
        } else if let Some(method) = value.get("method").and_then(Value::as_str) {
            let params = value.get("params").cloned().unwrap_or(Value::Null);
            let frame = NotificationFrame {
                method: method.to_string(),
                params,
            };
            // Snapshot the live subscribers WITHOUT holding the lock across
            // the awaited sends below — a subscriber unsubscribing while we
            // backpressure must not deadlock against this dispatch.
            let targets: Vec<(usize, mpsc::Sender<NotificationFrame>)> = {
                let sinks = inbox.notification_sinks.lock().await;
                sinks.to_vec()
            };
            if targets.is_empty() {
                tracing::trace!(
                    extension = %extension_id,
                    method = %method,
                    "Notification with no active subscribers; dropping",
                );
            } else {
                // Bounded fan-out: an un-draining subscriber BACKPRESSURES
                // this reader (and, through the stdout pipe, the child)
                // instead of growing host memory. `send` only errors when
                // the receiver is gone, so failure == dead subscriber.
                let mut dead = Vec::new();
                for (id, tx) in &targets {
                    if tx.send(frame.clone()).await.is_err() {
                        dead.push(*id);
                    }
                }
                if !dead.is_empty() {
                    let mut sinks = inbox.notification_sinks.lock().await;
                    sinks.retain(|(id, _)| !dead.contains(id));
                }
            }
        } else {
            tracing::trace!(
                extension = %extension_id,
                frame = %value,
                "Unrecognized frame; dropping",
            );
        }
    }

    pub fn restart_count(&self) -> usize {
        // Lifetime total — reporting/tests expect total restarts, not the
        // consecutive-failure budget counter (which resets on success).
        self.total_restarts.load(Ordering::Relaxed)
    }

    /// Public for tests: set the permission set used by inbound RPC handlers
    /// (e.g. memory.append). Called by the manager after manifest validation.
    pub async fn set_permissions(&self, perms: crate::extensions::permissions::PermissionSet) {
        *self.inbox.permissions.write().await = Some(perms);
    }

    /// Handle a JSON-RPC request initiated by the extension.
    ///
    /// Returns `Ok(result_value)` on success or `Err((code, message))` for a
    /// JSON-RPC error response. Currently routes:
    /// - `memory.append` (requires `memory.write`)
    /// - `memory.query`  (requires `memory.read`)
    ///   All other methods return -32601 (method not found).
    #[allow(clippy::doc_lazy_continuation)]
    async fn handle_inbound_request(
        inbox: &Arc<Inbox>,
        method: &str,
        params: Value,
    ) -> Result<Value, (i32, String)> {
        use crate::extensions::permissions::Permission;
        use crate::memory::store::{self, MemoryQuery};

        if inbox.exclusive_memory && matches!(method, "memory.append" | "memory.query") {
            return Err((
                -32000,
                "legacy extension memory is disabled by the host-selected backend".into(),
            ));
        }

        match method {
            "memory.append" => {
                Self::require_permission(inbox, Permission::MemoryWrite, "memory.write").await?;
                let namespace = Self::param_str(&params, "namespace")?;
                Self::require_namespace_matches(inbox, &namespace).await?;
                let content = Self::param_str(&params, "content")?;
                let tags = match params.get("tags") {
                    None | Some(Value::Null) => Vec::new(),
                    Some(Value::Array(arr)) => {
                        let mut out = Vec::with_capacity(arr.len());
                        for v in arr {
                            match v.as_str() {
                                Some(s) => out.push(s.to_string()),
                                None => {
                                    return Err((
                                        -32602,
                                        "tags must be an array of strings".to_string(),
                                    ))
                                }
                            }
                        }
                        out
                    }
                    _ => return Err((-32602, "tags must be an array of strings".to_string())),
                };
                let meta = match params.get("meta") {
                    None | Some(Value::Null) => None,
                    Some(v) => Some(v.clone()),
                };
                let record = store::new_record(namespace, content, tags, meta);
                let timestamp_ms = record.timestamp_ms;
                store::append(&record).map_err(|e| (-32000, e.to_string()))?;
                Ok(serde_json::json!({"ok": true, "timestamp_ms": timestamp_ms}))
            }
            "memory.query" => {
                Self::require_permission(inbox, Permission::MemoryRead, "memory.read").await?;
                let namespace = Self::param_str(&params, "namespace")?;
                Self::require_namespace_matches(inbox, &namespace).await?;
                let q = MemoryQuery {
                    content_contains: params
                        .get("content_contains")
                        .and_then(Value::as_str)
                        .map(str::to_string),
                    tag_prefix: params
                        .get("tag_prefix")
                        .and_then(Value::as_str)
                        .map(str::to_string),
                    since_ms: params.get("since_ms").and_then(Value::as_u64),
                    until_ms: params.get("until_ms").and_then(Value::as_u64),
                    limit: params
                        .get("limit")
                        .and_then(Value::as_u64)
                        .map(|n| n as usize),
                };
                let records = store::query(&namespace, &q).map_err(|e| (-32000, e.to_string()))?;
                Ok(serde_json::json!({"records": records}))
            }
            "config.get" => {
                let key = Self::param_str(&params, "key")?;
                Self::validate_config_key(&key)?;
                let value =
                    crate::extensions::config_store::read_plugin_config(&inbox.extension_id, &key);
                Ok(serde_json::json!({"value": value}))
            }
            "config.set" => {
                Self::require_permission(inbox, Permission::ConfigWrite, "config.write").await?;
                let key = Self::param_str(&params, "key")?;
                Self::validate_config_key(&key)?;
                let value = Self::param_str(&params, "value")?;
                crate::extensions::config_store::write_plugin_config(
                    &inbox.extension_id,
                    &key,
                    &value,
                )
                .map_err(|e| (-32000, e.to_string()))?;
                Ok(serde_json::json!({"ok": true}))
            }
            "config.subscribe" => {
                Self::require_permission(inbox, Permission::ConfigSubscribe, "config.subscribe")
                    .await?;
                // The long-lived watcher-to-notification bridge is wired at the manager/UI layer.
                // This phase exposes the authorized protocol ACK so plugins can opt in without
                // blocking initialize; direct store watchers are unit-tested in config_store.
                Ok(serde_json::json!({"ok": true}))
            }
            other => Err((-32601, format!("method not found: {other}"))),
        }
    }

    async fn require_permission(
        inbox: &Arc<Inbox>,
        perm: crate::extensions::permissions::Permission,
        wire: &str,
    ) -> Result<(), (i32, String)> {
        let guard = inbox.permissions.read().await;
        match guard.as_ref() {
            Some(set) if set.has(perm) => Ok(()),
            _ => Err((-32602, format!("permission denied: {wire} required"))),
        }
    }

    async fn require_namespace_matches(
        inbox: &Arc<Inbox>,
        namespace: &str,
    ) -> Result<(), (i32, String)> {
        if namespace == inbox.extension_id {
            Ok(())
        } else {
            Err((
                -32602,
                format!(
                    "namespace must equal extension id '{}' (got '{}')",
                    inbox.extension_id, namespace
                ),
            ))
        }
    }

    fn param_str(params: &Value, name: &str) -> Result<String, (i32, String)> {
        params
            .get(name)
            .and_then(Value::as_str)
            .map(str::to_string)
            .ok_or_else(|| (-32602, format!("missing or invalid '{name}' parameter")))
    }

    fn validate_config_key(key: &str) -> Result<(), (i32, String)> {
        let trimmed = key.trim();
        if trimmed.is_empty() {
            return Err((-32602, "config key must be non-empty".to_string()));
        }
        if trimmed.contains('.') || trimmed.contains('/') || trimmed.contains(' ') {
            return Err((
                -32602,
                "config key must not contain dots, slashes, or spaces".to_string(),
            ));
        }
        Ok(())
    }

    pub async fn initialize(
        &self,
        plugin_root: Option<PathBuf>,
        config: Value,
    ) -> Result<InitializeCapabilitiesResult, String> {
        // Retain only in memory: restart handshakes must receive the same
        // resolved config, while secrets never enter env, logs, or storage.
        *self.initialize_config.lock().await = config.clone();
        let params = InitializeParams {
            synaps_version: env!("CARGO_PKG_VERSION"),
            extension_protocol_version: CURRENT_EXTENSION_PROTOCOL_VERSION,
            plugin_id: self.id.clone(),
            plugin_root: plugin_root
                .or_else(|| self.cwd.clone())
                .map(|path| path.to_string_lossy().to_string()),
            config,
        };
        let value = self
            .call_no_restart(
                "initialize",
                serde_json::to_value(params).map_err(|e| e.to_string())?,
            )
            .await?;
        Self::parse_initialize_result(&self.id, value)
    }

    fn parse_initialize_result(
        id: &str,
        value: Value,
    ) -> Result<InitializeCapabilitiesResult, String> {
        let result: InitializeResult = serde_json::from_value(value)
            .map_err(|e| format!("Invalid initialize response from extension '{}': {}", id, e))?;
        if result.protocol_version != CURRENT_EXTENSION_PROTOCOL_VERSION {
            return Err(format!(
                "Extension '{}' initialize returned unsupported protocol_version {} (supported: {})",
                id, result.protocol_version, CURRENT_EXTENSION_PROTOCOL_VERSION,
            ));
        }
        Self::validate_registered_tool_specs(id, &result.capabilities.tools)?;
        Self::validate_registered_provider_specs(id, &result.capabilities.providers)?;
        Ok(InitializeCapabilitiesResult {
            tools: result.capabilities.tools,
            providers: result.capabilities.providers,
            capabilities: result.capabilities.capabilities,
        })
    }

    fn validate_registered_tool_specs(
        id: &str,
        tools: &[RegisteredExtensionToolSpec],
    ) -> Result<(), String> {
        use crate::extensions::validation::{validate_id_segment, IdValidationError};
        let mut names = HashSet::new();
        for tool in tools {
            let name = tool.name.trim();
            if let Err(err) = validate_id_segment(name) {
                return Err(match err {
                    IdValidationError::Empty => format!(
                        "Extension '{}' registered a tool with an empty tool name",
                        id
                    ),
                    IdValidationError::ContainsReserved { ch } => format!(
                        "Extension '{}' registered tool '{}' with invalid tool name: '{}' is reserved",
                        id, name, ch
                    ),
                    IdValidationError::TooLong { len, max } => format!(
                        "Extension '{}' registered tool '{}' with invalid tool name: must be at most {} chars (got {})",
                        id, name, max, len
                    ),
                    IdValidationError::ContainsWhitespace => format!(
                        "Extension '{}' registered tool '{}' with invalid tool name: must not contain whitespace",
                        id, name
                    ),
                    IdValidationError::ContainsControl { ch } => format!(
                        "Extension '{}' registered tool '{}' with invalid tool name: contains control character U+{:04X}",
                        id, name, ch as u32
                    ),
                });
            }
            if !names.insert(name.to_string()) {
                return Err(format!(
                    "Extension '{}' registered duplicate tool name '{}'",
                    id, name
                ));
            }
            if tool.description.trim().is_empty() {
                return Err(format!(
                    "Extension '{}' registered tool '{}' with an empty description",
                    id, name,
                ));
            }
            if !tool.input_schema.is_object() {
                return Err(format!(
                    "Extension '{}' registered tool '{}' with invalid input_schema: input_schema must be a JSON object",
                    id, name,
                ));
            }
        }
        Ok(())
    }

    fn validate_registered_provider_specs(
        id: &str,
        providers: &[RegisteredProviderSpec],
    ) -> Result<(), String> {
        use crate::extensions::validation::{validate_id_segment, IdValidationError};
        for provider in providers {
            let provider_id = provider.id.trim();
            match validate_id_segment(provider_id) {
                Ok(()) => {
                    if !Self::is_safe_provider_id(provider_id) {
                        return Err(format!(
                            "Extension '{}' registered provider '{}' with invalid provider id",
                            id, provider_id
                        ));
                    }
                }
                Err(IdValidationError::Empty) => {
                    return Err(format!(
                        "Extension '{}' registered provider with empty provider id",
                        id
                    ));
                }
                Err(err) => {
                    return Err(format!(
                        "Extension '{}' registered provider '{}' with invalid provider id: {}",
                        id, provider_id, err
                    ));
                }
            }
            if provider.display_name.trim().is_empty() {
                return Err(format!(
                    "Extension '{}' registered provider '{}' with empty display_name",
                    id, provider_id,
                ));
            }
            if provider.description.trim().is_empty() {
                return Err(format!(
                    "Extension '{}' registered provider '{}' with empty description",
                    id, provider_id,
                ));
            }
            if provider.models.is_empty() {
                return Err(format!(
                    "Extension '{}' registered provider '{}' must declare at least one model",
                    id, provider_id,
                ));
            }
            let mut model_ids = HashSet::new();
            for model in &provider.models {
                let model_id = model.id.trim();
                if let Err(err) = validate_id_segment(model_id) {
                    return Err(match err {
                        IdValidationError::Empty => format!(
                            "Extension '{}' registered provider '{}' with empty model id",
                            id, provider_id
                        ),
                        IdValidationError::ContainsReserved { ch } => format!(
                            "Extension '{}' registered provider '{}' with invalid model id '{}': '{}' is reserved",
                            id, provider_id, model_id, ch
                        ),
                        IdValidationError::TooLong { len, max } => format!(
                            "Extension '{}' registered provider '{}' with invalid model id '{}': must be at most {} chars (got {})",
                            id, provider_id, model_id, max, len
                        ),
                        IdValidationError::ContainsWhitespace => format!(
                            "Extension '{}' registered provider '{}' with invalid model id '{}': must not contain whitespace",
                            id, provider_id, model_id
                        ),
                        IdValidationError::ContainsControl { ch } => format!(
                            "Extension '{}' registered provider '{}' with invalid model id '{}': contains control character U+{:04X}",
                            id, provider_id, model_id, ch as u32
                        ),
                    });
                }
                if !model_ids.insert(model_id.to_string()) {
                    return Err(format!(
                        "Extension '{}' registered provider '{}' with duplicate model id '{}'",
                        id, provider_id, model_id,
                    ));
                }
            }
            if let Some(config_schema) = &provider.config_schema {
                if !config_schema.is_object() {
                    return Err(format!(
                        "Extension '{}' registered provider '{}' with invalid config_schema: config_schema must be a JSON object",
                        id, provider_id,
                    ));
                }
            }
        }
        Ok(())
    }

    fn is_safe_provider_id(id: &str) -> bool {
        !id.is_empty()
            && !id.contains(':')
            && id
                .chars()
                .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-' || c == '_')
    }

    #[doc(hidden)]
    pub async fn initialize_for_test(&self, plugin_root: Option<PathBuf>) -> Result<(), String> {
        self.initialize(plugin_root, Value::Object(Default::default()))
            .await
            .map(|_| ())
    }

    /// A `hook.handle` call's outcome as the handler's decision: a
    /// malformed `modify` blocks, every other failure (bad result, transport
    /// error, timeout) is `Continue`.
    fn hook_result(
        &self,
        outcome: Result<Result<Value, String>, tokio::time::error::Elapsed>,
    ) -> HookResult {
        match outcome {
            Ok(Ok(value)) => match serde_json::from_value(value.clone()) {
                Ok(result) => result,
                Err(error) => {
                    tracing::warn!(
                        extension = %self.id,
                        error = %error,
                        response = %value,
                        "Extension hook handler returned invalid result",
                    );
                    match value.get("action").and_then(Value::as_str) {
                        // A malformed `modify` is dangerous — it would let an
                        // unknown/partial input through. Fail safe by blocking.
                        Some("modify") => HookResult::Block {
                            reason: "Extension returned malformed modify result".to_string(),
                        },
                        // A malformed `replace` is safe to ignore: the original
                        // tool output is preserved by emit_after_tool_call.
                        // Explicit arm so this intent can't be silently changed
                        // by a future edit to the catch-all below.
                        Some("replace") => HookResult::Continue,
                        _ => HookResult::Continue,
                    }
                }
            },
            Ok(Err(e)) => {
                tracing::warn!(
                    extension = %self.id,
                    error = %e,
                    "Extension hook handler failed — continuing",
                );
                HookResult::Continue
            }
            Err(_) => {
                tracing::warn!(
                    extension = %self.id,
                    timeout_secs = 5,
                    "Extension hook handler timed out — continuing",
                );
                HookResult::Continue
            }
        }
    }

    async fn restart_locked(&self, state: &mut Option<ProcessState>) -> Result<(), String> {
        if self.stopped.load(Ordering::Acquire) {
            return Err(format!(
                "Extension '{}' was stopped (unloaded or reloaded); not respawning it",
                self.id,
            ));
        }
        self.inbox.invalidate_lifecycle().await;
        let attempted = self.restart_count.fetch_add(1, Ordering::Relaxed) + 1;
        self.total_restarts.fetch_add(1, Ordering::Relaxed);
        let max_attempts = self.restart_policy.max_attempts;
        if attempted > max_attempts as usize {
            *state = None;
            return Err(format!(
                "Extension '{}' exceeded restart limit ({})",
                self.id, max_attempts,
            ));
        }

        if let Some(old) = state.take() {
            old.reader_handle.abort();
            old.lifecycle.invalidate();
            let _ = old.lifecycle.child.lock().await.kill().await;
        }
        // Drain any stale pending entries before reusing the inbox.
        self.inbox
            .fail_all_pending("transport closed: process restarting")
            .await;

        let delay = self
            .restart_policy
            .delay_for_attempt(attempted as u32)
            .unwrap_or_default();

        tracing::warn!(
            extension = %self.id,
            attempt = attempted,
            max_attempts = max_attempts,
            delay_ms = delay.as_millis() as u64,
            "Restarting extension process after transport failure",
        );

        if !delay.is_zero() {
            tokio::time::sleep(delay).await;
        }

        *state = Some(
            Self::spawn_state(
                &self.id,
                &self.command,
                &self.args,
                self.cwd.as_ref(),
                self.inbox.clone(),
            )
            .await?,
        );
        self.initialize_locked(state).await?;
        // NOTE: the consecutive-failure counter is NOT reset here. A
        // successful handshake isn't proof of recovery — a crash-looping
        // extension can initialize fine and die on the next request
        // (always_exit fixture). The reset happens in call_inner once the
        // restarted process actually serves a call (preserves H-8 intent:
        // transient failures hours apart don't accumulate toward disable).
        Ok(())
    }

    async fn initialize_locked(&self, state: &mut Option<ProcessState>) -> Result<(), String> {
        let config = self.initialize_config.lock().await.clone();
        let params = InitializeParams {
            synaps_version: env!("CARGO_PKG_VERSION"),
            extension_protocol_version: CURRENT_EXTENSION_PROTOCOL_VERSION,
            plugin_id: self.id.clone(),
            plugin_root: self
                .cwd
                .clone()
                .map(|path| path.to_string_lossy().to_string()),
            config,
        };
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let value = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            self.call_once_locked(
                state.as_mut().expect("state should exist for initialize"),
                "initialize",
                serde_json::to_value(params).map_err(|e| e.to_string())?,
                id,
            ),
        )
        .await
        .map_err(|_| format!("Extension '{}' initialize timed out after 10s", self.id))??;
        let result = Self::parse_initialize_result(&self.id, value).map(|_| ());
        if let Some(state) = state.as_ref() {
            if result.is_ok() {
                state.lifecycle.initialized.store(true, Ordering::Release);
            } else {
                state.lifecycle.invalidate();
            }
        }
        result
    }

    /// Send a single JSON-RPC request and await the matching response,
    /// using the shared inbox for response delivery. The reader task runs
    /// concurrently and may dispatch interleaved notifications.
    async fn call_once_locked(
        &self,
        state: &mut ProcessState,
        method: &str,
        params: Value,
        id: u64,
    ) -> Result<Value, String> {
        let body = serde_json::to_string(&JsonRpcRequest {
            jsonrpc: "2.0",
            method: method.to_string(),
            params,
            id,
        })
        .map_err(|e| format!("Serialize error: {}", e))?;

        let (tx, rx) = oneshot::channel::<Result<Value, String>>();
        // Check if inbox is closed before registering — if the reader already
        // exited, no one will ever send a response on this channel.
        if self.inbox.closed.load(std::sync::atomic::Ordering::Acquire) {
            return Err("transport closed: inbox is shut down".to_string());
        }

        // Register pending BEFORE writing so the reader can route a fast
        // response without racing against the insert.
        self.inbox.pending.lock().await.insert(id, tx);

        // Double-check: if closed was set between our check and the insert,
        // remove the entry and fail immediately.
        if self.inbox.closed.load(std::sync::atomic::Ordering::Acquire) {
            self.inbox.pending.lock().await.remove(&id);
            return Err("transport closed: inbox shut down during registration".to_string());
        }

        let frame = format!("Content-Length: {}\r\n\r\n{}", body.len(), body);
        let write_result = {
            let mut stdin = state.stdin.lock().await;
            match stdin.write_all(frame.as_bytes()).await {
                Ok(()) => stdin.flush().await,
                Err(e) => Err(e),
            }
        };
        if let Err(e) = write_result {
            state.lifecycle.invalidate();
            // Make sure we don't leak the pending entry if the write fails.
            self.inbox.pending.lock().await.remove(&id);
            return Err(format!("Write error: {}", e));
        }

        match rx.await {
            Ok(payload) => payload,
            Err(_) => {
                state.lifecycle.invalidate();
                // Sender was dropped without sending — typically because the
                // reader task observed EOF/error after we registered. The
                // reader normally sends an Err first; this branch is a
                // belt-and-braces fallback.
                self.inbox.pending.lock().await.remove(&id);
                Err("transport closed: response channel dropped".to_string())
            }
        }
    }

    async fn call_no_restart(&self, method: &str, params: Value) -> Result<Value, String> {
        let _call_guard = self.call_lock.lock().await;
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let mut state_guard = self.state.lock().await;
        if state_guard.is_none() {
            *state_guard = Some(
                Self::spawn_state(
                    &self.id,
                    &self.command,
                    &self.args,
                    self.cwd.as_ref(),
                    self.inbox.clone(),
                )
                .await?,
            );
        }
        let state = state_guard.as_mut().expect("state should exist");
        let result = self.call_once_locked(state, method, params, id).await;
        if method == "initialize" {
            // Publish readiness while still serialized with replacement; never
            // accidentally mark a newer child ready using an older response.
            match result.as_ref() {
                Ok(value) if Self::parse_initialize_result(&self.id, value.clone()).is_ok() => {
                    state.lifecycle.initialized.store(true, Ordering::Release);
                }
                _ => state.lifecycle.invalidate(),
            }
        }
        result
    }

    /// Send one request only if the current child is still present. Teardown
    /// must never resurrect a child that concurrent cancellation already
    /// removed, so unlike [`Self::call`] this bypasses all spawn/restart paths.
    async fn call_no_spawn(&self, method: &str, params: Value) -> Result<Value, String> {
        let _call_guard = self.call_lock.lock().await;
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let mut state_guard = self.state.lock().await;
        let Some(state) = state_guard.as_mut() else {
            return Err(format!(
                "Extension '{}' process is not running; '{}' refused (no respawn)",
                self.id, method
            ));
        };
        self.call_once_locked(state, method, params, id).await
    }

    /// Task 20 lease path: one `tool.call` that NEVER (re)spawns. A missing
    /// process state fails typed instead of resurrecting a terminated
    /// child, and the restart machinery is bypassed entirely — a lease
    /// child that dies is a transport error, never a silent new process
    /// whose declarations were not re-validated.
    pub(crate) async fn call_tool_once_no_spawn(
        &self,
        name: &str,
        input: Value,
        session_id: Option<&str>,
    ) -> Result<Value, String> {
        let fut = async {
            let _call_guard = self.call_lock.lock().await;
            let id = self.next_id.fetch_add(1, Ordering::Relaxed);
            // Take the running state out while awaiting the response so a
            // concurrent cancellation can observe an empty slot and still
            // fail the shared inbox pending map. The state is restored only
            // if cancellation did not close the inbox.
            let mut state = {
                let mut state_guard = self.state.lock().await;
                state_guard.take().ok_or_else(|| {
                    format!(
                        "Extension '{}' process is not running; leased call refused (no respawn)",
                        self.id
                    )
                })?
            };
            let result = self
                .call_once_locked(
                    &mut state,
                    "tool.call",
                    tool_call_params(name, input, session_id),
                    id,
                )
                .await;
            if !self.inbox.closed.load(std::sync::atomic::Ordering::Acquire) {
                *self.state.lock().await = Some(state);
            } else {
                state.reader_handle.abort();
                state.lifecycle.invalidate();
                let _ = state.lifecycle.child.lock().await.kill().await;
            }
            result
        };
        match tokio::time::timeout(std::time::Duration::from_secs(120), fut).await {
            Ok(result) => result,
            Err(_) => Err(format!(
                "Extension '{}' leased tool.call timed out after 120s",
                self.id
            )),
        }
    }

    async fn call(&self, method: &str, params: Value) -> Result<Value, String> {
        let timeout_secs = if method == "tool.call" { 120 } else { 30 };
        let id_str = self.id.clone();
        let method_str = method.to_string();

        let result = tokio::time::timeout(
            std::time::Duration::from_secs(timeout_secs),
            self.call_inner(method, params),
        )
        .await;

        match result {
            Ok(inner) => inner,
            Err(_) => Err(format!(
                "Extension '{}' method '{}' timed out after {}s",
                id_str, method_str, timeout_secs
            )),
        }
    }

    async fn call_inner(&self, method: &str, params: Value) -> Result<Value, String> {
        let _call_guard = self.call_lock.lock().await;
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let mut state_guard = self.state.lock().await;
        if state_guard.is_none() {
            self.restart_locked(&mut state_guard).await?;
        }

        let result = self
            .call_once_locked(
                state_guard
                    .as_mut()
                    .expect("state should exist after restart"),
                method,
                params.clone(),
                id,
            )
            .await;

        match result {
            Ok(value) => {
                // Process served a real call — it has genuinely recovered.
                // Reset the consecutive-failure budget (H-8) so transient
                // crashes hours apart don't accumulate toward disable.
                self.restart_count.store(0, Ordering::Relaxed);
                Ok(value)
            }
            Err(first_error) if is_application_error(&first_error) => {
                // The process answered with a JSON-RPC `error` object
                // (unknown method, invalid params, a tool refusing its
                // input). That is a live, healthy transport — restarting
                // would only re-spawn the process to hear the same answer
                // again, so it also counts as a real recovery.
                self.restart_count.store(0, Ordering::Relaxed);
                Err(first_error)
            }
            Err(first_error) => {
                self.restart_locked(&mut state_guard).await?;
                let retry_id = self.next_id.fetch_add(1, Ordering::Relaxed);
                self.call_once_locked(
                    state_guard
                        .as_mut()
                        .expect("state should exist after restart"),
                    method,
                    params,
                    retry_id,
                )
                .await
                .inspect(|_| {
                    // Retry succeeded post-restart — also a real recovery.
                    self.restart_count.store(0, Ordering::Relaxed);
                })
                .map_err(|retry_error| {
                    format!(
                        "{}; retry after restart failed: {}",
                        first_error, retry_error
                    )
                })
            }
        }
    }

    /// Subscribe to JSON-RPC notifications emitted by the extension.
    ///
    /// Returns `(subscription_id, receiver)`. The id must be passed back to
    /// [`unsubscribe_notifications`] to cancel this subscription specifically
    /// — concurrent subscribers (e.g. background widget watchers running
    /// alongside `command.invoke` / `provider.stream`) coexist and each
    /// receives every notification frame.
    ///
    /// The receiver yields every notification frame (no `id`, has `method`)
    /// the extension sends until either:
    /// - the receiver is dropped (its slot is pruned on the next dispatch),
    /// - [`unsubscribe_notifications`] is called with this id, or
    /// - the reader observes EOF or a transport error (all subscribers are
    ///   dropped together).
    ///
    /// Internal API: exposed publicly with `#[doc(hidden)]` only so
    /// integration tests can exercise the bidirectional transport.
    #[doc(hidden)]
    pub async fn subscribe_notifications(&self) -> (usize, mpsc::Receiver<NotificationFrame>) {
        let (tx, rx) = mpsc::channel(NOTIFICATION_QUEUE_CAPACITY);
        let id = self
            .inbox
            .next_sink_id
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let mut sinks = self.inbox.notification_sinks.lock().await;
        // Opportunistically prune any subscribers whose receivers have been
        // dropped, so the Vec doesn't grow unboundedly across long-lived
        // extensions if callers forget to unsubscribe.
        sinks.retain(|(_, tx)| !tx.is_closed());
        sinks.push((id, tx));
        (id, rx)
    }

    /// Drop the notification subscription with the given id, if any.
    /// Unknown ids are silently ignored (idempotent — safe to call on
    /// belt-and-braces timeout paths even after the inner future already
    /// unsubscribed).
    #[doc(hidden)]
    pub async fn unsubscribe_notifications(&self, id: usize) {
        let mut sinks = self.inbox.notification_sinks.lock().await;
        sinks.retain(|(sub_id, tx)| *sub_id != id && !tx.is_closed());
    }

    /// Forward one notification frame received during `command.invoke`.
    ///
    /// Routes:
    /// - `command.output` whose `request_id` matches → sink as `Output(event)`
    ///   (other request_ids are ignored — concurrent invocations must use distinct ids
    ///   and may overlap in the same `subscribe_notifications` channel; mismatched
    ///   frames are dropped here intentionally).
    /// - `task.start|update|log|done` → sink as `Task(event)` regardless of request_id.
    /// - Anything else → logged at trace and dropped.
    ///
    /// CP-11 fix-3: the sink is BOUNDED
    /// ([`crate::extensions::invoke_output::InvokeEventSink`]) and its
    /// send is AWAITED — a hostile `command.output` flood backpressures
    /// this loop (and transitively the bounded notification queue and the
    /// sidecar's stdout pipe) instead of parking aggregate bytes in an
    /// unbounded queue. Liveness relies on the eagerly concurrent
    /// collector on the other end (see `invoke_output` module docs); if
    /// that consumer is gone, the send errs and `sink_open` latches false.
    pub(crate) async fn forward_invoke_command_frame(
        extension_id: &str,
        request_id: &str,
        sink: &crate::extensions::invoke_output::InvokeEventSink,
        sink_open: &mut bool,
        frame: NotificationFrame,
    ) -> bool {
        use crate::extensions::commands::parse_command_output;
        use crate::extensions::runtime::InvokeCommandEvent;
        use crate::extensions::tasks::{is_task_method, parse_task_event};

        let mut saw_done = false;
        if frame.method == "command.output" {
            match parse_command_output(&frame.params) {
                Ok(parsed) if parsed.request_id == request_id => {
                    if matches!(
                        parsed.event,
                        crate::extensions::commands::CommandOutputEvent::Done
                    ) {
                        saw_done = true;
                    }
                    if *sink_open
                        && sink
                            .send(InvokeCommandEvent::Output(parsed.event))
                            .await
                            .is_err()
                    {
                        *sink_open = false;
                    }
                }
                Ok(_) => {
                    // Different request_id — ignore (logged at trace for diagnostics).
                    tracing::trace!(
                        extension = %extension_id,
                        "Ignoring command.output for unrelated request_id",
                    );
                }
                Err(error) => {
                    tracing::warn!(
                        extension = %extension_id,
                        error = %error,
                        params = %frame.params,
                        "Skipping malformed command.output notification",
                    );
                }
            }
        } else if is_task_method(&frame.method) {
            match parse_task_event(&frame.method, &frame.params) {
                Ok(event) => {
                    if *sink_open && sink.send(InvokeCommandEvent::Task(event)).await.is_err() {
                        *sink_open = false;
                    }
                }
                Err(error) => {
                    tracing::warn!(
                        extension = %extension_id,
                        method = %frame.method,
                        error = %error,
                        params = %frame.params,
                        "Skipping malformed task notification",
                    );
                }
            }
        } else {
            tracing::trace!(
                extension = %extension_id,
                method = %frame.method,
                "Ignoring non-command/task notification during command.invoke",
            );
        }
        saw_done
    }

    /// Forward one notification frame received during `provider.stream`.
    ///
    /// - Frames whose method is not `provider.stream.event` are ignored (logged at trace).
    /// - Malformed event params are logged at warn and skipped (do not abort the call).
    /// - The sink is BOUNDED ([`PROVIDER_EVENT_QUEUE_CAPACITY`]): a slow
    ///   downstream backpressures this loop (and transitively the sidecar)
    ///   instead of retaining hostile flood events in host memory.
    /// - If the caller's sink has been closed, sets `sink_open = false` and stops forwarding,
    ///   but the in-flight request is still allowed to complete.
    async fn forward_provider_stream_frame(
        extension_id: &str,
        sink: &mpsc::Sender<ProviderStreamEvent>,
        sink_open: &mut bool,
        frame: NotificationFrame,
    ) {
        if frame.method != "provider.stream.event" {
            tracing::trace!(
                extension = %extension_id,
                method = %frame.method,
                "Ignoring non-stream notification during provider.stream",
            );
            return;
        }
        match parse_provider_stream_event(&frame.params) {
            Ok(event) => {
                if *sink_open && sink.send(event).await.is_err() {
                    *sink_open = false;
                }
            }
            Err(error) => {
                tracing::warn!(
                    extension = %extension_id,
                    error = %error,
                    params = %frame.params,
                    "Skipping malformed provider.stream.event notification",
                );
            }
        }
    }
}

/// `tool.call` params. `session_id` is the owning conversation id, `null`
/// for workers/no conversation and under the `SYNAPS_HOOK_SESSION_ID=0`
/// kill switch, so plugins can key tool state on the same value the hook
/// events carry. Always present (`null`, never omitted), like hook fields.
pub(crate) fn tool_call_params(name: &str, input: Value, session_id: Option<&str>) -> Value {
    tool_call_params_gated(
        name,
        input,
        session_id,
        crate::extensions::hooks::events::hook_session_id_enabled(),
    )
}

fn tool_call_params_gated(
    name: &str,
    input: Value,
    session_id: Option<&str>,
    session_id_enabled: bool,
) -> Value {
    let session_id = if session_id_enabled { session_id } else { None };
    serde_json::json!({
        "name": name,
        "input": input,
        "session_id": session_id,
    })
}

#[async_trait]
impl ExtensionHandler for ProcessExtension {
    fn id(&self) -> &str {
        &self.id
    }

    fn lifecycle_snapshot(&self) -> Option<ExtensionLifecycle> {
        let lifecycle = self.inbox.lifecycle.try_lock().ok()?.clone()?;
        lifecycle.snapshot(self.total_restarts.load(Ordering::Relaxed) > 0)
    }

    async fn call_tool(&self, name: &str, input: Value) -> Result<Value, String> {
        self.call_tool_for(name, input, None).await
    }

    async fn call_tool_for(
        &self,
        name: &str,
        input: Value,
        session_id: Option<&str>,
    ) -> Result<Value, String> {
        self.call("tool.call", tool_call_params(name, input, session_id))
            .await
    }

    async fn provider_complete(
        &self,
        params: ProviderCompleteParams,
    ) -> Result<ProviderCompleteResult, String> {
        let value = tokio::time::timeout(
            std::time::Duration::from_secs(60),
            self.call(
                "provider.complete",
                serde_json::to_value(params).map_err(|e| e.to_string())?,
            ),
        )
        .await
        .map_err(|_| format!("Extension '{}' provider.complete timed out", self.id))??;
        let result: ProviderCompleteResult = serde_json::from_value(value).map_err(|e| {
            format!(
                "Invalid provider.complete response from extension '{}': {}",
                self.id, e
            )
        })?;
        if result.content.is_empty() {
            return Err(format!(
                "Extension '{}' provider.complete returned empty content",
                self.id
            ));
        }
        Ok(result)
    }

    async fn provider_stream(
        &self,
        params: ProviderCompleteParams,
        sink: tokio::sync::mpsc::Sender<ProviderStreamEvent>,
    ) -> Result<ProviderCompleteResult, String> {
        // Subscribe BEFORE issuing the request so we don't miss early
        // notifications that may arrive before `call(...)` even starts polling.
        let (sub_id, mut rx) = self.subscribe_notifications().await;
        let params_value = serde_json::to_value(params).map_err(|e| e.to_string())?;

        let extension_id = self.id.clone();
        let stream_future = async {
            let mut call_fut = Box::pin(self.call("provider.stream", params_value));
            let mut sink_open = true;
            let response = loop {
                tokio::select! {
                    response = &mut call_fut => break response,
                    Some(frame) = rx.recv() => {
                        Self::forward_provider_stream_frame(
                            &extension_id, &sink, &mut sink_open, frame,
                        ).await;
                    }
                }
            };
            // Response received: drop our subscription so this receiver yields
            // `None` once buffered frames are drained, then flush any
            // remaining notifications before returning. Other concurrent
            // subscribers (if any) are untouched.
            self.unsubscribe_notifications(sub_id).await;
            while let Some(frame) = rx.recv().await {
                Self::forward_provider_stream_frame(&extension_id, &sink, &mut sink_open, frame)
                    .await;
            }
            response
        };

        let outcome = tokio::time::timeout(std::time::Duration::from_secs(60), stream_future).await;

        // Belt-and-braces: ensure our subscription is cleared on timeout too.
        // Idempotent if the inner future already unsubscribed.
        self.unsubscribe_notifications(sub_id).await;

        let value =
            outcome.map_err(|_| format!("Extension '{}' provider.stream timed out", self.id))??;

        let result: ProviderCompleteResult = serde_json::from_value(value).map_err(|e| {
            format!(
                "Invalid provider.stream response from extension '{}': {}",
                self.id, e
            )
        })?;
        // NOTE: empty `content` is permitted for streaming — output may have
        // been delivered entirely via TextDelta notifications.
        Ok(result)
    }

    async fn invoke_command(
        &self,
        command: &str,
        args: Vec<String>,
        request_id: &str,
        sink: crate::extensions::invoke_output::InvokeEventSink,
    ) -> Result<Value, String> {
        // Subscribe before issuing the request so we don't miss early events.
        let (sub_id, mut rx) = self.subscribe_notifications().await;
        let params = serde_json::json!({
            "command": command,
            "args": args,
            "request_id": request_id,
        });

        let extension_id = self.id.clone();
        let request_id_owned = request_id.to_string();
        let invoke_future = async {
            let mut call_fut = Box::pin(self.call("command.invoke", params));
            let mut sink_open = true;
            let response = loop {
                tokio::select! {
                    response = &mut call_fut => break response,
                    Some(frame) = rx.recv() => {
                        // The awaited bounded send paces a flood; while it
                        // waits, `call_fut` is simply not polled — safe
                        // because the eagerly concurrent collector always
                        // drains (see invoke_output module docs), and if it
                        // is gone the send errs immediately.
                        let _ = Self::forward_invoke_command_frame(
                            &extension_id, &request_id_owned, &sink, &mut sink_open, frame,
                        ).await;
                    }
                }
            };
            // Drain any notifications already buffered after the response lands, but
            // do not wait for the subscriber channel to close (that would deadlock
            // while this invocation still owns `rx`).
            self.unsubscribe_notifications(sub_id).await;
            while let Ok(frame) = rx.try_recv() {
                let _ = Self::forward_invoke_command_frame(
                    &extension_id,
                    &request_id_owned,
                    &sink,
                    &mut sink_open,
                    frame,
                )
                .await;
            }
            response
        };

        let outcome =
            tokio::time::timeout(std::time::Duration::from_secs(120), invoke_future).await;

        // Belt-and-braces: ensure our subscription is cleared on timeout too.
        // Idempotent if the inner future already unsubscribed. (On timeout
        // the dropped `invoke_future` also drops `sink` and `rx`, releasing
        // the collector and any dispatcher blocked on this subscription.)
        self.unsubscribe_notifications(sub_id).await;

        outcome.map_err(|_| format!("Extension '{}' command.invoke timed out", self.id))?
    }

    async fn handle(&self, event: &HookEvent) -> HookResult {
        let params = serde_json::to_value(event).unwrap_or(Value::Null);
        self.hook_result(
            tokio::time::timeout(
                std::time::Duration::from_secs(5),
                self.call("hook.handle", params),
            )
            .await,
        )
    }

    /// The worker path: one `hook.handle` on the running child only. No
    /// spawn, no restart (`call_no_spawn`): a missing child or a dead
    /// transport is a failed call, so `Continue`.
    async fn handle_no_spawn(&self, event: &HookEvent) -> HookResult {
        let params = serde_json::to_value(event).unwrap_or(Value::Null);
        self.hook_result(
            tokio::time::timeout(
                std::time::Duration::from_secs(5),
                self.call_no_spawn("hook.handle", params),
            )
            .await,
        )
    }

    async fn get_info(&self) -> Result<crate::extensions::info::PluginInfo, String> {
        let value = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            self.call("info.get", Value::Null),
        )
        .await
        .map_err(|_| format!("Extension '{}' info.get timed out", self.id))??;
        serde_json::from_value(value).map_err(|e| {
            format!(
                "Invalid info.get response from extension '{}': {}",
                self.id, e
            )
        })
    }

    async fn sidecar_spawn_args(&self) -> Result<crate::sidecar::spawn::SidecarSpawnArgs, String> {
        let value = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            self.call("sidecar.spawn_args", Value::Null),
        )
        .await
        .map_err(|_| format!("Extension '{}' sidecar.spawn_args timed out", self.id))??;
        serde_json::from_value(value).map_err(|e| {
            format!(
                "Invalid sidecar.spawn_args response from extension '{}': {}",
                self.id, e
            )
        })
    }

    async fn settings_editor_open(&self, category: &str, field: &str) -> Result<Value, String> {
        let params = crate::extensions::settings_editor::SettingsEditorOpenParams {
            category: category.to_string(),
            field: field.to_string(),
        };
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            self.call(
                "settings.editor.open",
                serde_json::to_value(params).map_err(|e| e.to_string())?,
            ),
        )
        .await
        .map_err(|_| format!("Extension '{}' settings.editor.open timed out", self.id))?
    }

    async fn settings_editor_key(
        &self,
        category: &str,
        field: &str,
        key: &str,
    ) -> Result<Value, String> {
        let mut params = serde_json::to_value(
            crate::extensions::settings_editor::SettingsEditorKeyParams {
                key: key.to_string(),
            },
        )
        .map_err(|e| e.to_string())?;
        if let Some(obj) = params.as_object_mut() {
            obj.insert("category".to_string(), Value::String(category.to_string()));
            obj.insert("field".to_string(), Value::String(field.to_string()));
        }
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            self.call("settings.editor.key", params),
        )
        .await
        .map_err(|_| format!("Extension '{}' settings.editor.key timed out", self.id))?
    }

    async fn settings_editor_commit(
        &self,
        category: &str,
        field: &str,
        value: Value,
    ) -> Result<Value, String> {
        let params = serde_json::json!({
            "category": category,
            "field": field,
            "value": value,
        });
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            self.call("settings.editor.commit", params),
        )
        .await
        .map_err(|_| format!("Extension '{}' settings.editor.commit timed out", self.id))?
    }

    async fn shutdown(&self) {
        // Terminal before anything else: a call already past authorization
        // must find this handler stopped, not respawn it (`restart_locked`).
        self.stopped.store(true, Ordering::Release);
        self.inbox.invalidate_lifecycle().await;
        // Lease teardown is terminal for this ProcessExtension. In
        // particular, a concurrent force_shutdown may already have removed
        // the child; sending shutdown through the normal restart-capable
        // call path would then spawn an unvalidated replacement solely to
        // shut it down.
        let _ = tokio::time::timeout(
            std::time::Duration::from_millis(500),
            self.call_no_spawn("shutdown", Value::Null),
        )
        .await;

        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        let mut state_guard = self.state.lock().await;
        if let Some(state) = state_guard.take() {
            state.reader_handle.abort();
            state.lifecycle.invalidate();
            let _ = state.lifecycle.child.lock().await.kill().await;
        }
        // Drop all active notification subscribers and signal pending callers.
        self.inbox.notification_sinks.lock().await.clear();
        self.inbox
            .fail_all_pending("transport closed: extension shutdown")
            .await;
    }

    async fn subscribe_notifications(
        &self,
    ) -> (usize, tokio::sync::mpsc::Receiver<NotificationFrame>) {
        ProcessExtension::subscribe_notifications(self).await
    }

    async fn restart_count(&self) -> usize {
        self.restart_count()
    }

    async fn health(&self) -> ExtensionHealth {
        // Exhaustion is judged on the consecutive-failure counter (resets on
        // successful restart); Degraded is judged on lifetime restarts so a
        // recovered-but-previously-crashed extension doesn't report Running.
        let consecutive = self.restart_count.load(Ordering::Relaxed);
        let lifetime = self.total_restarts.load(Ordering::Relaxed);
        let max = self.restart_policy.max_attempts as usize;
        if consecutive >= max {
            ExtensionHealth::Failed
        } else if lifetime > 0 {
            // Within budget. If the state slot is currently empty, we're
            // mid-restart; otherwise the process is alive but has previously
            // crashed at least once.
            let state_alive = self.state.try_lock().map(|g| g.is_some()).unwrap_or(true);
            if state_alive {
                ExtensionHealth::Degraded
            } else {
                ExtensionHealth::Restarting
            }
        } else {
            ExtensionHealth::Running
        }
    }
}

#[cfg(test)]
mod exclusive_memory_tests {
    use super::*;
    #[tokio::test]
    async fn exclusive_inbox_refuses_legacy_memory_without_writes() {
        let mut inbox = Inbox::new("fixture".into());
        inbox.exclusive_memory = true;
        let inbox = Arc::new(inbox);
        for method in ["memory.append", "memory.query"] {
            let result = ProcessExtension::handle_inbound_request(
                &inbox,
                method,
                serde_json::json!({"namespace":"fixture","content":"must not persist"}),
            )
            .await;
            assert!(result.unwrap_err().1.contains("disabled"));
        }
    }
}

#[cfg(test)]
mod tool_call_params_tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn carries_caller_session_id() {
        assert_eq!(
            tool_call_params_gated("echo", json!({"text": "hi"}), Some("sess-A"), true),
            json!({"name": "echo", "input": {"text": "hi"}, "session_id": "sess-A"})
        );
    }

    #[test]
    fn worker_call_has_null_session_id_key() {
        let params = tool_call_params_gated("echo", json!({}), None, true);
        assert_eq!(params.get("session_id"), Some(&Value::Null));
    }

    #[test]
    fn kill_switch_forces_null() {
        let params = tool_call_params_gated("echo", json!({}), Some("sess-A"), false);
        assert_eq!(params.get("session_id"), Some(&Value::Null));
        assert_eq!(params["name"], json!("echo"));
    }
}

#[cfg(test)]
mod stream_event_tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn parses_text_delta_with_delta_key() {
        let v = json!({"type": "text", "delta": "hi"});
        assert_eq!(
            parse_provider_stream_event(&v).unwrap(),
            ProviderStreamEvent::TextDelta { text: "hi".into() }
        );
    }

    #[test]
    fn parses_text_delta_with_text_key() {
        let v = json!({"type": "text", "text": "hi"});
        assert_eq!(
            parse_provider_stream_event(&v).unwrap(),
            ProviderStreamEvent::TextDelta { text: "hi".into() }
        );
    }

    #[test]
    fn parses_thinking_delta() {
        let v = json!({"type": "thinking", "delta": "hmm"});
        assert_eq!(
            parse_provider_stream_event(&v).unwrap(),
            ProviderStreamEvent::ThinkingDelta { text: "hmm".into() }
        );
        let v2 = json!({"type": "thinking", "text": "hmm"});
        assert_eq!(
            parse_provider_stream_event(&v2).unwrap(),
            ProviderStreamEvent::ThinkingDelta { text: "hmm".into() }
        );
    }

    #[test]
    fn parses_tool_use() {
        let v = json!({
            "type": "tool_use",
            "id": "t1",
            "name": "echo",
            "input": {"x": 1}
        });
        assert_eq!(
            parse_provider_stream_event(&v).unwrap(),
            ProviderStreamEvent::ToolUse {
                id: "t1".into(),
                name: "echo".into(),
                input: json!({"x": 1}),
            }
        );
    }

    #[test]
    fn tool_use_input_defaults_to_empty_object() {
        let v = json!({"type": "tool_use", "id": "t1", "name": "echo"});
        assert_eq!(
            parse_provider_stream_event(&v).unwrap(),
            ProviderStreamEvent::ToolUse {
                id: "t1".into(),
                name: "echo".into(),
                input: json!({}),
            }
        );
    }

    #[test]
    fn parses_usage_strips_type() {
        let v = json!({"type": "usage", "input_tokens": 5, "output_tokens": 7});
        assert_eq!(
            parse_provider_stream_event(&v).unwrap(),
            ProviderStreamEvent::Usage {
                usage: json!({"input_tokens": 5, "output_tokens": 7})
            }
        );
    }

    #[test]
    fn parses_error() {
        let v = json!({"type": "error", "message": "boom"});
        assert_eq!(
            parse_provider_stream_event(&v).unwrap(),
            ProviderStreamEvent::Error {
                message: "boom".into()
            }
        );
    }

    #[test]
    fn parses_done() {
        let v = json!({"type": "done"});
        assert_eq!(
            parse_provider_stream_event(&v).unwrap(),
            ProviderStreamEvent::Done
        );
    }

    #[test]
    fn nested_event_shape_matches_flat() {
        let flat = json!({"type": "text", "delta": "hi"});
        let nested = json!({"event": {"type": "text", "delta": "hi"}});
        assert_eq!(
            parse_provider_stream_event(&flat).unwrap(),
            parse_provider_stream_event(&nested).unwrap()
        );
    }

    #[test]
    fn missing_type_errors() {
        let v = json!({"delta": "hi"});
        let err = parse_provider_stream_event(&v).unwrap_err();
        assert!(err.contains("missing type"), "got: {err}");
    }

    #[test]
    fn unknown_type_errors_with_type() {
        let v = json!({"type": "wat"});
        let err = parse_provider_stream_event(&v).unwrap_err();
        assert!(err.contains("wat"), "got: {err}");
    }

    #[test]
    fn tool_use_missing_id_errors() {
        let v = json!({"type": "tool_use", "name": "echo"});
        let err = parse_provider_stream_event(&v).unwrap_err();
        assert!(err.contains("id"), "got: {err}");
    }

    #[test]
    fn tool_use_missing_name_errors() {
        let v = json!({"type": "tool_use", "id": "t1"});
        let err = parse_provider_stream_event(&v).unwrap_err();
        assert!(err.contains("name"), "got: {err}");
    }

    #[test]
    fn tool_use_empty_id_errors() {
        let v = json!({"type": "tool_use", "id": "", "name": "echo"});
        assert!(parse_provider_stream_event(&v).is_err());
    }

    #[test]
    fn tool_use_empty_name_errors() {
        let v = json!({"type": "tool_use", "id": "t1", "name": ""});
        assert!(parse_provider_stream_event(&v).is_err());
    }

    #[test]
    fn tool_use_non_object_input_errors() {
        let v = json!({"type": "tool_use", "id": "t1", "name": "echo", "input": "nope"});
        let err = parse_provider_stream_event(&v).unwrap_err();
        assert!(err.contains("input"), "got: {err}");
    }

    #[test]
    fn text_missing_delta_and_text_errors() {
        let v = json!({"type": "text"});
        let err = parse_provider_stream_event(&v).unwrap_err();
        assert!(err.contains("delta") || err.contains("text"), "got: {err}");
    }

    #[test]
    fn error_missing_message_errors() {
        let v = json!({"type": "error"});
        assert!(parse_provider_stream_event(&v).is_err());
    }

    #[test]
    fn error_empty_message_errors() {
        let v = json!({"type": "error", "message": ""});
        assert!(parse_provider_stream_event(&v).is_err());
    }
}

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

    #[tokio::test]
    async fn restart_policy_default_max_attempts_is_3() {
        // Use a command that won't actually run; we only need the struct.
        // Since `spawn` actually launches the process, use a trivial echoer
        // and immediately shut it down. Falls back to /bin/cat which reads
        // from stdin and stays alive.
        let ext = ProcessExtension::spawn("policy-test", "/bin/cat", &[])
            .await
            .expect("spawn /bin/cat");
        assert_eq!(ext.restart_policy.max_attempts, 3);
        ext.shutdown().await;
    }

    #[tokio::test]
    async fn with_restart_policy_overrides_default() {
        let ext = ProcessExtension::spawn("policy-test-override", "/bin/cat", &[])
            .await
            .expect("spawn /bin/cat");
        let custom = RestartPolicy {
            max_attempts: 7,
            ..RestartPolicy::default()
        };
        let ext = ext.with_restart_policy(custom);
        assert_eq!(ext.restart_policy.max_attempts, 7);
        ext.shutdown().await;
    }
}

#[cfg(test)]
mod capture_validator_tests {
    use super::*;
    use crate::extensions::permissions::{Permission, PermissionSet};

    fn perms_with(grants: &[Permission]) -> PermissionSet {
        let mut p = PermissionSet::new();
        for g in grants {
            p.grant(*g);
        }
        p
    }

    fn cap(kind: &str, name: &str, perms: &[&str]) -> CapabilityDeclaration {
        CapabilityDeclaration {
            kind: kind.to_string(),
            name: name.to_string(),
            permissions: perms.iter().map(|p| p.to_string()).collect(),
            params: serde_json::Value::Null,
        }
    }

    #[test]
    fn capability_validator_rejects_empty_kind() {
        let d = cap("   ", "Sample", &["audio.input"]);
        let perms = perms_with(&[Permission::AudioInput]);
        let err = validate_capability(&d, &perms).unwrap_err();
        assert!(err.contains("kind"), "got: {}", err);
    }

    #[test]
    fn capability_validator_rejects_empty_name() {
        let d = cap("capture", "   ", &["audio.input"]);
        let perms = perms_with(&[Permission::AudioInput]);
        let err = validate_capability(&d, &perms).unwrap_err();
        assert!(err.contains("name"), "got: {}", err);
    }

    #[test]
    fn capability_validator_rejects_unknown_permission_string() {
        let d = cap("capture", "Sample", &["audio.telepathy"]);
        let perms = perms_with(&[Permission::AudioInput, Permission::AudioOutput]);
        let err = validate_capability(&d, &perms).unwrap_err();
        assert!(
            err.contains("unknown permission") && err.contains("audio.telepathy"),
            "got: {}",
            err,
        );
    }

    #[test]
    fn capability_validator_requires_every_declared_permission() {
        let d = cap("capture", "Sample", &["audio.input"]);
        let perms = perms_with(&[]);
        let err = validate_capability(&d, &perms).unwrap_err();
        assert!(
            err.contains("audio.input") && err.contains("not granted"),
            "got: {}",
            err,
        );
    }

    #[test]
    fn capability_validator_accepts_when_all_permissions_granted() {
        let d = cap("capture", "Sample", &["audio.input", "audio.output"]);
        let perms = perms_with(&[Permission::AudioInput, Permission::AudioOutput]);
        validate_capability(&d, &perms).expect("should validate");
    }

    #[test]
    fn capability_validator_accepts_no_permissions() {
        // Capabilities that genuinely don't need any permission (e.g. a
        // pure-CPU OCR sidecar that takes images via stdin) should
        // validate cleanly with an empty permission list.
        let d = cap("ocr", "Tesseract", &[]);
        let perms = perms_with(&[]);
        validate_capability(&d, &perms).expect("should validate");
    }

    #[test]
    fn capability_validator_does_not_branch_on_kind() {
        // Two capabilities with identical permission requirements but
        // different `kind` values both validate identically — proves
        // the validator does not enumerate kinds.
        let perms = perms_with(&[Permission::AudioInput]);
        for kind in ["capture", "ocr", "agent", "foot_pedal", "eeg"] {
            let d = cap(kind, "Anything", &["audio.input"]);
            validate_capability(&d, &perms).expect("should validate");
        }
    }
}

#[cfg(test)]
mod invoke_command_dispatch_tests {
    //! Phase B Phase 2/3 — exercise the notification dispatcher used by
    //! `command.invoke`. Spawning a real subprocess is not required: we feed
    //! `NotificationFrame`s through the production async forwarder into the
    //! bounded CP-11 fix-3 sink and assert the collected ordering.
    use super::*;
    use crate::extensions::commands::CommandOutputEvent;
    use crate::extensions::invoke_output::{invoke_event_channel, InvokeOutputBudget};
    use crate::extensions::runtime::InvokeCommandEvent;
    use crate::extensions::tasks::{TaskEvent, TaskKind};
    use serde_json::json;

    fn frame(method: &str, params: serde_json::Value) -> NotificationFrame {
        NotificationFrame {
            method: method.to_string(),
            params,
        }
    }

    /// Feed `frames` through the production forwarder into the bounded
    /// sink (the small frame counts here never exceed channel capacity,
    /// so sequential awaited sends cannot park), then collect. Returns
    /// the retained events and whether a Done marker was observed.
    async fn dispatch(frames: Vec<NotificationFrame>) -> (Vec<InvokeCommandEvent>, bool) {
        let (tx, collector) = invoke_event_channel(InvokeOutputBudget::default());
        let mut open = true;
        let mut saw_done = false;
        for f in frames {
            saw_done |=
                ProcessExtension::forward_invoke_command_frame("ext-test", "r1", &tx, &mut open, f)
                    .await;
        }
        drop(tx);
        let report = collector.collect().await;
        (report.events, saw_done)
    }

    #[tokio::test]
    async fn forwards_mixed_event_stream_in_order() {
        let frames = vec![
            frame(
                "command.output",
                json!({"request_id":"r1","event":{"kind":"text","content":"A"}}),
            ),
            frame(
                "task.start",
                json!({"id":"dl","label":"Downloading","kind":"download"}),
            ),
            frame("task.update", json!({"id":"dl","current":50,"total":100})),
            frame(
                "command.output",
                json!({"request_id":"r1","event":{"kind":"system","content":"working"}}),
            ),
            frame("task.done", json!({"id":"dl"})),
            frame(
                "command.output",
                json!({"request_id":"r1","event":{"kind":"done"}}),
            ),
        ];

        let (events, saw_done) = dispatch(frames).await;
        assert!(saw_done, "should have observed the command Done marker");
        assert_eq!(events.len(), 6);
        assert_eq!(
            events[0],
            InvokeCommandEvent::Output(CommandOutputEvent::Text {
                content: "A".into()
            })
        );
        assert!(matches!(
            events[1],
            InvokeCommandEvent::Task(TaskEvent::Start {
                kind: TaskKind::Download,
                ..
            })
        ));
        assert!(matches!(
            events[2],
            InvokeCommandEvent::Task(TaskEvent::Update { .. })
        ));
        assert!(matches!(
            events[3],
            InvokeCommandEvent::Output(CommandOutputEvent::System { .. })
        ));
        assert!(matches!(
            events[4],
            InvokeCommandEvent::Task(TaskEvent::Done { error: None, .. })
        ));
        assert_eq!(
            events[5],
            InvokeCommandEvent::Output(CommandOutputEvent::Done)
        );
    }

    #[tokio::test]
    async fn ignores_command_output_for_unrelated_request_id() {
        let (events, saw_done) = dispatch(vec![frame(
            "command.output",
            json!({"request_id":"other","event":{"kind":"text","content":"x"}}),
        )])
        .await;
        assert!(!saw_done);
        assert!(events.is_empty());
    }

    #[tokio::test]
    async fn skips_malformed_command_output_without_aborting() {
        let (events, saw_done) = dispatch(vec![
            // Missing 'kind'
            frame("command.output", json!({"request_id":"r1","event":{}})),
            // Followed by a good event — must still be delivered.
            frame(
                "command.output",
                json!({"request_id":"r1","event":{"kind":"done"}}),
            ),
        ])
        .await;
        assert!(saw_done);
        assert_eq!(
            events,
            vec![InvokeCommandEvent::Output(CommandOutputEvent::Done)]
        );
    }

    #[tokio::test]
    async fn task_events_pass_through_regardless_of_request_id() {
        let (events, _) = dispatch(vec![frame("task.log", json!({"id":"abc","line":"..."}))]).await;
        match &events[..] {
            [InvokeCommandEvent::Task(TaskEvent::Log { id, line })] => {
                assert_eq!(id, "abc");
                assert_eq!(line, "...");
            }
            other => panic!("unexpected: {other:?}"),
        }
    }

    #[tokio::test]
    async fn unrelated_methods_are_dropped() {
        let (events, saw_done) = dispatch(vec![frame(
            "provider.stream.event",
            json!({"type":"text","delta":"x"}),
        )])
        .await;
        assert!(!saw_done);
        assert!(events.is_empty());
    }

    /// A closed collector (dropped receiver) latches `sink_open` false and
    /// never blocks or aborts the dispatch loop.
    #[tokio::test]
    async fn closed_sink_latches_open_false_without_blocking() {
        let (tx, collector) = invoke_event_channel(InvokeOutputBudget::default());
        drop(collector);
        let mut open = true;
        let saw_done = ProcessExtension::forward_invoke_command_frame(
            "ext-test",
            "r1",
            &tx,
            &mut open,
            frame(
                "command.output",
                json!({"request_id":"r1","event":{"kind":"done"}}),
            ),
        )
        .await;
        // Done detection is independent of sink state.
        assert!(saw_done);
        assert!(!open);
    }
}

#[cfg(all(test, unix))]
mod lifecycle_tests {
    use super::*;
    use std::time::Duration;

    // Offline stdio fixture: read exactly one request frame, then echo its id.
    // No Python, network, model calls, or filesystem fixtures are required.
    async fn fixture() -> ProcessExtension {
        let result = serde_json::json!({
            "protocol_version": CURRENT_EXTENSION_PROTOCOL_VERSION,
            "capabilities": {}
        })
        .to_string();
        let script = format!(
            r#"while IFS= read -r header; do
    length=${{header#Content-Length: }}
    length=${{length%?}}
    IFS= read -r blank || exit 1
    body=$(dd bs=1 count="$length" 2>/dev/null) || exit 1
    id=${{body##*\"id\":}}
    id=${{id%%[!0-9]*}}
    case "$body" in
        *'"method":"pending"'*) IFS= read -r hold; continue ;;
    esac
    reply='{{"jsonrpc":"2.0","id":'"$id"',"result":{result}}}'
    printf 'Content-Length: %s\r\n\r\n%s' "${{#reply}}" "$reply"
done"#
        );
        let ext = ProcessExtension::spawn("lifecycle-fixture", "/bin/sh", &["-c".into(), script])
            .await
            .expect("spawn offline lifecycle fixture");
        tokio::time::timeout(Duration::from_secs(3), ext.initialize_for_test(None))
            .await
            .expect("bounded initialize")
            .expect("initialize fixture");
        ext
    }

    /// Offline fixture that answers `initialize` normally and every other
    /// method with a JSON-RPC `-32601` error object (a live, healthy process
    /// that simply lacks the method — e.g. the optional `info.get` probe).
    async fn unknown_method_fixture() -> ProcessExtension {
        let init = serde_json::json!({
            "protocol_version": CURRENT_EXTENSION_PROTOCOL_VERSION,
            "capabilities": {}
        })
        .to_string();
        let script = format!(
            r#"while IFS= read -r header; do
    length=${{header#Content-Length: }}
    length=${{length%?}}
    IFS= read -r blank || exit 1
    body=$(dd bs=1 count="$length" 2>/dev/null) || exit 1
    id=${{body##*\"id\":}}
    id=${{id%%[!0-9]*}}
    case "$body" in
        *'"method":"initialize"'*) reply='{{"jsonrpc":"2.0","id":'"$id"',"result":{init}}}' ;;
        *) reply='{{"jsonrpc":"2.0","id":'"$id"',"error":{{"code":-32601,"message":"unknown method"}}}}' ;;
    esac
    printf 'Content-Length: %s

%s' "${{#reply}}" "$reply"
done"#
        );
        let ext =
            ProcessExtension::spawn("unknown-method-fixture", "/bin/sh", &["-c".into(), script])
                .await
                .expect("spawn offline unknown-method fixture");
        tokio::time::timeout(Duration::from_secs(3), ext.initialize_for_test(None))
            .await
            .expect("bounded initialize")
            .expect("initialize fixture");
        ext
    }

    /// A JSON-RPC `error` reply proves the process is alive: the call must
    /// surface it verbatim, with no restart, no generation change and no
    /// consumption of the restart budget. (Regression: the optional
    /// `info.get` probe used to re-spawn every tool-bearing extension at
    /// boot just to hear "unknown method" a second time.)
    #[tokio::test]
    async fn application_error_reply_never_restarts_the_process() {
        let ext = unknown_method_fixture().await;
        let before = ext.lifecycle_snapshot().unwrap();
        for method in ["info.get", "sidecar.spawn_args", "no.such.method"] {
            let error = tokio::time::timeout(Duration::from_secs(3), ext.call(method, Value::Null))
                .await
                .expect("bounded call")
                .expect_err("fixture answers every non-initialize method with an error");
            assert!(is_application_error(&error), "{method}: {error}");
            assert!(error.contains("unknown method"), "{method}: {error}");
            assert!(
                !error.contains("retry after restart failed"),
                "{method}: must not restart on an application error: {error}"
            );
        }
        assert_eq!(ext.total_restarts.load(Ordering::Relaxed), 0);
        assert_eq!(ext.restart_count.load(Ordering::Relaxed), 0);
        let after = ext.lifecycle_snapshot().unwrap();
        assert_eq!(after.generation, before.generation);
        assert_eq!(after.health, before.health);
        // Transport failures still take the restart path.
        assert!(!is_application_error(
            "transport closed: response channel dropped"
        ));
        assert!(!is_application_error("Write error: broken pipe"));
        ext.force_shutdown().await;
    }

    async fn wait_failed(ext: &ProcessExtension) -> ExtensionLifecycle {
        tokio::time::timeout(Duration::from_secs(3), async {
            loop {
                if let Some(snapshot) = ext.lifecycle_snapshot() {
                    if snapshot.health == ExtensionHealth::Failed {
                        return snapshot;
                    }
                }
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("lifecycle must detect failure without another RPC")
    }

    #[tokio::test]
    async fn actual_child_exit_is_detected_while_rpc_holds_state_and_reader_is_stopped() {
        let ext = fixture().await;
        let before = ext.lifecycle_snapshot().expect("live snapshot");
        assert_eq!(before.health, ExtensionHealth::Running);
        let call_guard = ext.call_lock.lock().await;
        let mut state_guard = ext.state.lock().await;
        let state = state_guard.as_mut().unwrap();
        let lifecycle = state.lifecycle.clone();
        state.reader_handle.abort();
        // No reader can observe EOF or close the response channel in this test.
        tokio::time::timeout(Duration::from_secs(3), async {
            while !state.reader_handle.is_finished() {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("reader cancellation");
        {
            let rpc = ext.call_once_locked(state, "pending", Value::Null, 900);
            tokio::pin!(rpc);
            tokio::select! {
                biased;
                result = &mut rpc => panic!("RPC must remain pending: {result:?}"),
                _ = tokio::time::sleep(Duration::from_millis(20)) => {}
            }
            assert!(ext.inbox.pending.lock().await.contains_key(&900));
            assert_eq!(ext.lifecycle_snapshot(), Some(before));
            lifecycle.child.lock().await.start_kill().unwrap();
            let dead = wait_failed(&ext).await;
            assert_ne!(dead.generation, before.generation);
            assert_eq!(ext.lifecycle_snapshot(), Some(dead));
            assert!(!ext.inbox.closed.load(Ordering::Acquire));
            assert!(ext.inbox.pending.lock().await.contains_key(&900));
        }
        drop(state_guard);
        drop(call_guard);
        ext.force_shutdown().await;
    }

    #[tokio::test]
    async fn shutdown_invalidates_generation_before_waiting_for_rpc_lock() {
        let ext = Arc::new(fixture().await);
        let before = ext.lifecycle_snapshot().unwrap();
        let call_guard = ext.call_lock.lock().await;
        let shutdown_ext = ext.clone();
        let shutdown = tokio::spawn(async move { shutdown_ext.shutdown().await });
        let dead = wait_failed(&ext).await;
        assert_ne!(dead.generation, before.generation);
        assert!(!shutdown.is_finished());
        drop(call_guard);
        tokio::time::timeout(Duration::from_secs(3), shutdown)
            .await
            .expect("bounded shutdown")
            .expect("shutdown task");
        assert_eq!(ext.lifecycle_snapshot(), Some(dead));
    }

    /// A call that reaches a handler after `shutdown` (it was authorized
    /// before an unload or reload) fails; it must not respawn the process
    /// and run a second copy beside the replacement.
    #[tokio::test]
    async fn a_call_after_shutdown_fails_without_respawning() {
        let ext = fixture().await;
        ext.shutdown().await;
        let err = ext
            .call_inner("tool.call", serde_json::json!({}))
            .await
            .expect_err("a stopped handler must refuse the call");
        assert!(err.contains("was stopped"), "{err}");
        assert!(ext.state.lock().await.is_none(), "no process was respawned");
        assert_eq!(ext.total_restarts.load(Ordering::Relaxed), 0);
    }

    #[tokio::test]
    async fn in_place_restart_changes_generation_and_old_invalidation_is_isolated() {
        let ext = fixture().await;
        let before = ext.lifecycle_snapshot().unwrap();
        let mut state = ext.state.lock().await;
        let old = state.as_ref().unwrap().lifecycle.clone();
        {
            // Hold teardown at the child lock to inspect the invalidation window
            // before a replacement can be spawned or initialized.
            let child_guard = old.child.lock().await;
            let restart = ext.restart_locked(&mut state);
            tokio::pin!(restart);
            tokio::select! {
                biased;
                result = &mut restart => panic!("restart must await child lock: {result:?}"),
                _ = tokio::time::sleep(Duration::from_millis(20)) => {}
            }
            let replacing = ext.lifecycle_snapshot().unwrap();
            assert_ne!(replacing.generation, before.generation);
            assert_eq!(replacing.health, ExtensionHealth::Failed);
            drop(child_guard);
            tokio::time::timeout(Duration::from_secs(5), restart)
                .await
                .expect("bounded restart")
                .expect("restart fixture");
        }
        let after = ext.lifecycle_snapshot().unwrap();
        assert_ne!(after.generation, before.generation);
        assert_eq!(after.health, ExtensionHealth::Degraded);
        assert_eq!(old.snapshot(false).unwrap().health, ExtensionHealth::Failed);
        old.invalidate();
        assert_eq!(ext.lifecycle_snapshot(), Some(after));
        drop(state);
        ext.force_shutdown().await;
        let stopped = ext.lifecycle_snapshot().unwrap();
        assert_ne!(stopped.generation, after.generation);
        assert_eq!(stopped.health, ExtensionHealth::Failed);
    }

    #[tokio::test]
    async fn transport_eof_invalidates_generation_even_while_child_is_alive() {
        // Close stdout, but keep the actual child blocked on stdin.
        let ext = ProcessExtension::spawn(
            "lifecycle-eof",
            "/bin/sh",
            &[
                "-c".into(),
                "IFS= read -r trigger; exec 1>&-; IFS= read -r hold".into(),
            ],
        )
        .await
        .unwrap();
        let mut state_guard = ext.state.lock().await;
        let state = state_guard.as_mut().unwrap();
        state.lifecycle.initialized.store(true, Ordering::Release);
        let before = ext.lifecycle_snapshot().unwrap();
        state
            .stdin
            .lock()
            .await
            .write_all(b"close\n")
            .await
            .unwrap();
        let dead = wait_failed(&ext).await;
        assert_ne!(dead.generation, before.generation);
        assert!(state
            .lifecycle
            .child
            .lock()
            .await
            .try_wait()
            .unwrap()
            .is_none());
        drop(state_guard);
        ext.force_shutdown().await;
    }

    #[tokio::test]
    async fn contended_child_probe_is_unavailable_without_changing_generation() {
        let ext = fixture().await;
        let before = ext.lifecycle_snapshot().unwrap();
        let lifecycle = ext.inbox.lifecycle.lock().await.clone().unwrap();
        let child_guard = lifecycle.child.lock().await;
        assert_eq!(ext.lifecycle_snapshot(), None);
        drop(child_guard);
        assert_eq!(ext.lifecycle_snapshot(), Some(before));
        let slot_guard = ext.inbox.lifecycle.lock().await;
        assert_eq!(ext.lifecycle_snapshot(), None);
        drop(slot_guard);
        assert_eq!(ext.lifecycle_snapshot(), Some(before));
        ext.force_shutdown().await;
    }

    /// Offline fixture: `initialize` answers normally, `hook.handle`
    /// replaces every output with `CRUSHED`.
    async fn replace_fixture() -> ProcessExtension {
        let init = serde_json::json!({
            "protocol_version": CURRENT_EXTENSION_PROTOCOL_VERSION,
            "capabilities": {}
        })
        .to_string();
        let script = format!(
            r#"while IFS= read -r header; do
    length=${{header#Content-Length: }}
    length=${{length%?}}
    IFS= read -r blank || exit 1
    body=$(dd bs=1 count="$length" 2>/dev/null) || exit 1
    id=${{body##*\"id\":}}
    id=${{id%%[!0-9]*}}
    case "$body" in
        *'"method":"initialize"'*) reply='{{"jsonrpc":"2.0","id":'"$id"',"result":{init}}}' ;;
        *'"method":"hook.handle"'*) reply='{{"jsonrpc":"2.0","id":'"$id"',"result":{{"action":"replace","output":"CRUSHED"}}}}' ;;
        *) reply='{{"jsonrpc":"2.0","id":'"$id"',"result":{{}}}}' ;;
    esac
    printf 'Content-Length: %s\r\n\r\n%s' "${{#reply}}" "$reply"
done"#
        );
        let ext = ProcessExtension::spawn("replace-fixture", "/bin/sh", &["-c".into(), script])
            .await
            .expect("spawn offline replace fixture");
        tokio::time::timeout(Duration::from_secs(3), ext.initialize_for_test(None))
            .await
            .expect("bounded initialize")
            .expect("initialize fixture");
        ext
    }

    /// The subagent-worker hook path (`handle_no_spawn`) uses the running
    /// child and never starts or restarts one: a crashed or stopped
    /// extension passes the output through (`Continue`) at once, and only
    /// the restart-capable `handle` (the foreground) brings it back.
    #[tokio::test]
    async fn a_worker_hook_never_respawns_a_crashed_or_stopped_extension() {
        let ext = replace_fixture().await;
        let event = HookEvent::after_tool_call(
            "bash",
            serde_json::json!({"command": "ls"}),
            "RAW".to_string(),
        );
        let crushed = HookResult::Replace { output: "CRUSHED".into() };
        async fn bounded<F: std::future::Future>(fut: F) -> F::Output {
            tokio::time::timeout(Duration::from_secs(3), fut)
                .await
                .expect("bounded: never hangs")
        }
        assert_eq!(bounded(ext.handle_no_spawn(&event)).await, crushed);

        // Crash: the child dies; the reader sees EOF and closes the inbox.
        let lifecycle = ext.state.lock().await.as_ref().unwrap().lifecycle.clone();
        lifecycle.child.lock().await.start_kill().unwrap();
        tokio::time::timeout(Duration::from_secs(3), async {
            while !ext.inbox.closed.load(Ordering::Acquire) {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("reader observes the crash");
        for _ in 0..3 {
            assert_eq!(
                bounded(ext.handle_no_spawn(&event)).await,
                HookResult::Continue,
                "crashed: pass-through"
            );
        }
        assert_eq!(ext.total_restarts.load(Ordering::Relaxed), 0, "a worker never restarts it");
        assert!(ext.inbox.closed.load(Ordering::Acquire), "nothing was respawned");

        // The foreground's restart-capable path brings it back; workers then
        // use the restarted child.
        assert_eq!(bounded(ext.handle(&event)).await, crushed);
        assert_eq!(ext.total_restarts.load(Ordering::Relaxed), 1);
        assert_eq!(bounded(ext.handle_no_spawn(&event)).await, crushed);

        // Stopped (unload/reload): pass-through, and still no respawn.
        ext.shutdown().await;
        assert_eq!(bounded(ext.handle_no_spawn(&event)).await, HookResult::Continue);
        assert!(ext.state.lock().await.is_none(), "no process was respawned");
        assert_eq!(ext.total_restarts.load(Ordering::Relaxed), 1);
    }
}