//! Hook runtimes (in-process, command, HTTP) and deterministic default engine.

// On wasm32, use tokio_with_wasm as a drop-in replacement for tokio.
#[cfg(target_arch = "wasm32")]
mod tokio {
    pub use tokio_with_wasm::alias::*;
}

// Skill registration
inventory::submit! {
    meerkat_skills::SkillRegistration {
        id: "hook-authoring",
        name: "Hook Authoring",
        description: "Writing hooks for the 8 hook points, execution modes, and decision semantics",
        scope: meerkat_core::skills::SkillScope::Builtin,
        requires_capabilities: &["hooks"],
        body: include_str!("../skills/hook-authoring/SKILL.md"),
        extensions: &[],
    }
}

// Capability registration
inventory::submit! {
    meerkat_contracts::CapabilityRegistration {
        id: meerkat_contracts::CapabilityId::Hooks,
        description: "8 hook points, 3 runtimes (in-process/command/HTTP), observe/allow/deny semantics",
        scope: meerkat_contracts::CapabilityScope::Universal,
        requires_feature: None,
        prerequisites: &[],
        status_resolver: None,
    }
}

use chrono::Utc;
use futures::StreamExt;
use meerkat_core::config::HookInProcessRuntimeConfig;
use meerkat_core::time_compat::Duration;
use meerkat_core::{
    HookCapability, HookDecision, HookEngine, HookEngineError, HookEntryConfig, HookExecutionMode,
    HookExecutionReport, HookId, HookInvocation, HookOutcome, HookPatch, HookPatchEnvelope,
    HookRevision, HookRunOverrides, HookRuntimeKind, HooksConfig, SessionId,
};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(not(target_arch = "wasm32"))]
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
#[cfg(not(target_arch = "wasm32"))]
use tokio::process::Command;
use tokio::sync::{Mutex, Semaphore};
use tokio::time::timeout;

pub use meerkat_core::config::HookInProcessHandlerId as InProcessHookHandlerId;

/// Response returned by runtime adapters.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub struct RuntimeHookResponse {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub decision: Option<HookDecision>,
    /// Retired compatibility field. `HookPatch` has no valid variants, so
    /// non-empty legacy semantic patch payloads fail deserialization.
    #[serde(default)]
    pub patches: Vec<HookPatch>,
}

#[derive(Debug, Clone, Deserialize)]
struct CommandRuntimeConfig {
    command: String,
    #[serde(default)]
    args: Vec<String>,
    #[serde(default)]
    env: HashMap<String, String>,
}

#[derive(Debug, Clone, Deserialize)]
struct HttpRuntimeConfig {
    url: String,
    #[serde(default = "default_http_method")]
    method: String,
    #[serde(default)]
    headers: HashMap<String, String>,
}

fn default_http_method() -> String {
    "POST".to_string()
}

#[cfg(not(target_arch = "wasm32"))]
type HandlerFuture = Pin<Box<dyn Future<Output = Result<RuntimeHookResponse, String>> + Send>>;
#[cfg(target_arch = "wasm32")]
type HandlerFuture = Pin<Box<dyn Future<Output = Result<RuntimeHookResponse, String>>>>;

pub type InProcessHookHandler = Arc<dyn Fn(HookInvocation) -> HandlerFuture + Send + Sync>;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct HookPatchDeliveryTarget {
    session_id: SessionId,
}

impl HookPatchDeliveryTarget {
    pub fn for_session(session_id: SessionId) -> Self {
        Self { session_id }
    }

    pub fn session_id(&self) -> &SessionId {
        &self.session_id
    }
}

#[derive(Debug, Clone)]
struct DeliveredHookPatch {
    target: HookPatchDeliveryTarget,
    envelope: HookPatchEnvelope,
}

#[derive(Debug, Clone)]
struct LocalHookPatchDelivery {
    queue: Arc<Mutex<Vec<DeliveredHookPatch>>>,
}

impl LocalHookPatchDelivery {
    fn new() -> Self {
        Self {
            queue: Arc::new(Mutex::new(Vec::new())),
        }
    }

    async fn publish(&self, target: HookPatchDeliveryTarget, patches: Vec<HookPatchEnvelope>) {
        if patches.is_empty() {
            return;
        }

        let mut guard = self.queue.lock().await;
        let mut patches = patches;
        guard.reserve(patches.len());
        let Some(last) = patches.pop() else {
            return;
        };
        for envelope in patches {
            guard.push(DeliveredHookPatch {
                target: target.clone(),
                envelope,
            });
        }
        guard.push(DeliveredHookPatch {
            target,
            envelope: last,
        });
    }

    async fn drain(&self, target: &HookPatchDeliveryTarget) -> Vec<HookPatchEnvelope> {
        let mut guard = self.queue.lock().await;
        let queued = guard.len();
        let mut remaining = Vec::with_capacity(queued);
        let mut drained = Vec::new();
        for published in guard.drain(..) {
            if &published.target == target {
                drained.push(published.envelope);
            } else {
                remaining.push(published);
            }
        }
        *guard = remaining;
        drained
    }
}

/// Deterministic hook engine used by all control surfaces.
#[derive(Clone)]
pub struct DefaultHookEngine {
    config: HooksConfig,
    base_entries: Arc<Vec<HookEntryConfig>>,
    base_validation_error: Option<String>,
    http_client: Arc<OnceLock<reqwest::Client>>,
    in_process_handlers:
        Arc<std::sync::RwLock<HashMap<InProcessHookHandlerId, InProcessHookHandler>>>,
    background_patch_delivery: LocalHookPatchDelivery,
    background_slots: Arc<Semaphore>,
    revision: Arc<AtomicU64>,
}

impl DefaultHookEngine {
    pub fn new(config: HooksConfig) -> Self {
        let base_validation_error = config
            .entries
            .iter()
            .find_map(|entry| Self::validate_entry(entry).err())
            .map(|err| err.to_string());
        let background_max_concurrency = config.background_max_concurrency.max(1);

        Self {
            base_entries: Arc::new(config.entries.clone()),
            config,
            base_validation_error,
            http_client: Arc::new(OnceLock::new()),
            in_process_handlers: Arc::new(std::sync::RwLock::new(HashMap::new())),
            background_patch_delivery: LocalHookPatchDelivery::new(),
            background_slots: Arc::new(Semaphore::new(background_max_concurrency)),
            revision: Arc::new(AtomicU64::new(1)),
        }
    }

    pub fn with_in_process_handler(
        self,
        name: impl Into<InProcessHookHandlerId>,
        handler: InProcessHookHandler,
    ) -> Self {
        let next = self;
        let name = name.into();
        match next.in_process_handlers.write() {
            Ok(mut map) => {
                map.insert(name, handler);
            }
            Err(err) => {
                tracing::warn!("failed to register in-process hook handler: {}", err);
            }
        }
        next
    }

    pub async fn register_in_process_handler(
        &self,
        name: impl Into<InProcessHookHandlerId>,
        handler: InProcessHookHandler,
    ) {
        match self.in_process_handlers.write() {
            Ok(mut map) => {
                map.insert(name.into(), handler);
            }
            Err(err) => {
                tracing::warn!("failed to register in-process hook handler: {}", err);
            }
        }
    }

    fn next_revision(&self) -> HookRevision {
        HookRevision(self.revision.fetch_add(1, Ordering::SeqCst))
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn read_stream_limited<R>(
        mut stream: R,
        byte_limit: usize,
        stream_name: &str,
    ) -> Result<Vec<u8>, String>
    where
        R: AsyncRead + Unpin,
    {
        let mut out = Vec::with_capacity(byte_limit.min(8 * 1024));
        let mut chunk = [0u8; 8 * 1024];

        loop {
            let read = stream
                .read(&mut chunk)
                .await
                .map_err(|err| format!("failed reading {stream_name}: {err}"))?;
            if read == 0 {
                break;
            }
            out.extend_from_slice(&chunk[..read]);
            if out.len() > byte_limit {
                return Err(format!(
                    "{} exceeds max size: {} > {}",
                    stream_name,
                    out.len(),
                    byte_limit
                ));
            }
        }

        Ok(out)
    }

    fn effective_entries<'a>(
        &'a self,
        overrides: Option<&'a HookRunOverrides>,
    ) -> Result<std::borrow::Cow<'a, [HookEntryConfig]>, HookEngineError> {
        if let Some(reason) = &self.base_validation_error {
            return Err(HookEngineError::InvalidConfiguration(reason.clone()));
        }

        if let Some(overrides) = overrides {
            if overrides.disable.is_empty() && overrides.entries.is_empty() {
                return Ok(std::borrow::Cow::Borrowed(self.base_entries.as_slice()));
            }

            let mut entries = Vec::with_capacity(self.base_entries.len() + overrides.entries.len());
            if overrides.disable.is_empty() {
                entries.extend(self.base_entries.iter().cloned());
            } else {
                let disabled: HashSet<HookId> = overrides.disable.iter().cloned().collect();
                entries.extend(
                    self.base_entries
                        .iter()
                        .filter(|entry| !disabled.contains(&entry.id))
                        .cloned(),
                );
            }
            entries.extend(overrides.entries.clone());

            for entry in &entries {
                Self::validate_entry(entry)?;
            }

            return Ok(std::borrow::Cow::Owned(entries));
        }

        Ok(std::borrow::Cow::Borrowed(self.base_entries.as_slice()))
    }

    async fn drain_published_patches_for_session(
        &self,
        session_id: &SessionId,
    ) -> Vec<HookPatchEnvelope> {
        let target = HookPatchDeliveryTarget::for_session(session_id.clone());
        self.background_patch_delivery.drain(&target).await
    }

    async fn publish_patches(
        &self,
        target: HookPatchDeliveryTarget,
        patches: Vec<HookPatchEnvelope>,
    ) {
        self.background_patch_delivery
            .publish(target, patches)
            .await;
    }

    fn validate_entry(entry: &HookEntryConfig) -> Result<(), HookEngineError> {
        if entry.id.0.trim().is_empty() {
            return Err(HookEngineError::InvalidConfiguration(
                "hook id cannot be empty".to_string(),
            ));
        }

        if entry.mode == HookExecutionMode::Background
            && entry.point.is_pre()
            && entry.capability != HookCapability::Observe
        {
            return Err(HookEngineError::InvalidConfiguration(format!(
                "pre_* background hooks must be observe-only: {}",
                entry.id
            )));
        }

        Ok(())
    }

    async fn execute_one(
        &self,
        entry: HookEntryConfig,
        registration_index: usize,
        invocation: HookInvocation,
    ) -> Result<HookOutcome, HookEngineError> {
        let start = meerkat_core::time_compat::Instant::now();
        let timeout_ms = entry.timeout_ms.unwrap_or(self.config.default_timeout_ms);

        let mut outcome = HookOutcome {
            hook_id: entry.id.clone(),
            point: entry.point,
            priority: entry.priority,
            registration_index,
            decision: None,
            patches: Vec::new(),
            published_patches: Vec::new(),
            error: None,
            duration_ms: None,
        };

        let runtime_result = timeout(
            Duration::from_millis(timeout_ms),
            self.invoke_runtime(&entry, invocation.clone()),
        )
        .await;

        let response = match runtime_result {
            Err(_) => {
                return Err(HookEngineError::Timeout {
                    hook_id: entry.id.clone(),
                    timeout_ms,
                });
            }
            Ok(response) => response?,
        };
        outcome.decision = runtime_decision_with_configured_hook_id(response.decision, &entry.id);
        outcome.patches = response.patches;

        if entry.mode == HookExecutionMode::Background {
            if entry.point.is_pre() {
                if !outcome.patches.is_empty()
                    || matches!(outcome.decision, Some(HookDecision::Deny { .. }))
                {
                    outcome.error = Some("pre_* background hooks are observe-only".to_string());
                }
                outcome.patches.clear();
                outcome.decision = None;
            } else if entry.point.is_post() {
                let mut envelopes = Vec::new();
                for patch in outcome.patches.drain(..) {
                    envelopes.push(HookPatchEnvelope {
                        revision: self.next_revision(),
                        hook_id: entry.id.clone(),
                        point: entry.point,
                        patch,
                        published_at: Utc::now(),
                    });
                }
                outcome.published_patches = envelopes;
                outcome.decision = None;
            }
        }

        outcome.duration_ms = Some(start.elapsed().as_millis() as u64);
        Ok(outcome)
    }

    async fn invoke_runtime(
        &self,
        entry: &HookEntryConfig,
        invocation: HookInvocation,
    ) -> Result<RuntimeHookResponse, HookEngineError> {
        let runtime_config =
            entry
                .runtime
                .config_value()
                .map_err(|err| HookEngineError::ExecutionFailed {
                    hook_id: entry.id.clone(),
                    reason: format!("invalid runtime config payload: {err}"),
                })?;

        match entry.runtime.kind {
            HookRuntimeKind::InProcess => {
                let cfg: HookInProcessRuntimeConfig = entry
                    .runtime
                    .in_process_config()
                    .map_err(|err| HookEngineError::ExecutionFailed {
                        hook_id: entry.id.clone(),
                        reason: format!("invalid in_process runtime config: {err}"),
                    })?
                    .ok_or_else(|| HookEngineError::ExecutionFailed {
                        hook_id: entry.id.clone(),
                        reason: "runtime kind is not in_process".to_string(),
                    })?;
                let handler = {
                    let handlers = self.in_process_handlers.read().map_err(|err| {
                        HookEngineError::ExecutionFailed {
                            hook_id: entry.id.clone(),
                            reason: format!("in-process handler lock poisoned: {err}"),
                        }
                    })?;
                    handlers.get(&cfg.handler).cloned().ok_or_else(|| {
                        HookEngineError::ExecutionFailed {
                            hook_id: entry.id.clone(),
                            reason: format!("in-process handler '{}' not registered", cfg.handler),
                        }
                    })?
                };
                (handler)(invocation)
                    .await
                    .map_err(|reason| HookEngineError::ExecutionFailed {
                        hook_id: entry.id.clone(),
                        reason,
                    })
            }
            HookRuntimeKind::Command => {
                let cfg: CommandRuntimeConfig =
                    serde_json::from_value(runtime_config).map_err(|err| {
                        HookEngineError::ExecutionFailed {
                            hook_id: entry.id.clone(),
                            reason: format!("invalid command runtime config: {err}"),
                        }
                    })?;
                self.invoke_command_runtime(entry, &cfg.command, &cfg.args, &cfg.env, invocation)
                    .await
            }
            HookRuntimeKind::Http => {
                let cfg: HttpRuntimeConfig =
                    serde_json::from_value(runtime_config).map_err(|err| {
                        HookEngineError::ExecutionFailed {
                            hook_id: entry.id.clone(),
                            reason: format!("invalid http runtime config: {err}"),
                        }
                    })?;
                self.invoke_http_runtime(entry, &cfg.url, &cfg.method, &cfg.headers, invocation)
                    .await
            }
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn invoke_command_runtime(
        &self,
        entry: &HookEntryConfig,
        command: &str,
        args: &[String],
        env: &HashMap<String, String>,
        invocation: HookInvocation,
    ) -> Result<RuntimeHookResponse, HookEngineError> {
        let mut child = Command::new(command)
            .args(args)
            .envs(env)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .map_err(|err| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("failed to spawn command hook: {err}"),
            })?;

        let payload =
            serde_json::to_vec(&invocation).map_err(|err| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("failed to encode invocation payload: {err}"),
            })?;

        if payload.len() > self.config.payload_max_bytes {
            return Err(HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!(
                    "hook payload exceeds max size: {} > {}",
                    payload.len(),
                    self.config.payload_max_bytes
                ),
            });
        }

        if let Some(mut stdin) = child.stdin.take() {
            stdin
                .write_all(&payload)
                .await
                .map_err(|err| HookEngineError::ExecutionFailed {
                    hook_id: entry.id.clone(),
                    reason: format!("failed to write command hook stdin: {err}"),
                })?;
        }

        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: "command hook stdout pipe unavailable".to_string(),
            })?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: "command hook stderr pipe unavailable".to_string(),
            })?;

        let max_output_bytes = self.config.payload_max_bytes;
        let stdout_task = tokio::spawn(Self::read_stream_limited(
            stdout,
            max_output_bytes,
            "command stdout",
        ));
        let stderr_task = tokio::spawn(Self::read_stream_limited(
            stderr,
            max_output_bytes,
            "command stderr",
        ));

        let status = child
            .wait()
            .await
            .map_err(|err| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("failed waiting for command hook: {err}"),
            })?;

        let stdout = stdout_task
            .await
            .map_err(|err| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("command stdout task join failed: {err}"),
            })?
            .map_err(|reason| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason,
            })?;
        let stderr = stderr_task
            .await
            .map_err(|err| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("command stderr task join failed: {err}"),
            })?
            .map_err(|reason| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason,
            })?;

        if !status.success() {
            let stderr = String::from_utf8_lossy(&stderr).to_string();
            return Err(HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("command hook exited with {status}: {stderr}"),
            });
        }

        serde_json::from_slice::<RuntimeHookResponse>(&stdout).map_err(|err| {
            HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("invalid command hook response: {err}"),
            }
        })
    }

    #[cfg(target_arch = "wasm32")]
    async fn invoke_command_runtime(
        &self,
        entry: &HookEntryConfig,
        _command: &str,
        _args: &[String],
        _env: &HashMap<String, String>,
        _invocation: HookInvocation,
    ) -> Result<RuntimeHookResponse, HookEngineError> {
        Err(HookEngineError::ExecutionFailed {
            hook_id: entry.id.clone(),
            reason: "command hooks are not supported on wasm32".to_string(),
        })
    }

    async fn invoke_http_runtime(
        &self,
        entry: &HookEntryConfig,
        url: &str,
        method: &str,
        headers: &HashMap<String, String>,
        invocation: HookInvocation,
    ) -> Result<RuntimeHookResponse, HookEngineError> {
        let payload =
            serde_json::to_vec(&invocation).map_err(|err| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("failed to encode invocation payload: {err}"),
            })?;

        if payload.len() > self.config.payload_max_bytes {
            return Err(HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!(
                    "hook payload exceeds max size: {} > {}",
                    payload.len(),
                    self.config.payload_max_bytes
                ),
            });
        }

        let method = reqwest::Method::from_bytes(method.as_bytes()).map_err(|err| {
            HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("invalid HTTP method '{method}': {err}"),
            }
        })?;

        let http_client = self.http_client.get_or_init(reqwest::Client::new);
        let mut req = http_client
            .request(method, url)
            .header("content-type", "application/json")
            .body(payload);

        for (name, value) in headers {
            req = req.header(name, value);
        }

        let response = req
            .send()
            .await
            .map_err(|err| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("HTTP hook request failed: {err}"),
            })?;

        let status = response.status();
        let mut bytes = Vec::new();
        let mut stream = response.bytes_stream();
        while let Some(chunk) = stream.next().await {
            let chunk = chunk.map_err(|err| HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("failed reading HTTP hook response body: {err}"),
            })?;
            bytes.extend_from_slice(&chunk);
            if bytes.len() > self.config.payload_max_bytes {
                return Err(HookEngineError::ExecutionFailed {
                    hook_id: entry.id.clone(),
                    reason: format!(
                        "HTTP hook response exceeds max size: {} > {}",
                        bytes.len(),
                        self.config.payload_max_bytes
                    ),
                });
            }
        }

        if !status.is_success() {
            let body = String::from_utf8_lossy(&bytes);
            return Err(HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("HTTP hook returned {status}: {body}"),
            });
        }

        serde_json::from_slice::<RuntimeHookResponse>(&bytes).map_err(|err| {
            HookEngineError::ExecutionFailed {
                hook_id: entry.id.clone(),
                reason: format!("invalid HTTP hook response: {err}"),
            }
        })
    }
}

fn runtime_decision_with_configured_hook_id(
    decision: Option<HookDecision>,
    hook_id: &HookId,
) -> Option<HookDecision> {
    match decision {
        Some(HookDecision::Deny {
            reason_code,
            message,
            payload,
            ..
        }) => Some(HookDecision::Deny {
            hook_id: hook_id.clone(),
            reason_code,
            message,
            payload,
        }),
        decision => decision,
    }
}

#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
impl HookEngine for DefaultHookEngine {
    fn matching_hooks(
        &self,
        invocation: &HookInvocation,
        overrides: Option<&HookRunOverrides>,
    ) -> Result<Vec<HookId>, HookEngineError> {
        let entries = self.effective_entries(overrides)?;
        Ok(entries
            .iter()
            .filter(|entry| entry.enabled && entry.point == invocation.point)
            .map(|entry| entry.id.clone())
            .collect())
    }

    async fn execute(
        &self,
        invocation: HookInvocation,
        overrides: Option<&HookRunOverrides>,
    ) -> Result<HookExecutionReport, HookEngineError> {
        let mut foreground = Vec::new();
        let mut background = Vec::new();
        let entries = self.effective_entries(overrides)?;
        for (registration_index, entry) in entries
            .iter()
            .filter(|entry| entry.enabled && entry.point == invocation.point)
            .cloned()
            .enumerate()
        {
            if entry.mode == HookExecutionMode::Background {
                background.push((registration_index, entry));
            } else {
                foreground.push((registration_index, entry));
            }
        }

        if foreground.is_empty() && background.is_empty() {
            return Ok(HookExecutionReport::empty());
        }

        foreground.sort_by(|(a_idx, a_entry), (b_idx, b_entry)| {
            a_entry
                .priority
                .cmp(&b_entry.priority)
                .then_with(|| a_idx.cmp(b_idx))
        });

        let mut merged = HookExecutionReport::empty();
        for (registration_index, entry) in foreground {
            let outcome = self
                .execute_one(entry, registration_index, invocation.clone())
                .await?;

            if let Some(decision) = outcome.decision.clone() {
                match &decision {
                    HookDecision::Deny { .. } => {
                        merged.decision = Some(decision);
                    }
                    HookDecision::Allow => {
                        if !matches!(merged.decision, Some(HookDecision::Deny { .. })) {
                            merged.decision = Some(HookDecision::Allow);
                        }
                    }
                }
            }
            merged.patches.extend(outcome.patches.clone());

            let should_stop = matches!(outcome.decision, Some(HookDecision::Deny { .. }));
            merged.outcomes.push(outcome);
            if should_stop {
                break;
            }
        }

        if !matches!(merged.decision, Some(HookDecision::Deny { .. })) {
            for (registration_index, entry) in background {
                let permit = match self.background_slots.clone().try_acquire_owned() {
                    Ok(permit) => permit,
                    Err(_) => {
                        tracing::warn!("background hook queue full, skipping {}", entry.id);
                        continue;
                    }
                };
                let engine = self.clone();
                let invocation_cloned = invocation.clone();
                tokio::spawn(async move {
                    let _permit = permit;
                    let delivery_target =
                        HookPatchDeliveryTarget::for_session(invocation_cloned.session_id.clone());
                    let outcome = match engine
                        .execute_one(entry, registration_index, invocation_cloned)
                        .await
                    {
                        Ok(outcome) => outcome,
                        Err(error) => {
                            tracing::warn!(
                                error = %error,
                                "background hook execution failed"
                            );
                            return;
                        }
                    };
                    if let Some(error) = &outcome.error {
                        tracing::warn!(
                            hook_id = %outcome.hook_id,
                            point = ?outcome.point,
                            error = %error,
                            "background hook execution produced an error"
                        );
                    }
                    if !outcome.published_patches.is_empty() {
                        engine
                            .publish_patches(delivery_target, outcome.published_patches)
                            .await;
                    }
                });
            }
        }

        Ok(merged)
    }

    async fn drain_published_patches(
        &self,
        session_id: &SessionId,
    ) -> Result<Vec<HookPatchEnvelope>, HookEngineError> {
        Ok(self.drain_published_patches_for_session(session_id).await)
    }
}

#[cfg(test)]
#[allow(
    clippy::expect_used,
    clippy::field_reassign_with_default,
    clippy::panic,
    clippy::unwrap_used
)]
mod tests {
    use super::*;
    use meerkat_core::{
        HookFailurePolicy, HookLlmRequest, HookPoint, HookReasonCode, HookRuntimeConfig,
        HookRuntimeKind, SessionId,
    };
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

    fn static_handler(response: RuntimeHookResponse) -> InProcessHookHandler {
        Arc::new(move |_invocation| {
            let response = response.clone();
            Box::pin(async move { Ok(response) })
        })
    }

    fn delayed_handler(delay_ms: u64, response: RuntimeHookResponse) -> InProcessHookHandler {
        Arc::new(move |_invocation| {
            let response = response.clone();
            Box::pin(async move {
                tokio::time::sleep(Duration::from_millis(delay_ms)).await;
                Ok(response)
            })
        })
    }

    fn runtime_in_process(name: &str) -> HookRuntimeConfig {
        HookRuntimeConfig::in_process(name).unwrap_or_default()
    }

    #[test]
    fn in_process_handler_id_preserves_string_wire_shape() {
        let id: InProcessHookHandlerId = serde_json::from_value(serde_json::json!("handler-a"))
            .expect("handler id should deserialize from string");

        assert_eq!(id.as_str(), "handler-a");
        assert_eq!(
            serde_json::to_value(&id).expect("handler id should serialize"),
            serde_json::json!("handler-a")
        );
    }

    #[tokio::test]
    async fn deterministic_merge_by_priority_then_registration() {
        let mut config = HooksConfig::default();
        config.entries = vec![
            HookEntryConfig {
                id: HookId::new("hook-a"),
                point: HookPoint::PreLlmRequest,
                priority: 10,
                runtime: runtime_in_process("a"),
                capability: HookCapability::Observe,
                ..Default::default()
            },
            HookEntryConfig {
                id: HookId::new("hook-b"),
                point: HookPoint::PreLlmRequest,
                priority: 5,
                runtime: runtime_in_process("b"),
                capability: HookCapability::Guardrail,
                ..Default::default()
            },
        ];

        let engine = DefaultHookEngine::new(config);
        engine
            .register_in_process_handler(
                "a",
                static_handler(RuntimeHookResponse {
                    decision: None,
                    patches: Vec::new(),
                }),
            )
            .await;
        engine
            .register_in_process_handler(
                "b",
                static_handler(RuntimeHookResponse {
                    decision: Some(HookDecision::Allow),
                    patches: Vec::new(),
                }),
            )
            .await;

        let report = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreLlmRequest,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: Some(HookLlmRequest {
                        max_tokens: 256,
                        temperature: None,
                        provider_params: None,
                        message_count: 1,
                    }),
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .unwrap();

        let hook_ids: Vec<_> = report
            .outcomes
            .iter()
            .map(|outcome| &outcome.hook_id)
            .collect();
        assert_eq!(
            hook_ids,
            vec![&HookId::new("hook-b"), &HookId::new("hook-a")]
        );
        assert!(matches!(report.decision, Some(HookDecision::Allow)));
        assert!(report.patches.is_empty());
    }

    #[tokio::test]
    async fn background_hooks_are_observation_only_and_publish_no_patches() {
        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: HookId::new("hook-post-bg"),
            point: HookPoint::PostToolExecution,
            mode: HookExecutionMode::Background,
            capability: HookCapability::Observe,
            runtime: runtime_in_process("post-bg"),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        engine
            .register_in_process_handler(
                "post-bg",
                static_handler(RuntimeHookResponse {
                    decision: None,
                    patches: Vec::new(),
                }),
            )
            .await;

        let session_id = SessionId::new();
        let report = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PostToolExecution,
                    session_id: session_id.clone(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .unwrap();

        tokio::time::sleep(Duration::from_millis(10)).await;
        assert!(report.patches.is_empty());
        assert!(report.published_patches.is_empty());
        assert!(
            engine
                .drain_published_patches(&session_id)
                .await
                .unwrap()
                .is_empty()
        );
    }

    #[tokio::test]
    async fn legacy_semantic_patch_payload_from_command_runtime_fails_closed() {
        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: HookId::new("legacy-patch-command"),
            point: HookPoint::PreLlmRequest,
            capability: HookCapability::Rewrite,
            runtime: HookRuntimeConfig::new(
                HookRuntimeKind::Command,
                Some(serde_json::json!({
                    "command": "sh",
                    "args": [
                        "-c",
                        "cat >/dev/null; printf '%s' '{\"patches\":[{\"patch_type\":\"llm_request\",\"max_tokens\":1}]}'"
                    ],
                    "env": {}
                })),
            )
            .unwrap_or_default(),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        let err = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreLlmRequest,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: Some(HookLlmRequest {
                        max_tokens: 256,
                        temperature: None,
                        provider_params: None,
                        message_count: 1,
                    }),
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .expect_err("legacy semantic patch payloads must fail closed");

        assert!(matches!(
            err,
            HookEngineError::ExecutionFailed { ref hook_id, .. }
                if hook_id == &HookId::new("legacy-patch-command")
        ));
    }

    #[tokio::test]
    async fn deterministic_merge_ignores_completion_order() {
        let mut config = HooksConfig::default();
        config.entries = vec![
            HookEntryConfig {
                id: HookId::new("slow-low-priority"),
                point: HookPoint::PreLlmRequest,
                priority: 100,
                runtime: runtime_in_process("slow"),
                capability: HookCapability::Observe,
                ..Default::default()
            },
            HookEntryConfig {
                id: HookId::new("fast-high-priority"),
                point: HookPoint::PreLlmRequest,
                priority: 1,
                runtime: runtime_in_process("fast"),
                capability: HookCapability::Observe,
                ..Default::default()
            },
        ];

        let engine = DefaultHookEngine::new(config);
        engine
            .register_in_process_handler(
                "slow",
                delayed_handler(
                    100,
                    RuntimeHookResponse {
                        decision: None,
                        patches: Vec::new(),
                    },
                ),
            )
            .await;
        engine
            .register_in_process_handler(
                "fast",
                delayed_handler(
                    1,
                    RuntimeHookResponse {
                        decision: None,
                        patches: Vec::new(),
                    },
                ),
            )
            .await;

        let report = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreLlmRequest,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: Some(HookLlmRequest {
                        max_tokens: 256,
                        temperature: None,
                        provider_params: None,
                        message_count: 1,
                    }),
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .unwrap();

        let hook_ids: Vec<_> = report
            .outcomes
            .iter()
            .map(|outcome| &outcome.hook_id)
            .collect();
        assert_eq!(
            hook_ids,
            vec![
                &HookId::new("fast-high-priority"),
                &HookId::new("slow-low-priority")
            ]
        );
        assert!(report.patches.is_empty());
    }

    #[tokio::test]
    async fn observe_runtime_error_returns_typed_engine_failure() {
        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: HookId::new("observe-missing-handler"),
            point: HookPoint::PreToolExecution,
            capability: HookCapability::Observe,
            runtime: runtime_in_process("missing"),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        let err = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreToolExecution,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .expect_err("hook runtime errors must fail closed through typed engine errors");

        assert!(matches!(
            err,
            HookEngineError::ExecutionFailed { ref hook_id, .. }
                if hook_id == &HookId::new("observe-missing-handler")
        ));
    }

    #[tokio::test]
    async fn rewrite_runtime_error_returns_typed_engine_failure() {
        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: HookId::new("rewrite-missing-handler"),
            point: HookPoint::PreToolExecution,
            capability: HookCapability::Rewrite,
            runtime: runtime_in_process("missing"),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        let err = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreToolExecution,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .expect_err("hook runtime errors must not be converted into hook-local denials");

        assert!(matches!(
            err,
            HookEngineError::ExecutionFailed { ref hook_id, .. }
                if hook_id == &HookId::new("rewrite-missing-handler")
        ));
    }

    #[tokio::test]
    async fn deny_short_circuits_lower_priority_hooks() {
        let mut config = HooksConfig::default();
        config.entries = vec![
            HookEntryConfig {
                id: HookId::new("guardrail-deny"),
                point: HookPoint::PreToolExecution,
                priority: 1,
                capability: HookCapability::Guardrail,
                runtime: runtime_in_process("guardrail"),
                ..Default::default()
            },
            HookEntryConfig {
                id: HookId::new("observer-late"),
                point: HookPoint::PreToolExecution,
                priority: 100,
                capability: HookCapability::Observe,
                runtime: runtime_in_process("observer"),
                ..Default::default()
            },
        ];

        let observed_runs = Arc::new(AtomicUsize::new(0));
        let observer_runs = observed_runs.clone();

        let engine = DefaultHookEngine::new(config);
        engine
            .register_in_process_handler(
                "guardrail",
                static_handler(RuntimeHookResponse {
                    decision: Some(HookDecision::deny(
                        HookId::new("guardrail-deny"),
                        HookReasonCode::PolicyViolation,
                        "deny",
                        None,
                    )),
                    patches: Vec::new(),
                }),
            )
            .await;
        engine
            .register_in_process_handler(
                "observer",
                Arc::new(move |_invocation| {
                    let observer_runs = observer_runs.clone();
                    Box::pin(async move {
                        observer_runs.fetch_add(1, AtomicOrdering::SeqCst);
                        Ok(RuntimeHookResponse {
                            decision: Some(HookDecision::Allow),
                            patches: Vec::new(),
                        })
                    })
                }),
            )
            .await;

        let report = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreToolExecution,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .unwrap();

        assert!(matches!(report.decision, Some(HookDecision::Deny { .. })));
        assert_eq!(report.outcomes.len(), 1);
        assert_eq!(observed_runs.load(AtomicOrdering::SeqCst), 0);
    }

    #[tokio::test]
    async fn deny_decision_uses_configured_hook_id_over_runtime_payload() {
        let configured_hook_id = HookId::new("configured-guardrail");
        let returned_hook_id = HookId::new("runtime-payload-identity");

        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: configured_hook_id.clone(),
            point: HookPoint::PreToolExecution,
            capability: HookCapability::Guardrail,
            runtime: runtime_in_process("guardrail"),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        engine
            .register_in_process_handler(
                "guardrail",
                static_handler(RuntimeHookResponse {
                    decision: Some(HookDecision::deny(
                        returned_hook_id,
                        HookReasonCode::PolicyViolation,
                        "deny",
                        None,
                    )),
                    patches: Vec::new(),
                }),
            )
            .await;

        let report = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreToolExecution,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .unwrap();

        match report.decision {
            Some(HookDecision::Deny { hook_id, .. }) => {
                assert_eq!(hook_id, configured_hook_id);
            }
            decision => panic!("expected deny decision, got {decision:?}"),
        }

        assert_eq!(report.outcomes.len(), 1);
        assert_eq!(report.outcomes[0].hook_id, configured_hook_id);
        match &report.outcomes[0].decision {
            Some(HookDecision::Deny { hook_id, .. }) => {
                assert_eq!(hook_id, &configured_hook_id);
            }
            decision => panic!("expected outcome deny decision, got {decision:?}"),
        }
    }

    #[tokio::test]
    async fn run_started_background_guardrail_is_rejected() {
        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: HookId::new("run-start-bg-guardrail"),
            point: HookPoint::RunStarted,
            mode: HookExecutionMode::Background,
            capability: HookCapability::Guardrail,
            runtime: runtime_in_process("guardrail"),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        let err = engine
            .execute(
                HookInvocation {
                    point: HookPoint::RunStarted,
                    session_id: SessionId::new(),
                    turn_number: Some(0),
                    prompt_input: None,
                    prompt: Some("hi".to_string()),
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .expect_err("invalid background pre hook must be rejected");

        assert!(matches!(err, HookEngineError::InvalidConfiguration(_)));
    }

    #[tokio::test]
    async fn fail_open_override_cannot_convert_runtime_error_to_success() {
        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: HookId::new("rewrite-explicit-open"),
            point: HookPoint::PreToolExecution,
            capability: HookCapability::Rewrite,
            failure_policy: Some(HookFailurePolicy::FailOpen),
            runtime: runtime_in_process("missing"),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        let err = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreToolExecution,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .expect_err("legacy fail_open override must not make runtime errors warning-only");

        assert!(matches!(
            err,
            HookEngineError::ExecutionFailed { ref hook_id, .. }
                if hook_id == &HookId::new("rewrite-explicit-open")
        ));
    }

    #[tokio::test]
    async fn background_failure_policy_is_inert_compatibility_config() {
        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: HookId::new("invalid-background-fail-closed"),
            point: HookPoint::PostToolExecution,
            mode: HookExecutionMode::Background,
            capability: HookCapability::Observe,
            failure_policy: Some(HookFailurePolicy::FailClosed),
            runtime: runtime_in_process("bg"),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        let report = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PostToolExecution,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .expect("legacy failure_policy must not be an admission authority");

        assert!(report.decision.is_none());
    }

    #[tokio::test]
    async fn command_runtime_hook_executes() {
        let mut config = HooksConfig::default();
        config.entries = vec![HookEntryConfig {
            id: HookId::new("command-hook"),
            point: HookPoint::PreToolExecution,
            runtime: HookRuntimeConfig::new(
                HookRuntimeKind::Command,
                Some(serde_json::json!({
                    "command": "sh",
                    "args": ["-c", "cat >/dev/null; printf '{\"patches\":[]}'"],
                    "env": {}
                })),
            )
            .unwrap_or_default(),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        let report = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreToolExecution,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .unwrap();
        assert_eq!(report.outcomes.len(), 1);
        assert!(
            report.outcomes[0].error.is_none(),
            "command runtime error: {:?}",
            report.outcomes[0].error
        );
    }

    #[tokio::test]
    #[ignore = "integration-real: binds TCP port and makes real HTTP request"]
    async fn http_runtime_hook_executes() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        // Bind a mock HTTP server. The listener is ready immediately after bind
        // (no sleep race needed).
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            // Accept connections in a loop so retries/keep-alive don't break.
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    break;
                };
                tokio::spawn(async move {
                    let mut buf = vec![0_u8; 4096];
                    let _ = socket.read(&mut buf).await;
                    let body = r#"{"patches":[]}"#;
                    let response = format!(
                        "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
                        body.len(),
                        body
                    );
                    let _ = socket.write_all(response.as_bytes()).await;
                });
            }
        });

        let mut config = HooksConfig::default();
        // Use an explicit generous timeout so this test doesn't flake under
        // parallel test load (the default 5 000 ms is borderline on CI).
        config.default_timeout_ms = 30_000;
        config.entries = vec![HookEntryConfig {
            id: HookId::new("http-hook"),
            point: HookPoint::PreToolExecution,
            runtime: HookRuntimeConfig::new(
                HookRuntimeKind::Http,
                Some(serde_json::json!({
                    "url": format!("http://{}/hook", addr),
                    "method": "POST",
                    "headers": {}
                })),
            )
            .unwrap_or_default(),
            ..Default::default()
        }];

        let engine = DefaultHookEngine::new(config);
        let report = engine
            .execute(
                HookInvocation {
                    point: HookPoint::PreToolExecution,
                    session_id: SessionId::new(),
                    turn_number: Some(1),
                    prompt_input: None,
                    prompt: None,
                    error_report: None,
                    error_class: None,
                    error: None,
                    llm_request: None,
                    llm_response: None,
                    tool_call: None,
                    tool_result: None,
                },
                None,
            )
            .await
            .unwrap();
        assert_eq!(report.outcomes.len(), 1);
        assert!(
            report.outcomes[0].error.is_none(),
            "http runtime error: {:?}",
            report.outcomes[0].error
        );
    }
}
