khive_runtime/

pack.rs

// FILE SIZE JUSTIFICATION: pack.rs is the load-bearing dispatch core — VerbRegistry,
// VerbRegistryBuilder, PackRuntime, DispatchHook, and their test scaffolding all
// share internal state (packs Vec, gate, event_store) that cannot be cleanly split
// without exposing private fields or duplicating the scaffolding. Inline tests cover
// collision detection and dispatch path that require direct access to VerbRegistry
// internals. Split plan: when the verb surface reaches a stable v1 API, extract
// VerbRegistryBuilder into `pack/builder.rs` and gate/event logic into `pack/dispatch.rs`.
//! Pack runtime trait and verb registry.
//!
//! `PackRuntime` mirrors `Pack`'s const associated items as methods for object safety.
//! Build a [`VerbRegistry`] via `VerbRegistryBuilder::build()`; registration is builder-only.

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::time::Instant;

use crate::runtime::NamespaceToken;
use async_trait::async_trait;
use khive_gate::{ActorRef, AllowAllGate, AuditEvent, GateDecision, GateRef, GateRequest};
use khive_storage::{Event, EventStore, EventView, SubstrateKind};
use khive_types::{EventKind, EventOutcome, Namespace};
use serde_json::Value;

pub use khive_types::{
    EdgeEndpointRule, EndpointKind, HandlerDef, NoteKindSpec, NoteLifecycleSpec, PackSchemaPlan,
    ParamDef, VerbCategory, VerbPresentationPolicy, Visibility,
};
// Backward-compat re-export.
#[allow(deprecated)]
pub use khive_types::VerbDef;

use crate::validation::ValidationRule;

/// Pack-auxiliary schema plan.
///
/// Declares `CREATE TABLE IF NOT EXISTS` statements for pack-owned tables that
/// are NOT part of the core substrate schema (entities, notes, edges, events).
/// Applied at boot via `StorageBackend::apply_schema` / `apply_pack_schema_plan`.
///
/// Core substrate tables evolve through versioned migrations. Pack schema is
/// strictly for pack-auxiliary tables (e.g. GTD lifecycle audit, memory index).
/// v1 pack schemas are non-versioned.
#[derive(Debug, Default, Clone)]
pub struct SchemaPlan {
    /// Owning pack name.
    pub pack: &'static str,
    /// DDL statements applied idempotently at boot.
    /// Each entry must be a self-contained `CREATE TABLE IF NOT EXISTS` or
    /// similar idempotent statement.
    pub statements: &'static [&'static str],
}

impl SchemaPlan {
    /// Construct a `SchemaPlan` with no statements.
    ///
    /// Packs whose state lives entirely in the core substrate tables (entities,
    /// notes, edges) use this as their `schema_plan()` return value.
    pub const fn empty() -> Self {
        Self {
            pack: "",
            statements: &[],
        }
    }

    /// Returns `true` when the plan contains no DDL statements.
    pub fn is_empty(&self) -> bool {
        self.statements.is_empty()
    }
}

/// Hook called after every successful verb dispatch (Issue #158).
///
/// Packs observe enriched event views so provenance-aware consumers can use
/// `view.observations` while legacy folds can still consume `view.event`.
#[async_trait]
pub trait DispatchHook: Send + Sync {
    /// Called with the dispatch-outcome event view after a successful pack dispatch.
    ///
    /// Errors are logged via `tracing::warn!` and never propagated to the
    /// caller; the dispatch has already succeeded.
    async fn on_dispatch(&self, view: &EventView);
}

use crate::error::{
    CircularPackDependency, MissingPackDependencies, MissingPackDependency, RuntimeError,
};
use crate::KhiveRuntime;

/// Async dispatch trait for packs.
///
/// This is the object-safe behavioral counterpart to `khive_types::Pack`.
/// `Pack` uses const associated items (not object-safe in Rust); this trait
/// mirrors that metadata as methods and adds async dispatch.
///
/// Registration requires `P: Pack + PackRuntime` — the compiler enforces
/// that every runtime pack also declares its vocabulary via `Pack`.
#[async_trait]
pub trait PackRuntime: Send + Sync {
    /// Pack name — must equal `<Self as Pack>::NAME`.
    fn name(&self) -> &str;

    /// Note kinds this pack owns — must equal `<Self as Pack>::NOTE_KINDS`.
    fn note_kinds(&self) -> &'static [&'static str];

    /// Entity kinds this pack owns — must equal `<Self as Pack>::ENTITY_KINDS`.
    fn entity_kinds(&self) -> &'static [&'static str];

    /// Handlers this pack registers — must equal `<Self as Pack>::HANDLERS`.
    fn handlers(&self) -> &'static [HandlerDef];

    /// Pack-extensible edge endpoint rules — must equal `<Self as Pack>::EDGE_RULES`.
    /// Defaults to empty so existing packs that don't extend the edge contract
    /// can ignore it.
    fn edge_rules(&self) -> &'static [EdgeEndpointRule] {
        &[]
    }

    /// Pack names whose vocabulary this pack references.
    /// Defaults to empty so existing packs compile without changes.
    fn requires(&self) -> &'static [&'static str] {
        &[]
    }

    /// NoteKindSpec declarations for note kinds this pack owns.
    ///
    /// Packs that introduce note kinds with explicit lifecycle semantics
    /// declare the spec here.  The runtime collects these for introspection
    /// and future enforcement.  Defaults to empty so existing packs compile
    /// without changes.
    fn note_kind_specs(&self) -> &'static [NoteKindSpec] {
        &[]
    }

    /// Optional per-kind hook for shared CRUD specialization.
    ///
    /// When a kind is owned by this pack (declared in `note_kinds()` or
    /// `entity_kinds()`), returning `Some(hook)` opts that kind into
    /// pack-specific behavior — defaults, derived properties, side-effect
    /// edges — through the shared `create` path. Returning `None` keeps
    /// the kind as plain storage with no specialization.
    fn kind_hook(&self, _kind: &str) -> Option<Arc<dyn KindHook>> {
        None
    }

    /// Pack-auxiliary schema.
    ///
    /// Returns DDL statements for pack-owned tables that are NOT part of the
    /// core substrate schema. Statements are idempotent (`CREATE TABLE IF NOT
    /// EXISTS`) so callers can apply them safely on every registration. Core
    /// substrate tables evolve through versioned migrations; pack schema is
    /// strictly pack-auxiliary.
    ///
    /// Defaults to an empty plan — packs that store everything in the core
    /// substrate tables (entities, notes, edges, events) return this default.
    ///
    /// Plans are aggregated via [`VerbRegistry::all_schema_plans`] and applied
    /// at startup via `KhiveMcpServer::with_packs` (c12). Packs that need their
    /// schema present (e.g. GTD) also self-bootstrap lazily on first call for
    /// robustness in test contexts that create fresh in-memory databases.
    fn schema_plan(&self) -> SchemaPlan {
        SchemaPlan::empty()
    }

    /// Domain-specific validation rules contributed by this pack.
    ///
    /// Rule IDs MUST follow the `<pack>/<rule-id>` namespace convention.
    /// Built-in rules (no pack prefix) are reserved for the `khive-runtime`
    /// validation infrastructure.
    ///
    /// Defaults to empty — packs with no domain-specific rules return `&[]`.
    fn validation_rules(&self) -> &'static [ValidationRule] {
        &[]
    }

    /// Register custom embedding providers with the runtime.
    ///
    /// Called by the transport during pack initialisation, before the first verb
    /// dispatch, so that `KhiveRuntime::embedder(name)` resolves provider names
    /// declared here.
    ///
    /// Implement this method to contribute non-lattice embedding backends:
    ///
    /// ```ignore
    /// fn register_embedders(&self, runtime: &KhiveRuntime) {
    ///     runtime.register_embedder(MyCustomProvider::new());
    /// }
    /// ```
    ///
    /// The default no-op preserves backwards compatibility — packs that only
    /// use built-in lattice models do not need to override this method.
    fn register_embedders(&self, _runtime: &KhiveRuntime) {}

    /// Install a pack-owned entity-type validator on the runtime.
    ///
    /// Called by the transport during pack initialisation, after the registry
    /// is built and before the first verb dispatch, so that `create_many` and
    /// `create_entity` reject unregistered `entity_type` values at the runtime
    /// layer in addition to the handler layer.
    ///
    /// Packs that own `EntityTypeRegistry` vocabularies (e.g. `KgPack`) should
    /// override this to install their registry's `resolve` function.  The
    /// default no-op leaves the runtime validator absent (skip-when-None), which
    /// is the correct behaviour for bare runtimes without packs.
    fn register_entity_type_validator(&self, _runtime: &KhiveRuntime) {}

    /// Warm up any in-memory state from persisted snapshots (optional).
    ///
    /// Called by the transport after all packs are registered but before
    /// serving the first request, giving packs a chance to pre-load expensive
    /// in-memory structures (e.g. ANN indexes) so that the first query does
    /// not incur rebuild latency.
    ///
    /// The default no-op is correct for all packs that have no warm-start
    /// state. Packs that override this must make it idempotent and infallible:
    /// any errors are logged internally, not propagated to the caller.
    async fn warm(&self) {}

    /// Dispatch a verb call. Returns serialized JSON response.
    ///
    /// The `registry` parameter gives the handler access to the merged
    /// vocabulary and kind hooks across all loaded packs.
    /// The `token` is an authorized namespace token minted by the dispatch
    /// boundary after gate authorization — handlers must use it directly.
    async fn dispatch(
        &self,
        verb: &str,
        params: Value,
        registry: &VerbRegistry,
        token: &NamespaceToken,
    ) -> Result<Value, RuntimeError>;
}

/// Per-kind specialization for shared CRUD.
///
/// Packs implement `KindHook` for kinds they own that need:
/// - **Defaults** filled into create args (e.g. `status="inbox"` for tasks)
/// - **Derived properties** computed from args (e.g. salience from priority)
/// - **Side-effect writes** after the storage commit (e.g. `depends_on` edges)
///
/// Hooks are stateless from the framework's perspective — they receive the
/// runtime as a method parameter and operate on the args `Value` directly.
/// The pack registers them via [`PackRuntime::kind_hook`].
///
/// Lifecycle verbs (e.g. gtd's `complete`, `transition`) remain pack-owned
/// verbs and do not flow through this trait — only the create path does.
#[async_trait]
pub trait KindHook: Send + Sync + std::fmt::Debug {
    /// Mutate args before the storage write. Fill defaults, normalize values,
    /// rearrange user-facing fields into the storage shape expected by the
    /// shared CRUD handler.
    ///
    /// Returning an error aborts the create call (no storage write happens).
    async fn prepare_create(
        &self,
        runtime: &KhiveRuntime,
        args: &mut Value,
    ) -> Result<(), RuntimeError>;

    /// Fire side effects after a successful storage write — graph edges,
    /// derived observations, etc. The newly created record's UUID is passed
    /// so the hook can attach metadata referencing it.
    ///
    /// Errors here are **logged but not propagated** — the storage write has
    /// already succeeded; failing the call would mislead the caller.
    /// Implementations should `tracing::warn!` and return `Ok(())` for
    /// best-effort side effects.
    async fn after_create(
        &self,
        runtime: &KhiveRuntime,
        id: uuid::Uuid,
        args: &Value,
    ) -> Result<(), RuntimeError>;
}

/// Optional sub-trait for packs that own private SQL tables and issue UUIDs
/// that must be reachable through the generic `get(id)` and `delete(id)` verbs.
///
/// Implementing both methods is required — the sub-trait bundles them atomically
/// so partial implementation is a compile-time error, not a runtime surprise.
/// Packs whose records live in the shared entity/note substrate (gtd, memory)
/// do not implement this sub-trait.
#[async_trait]
pub trait PackByIdResolver: Send + Sync {
    /// Attempt to resolve a live (non-deleted) UUID owned by this pack's private tables.
    ///
    /// Returns `Some(Resolved::PackRecord { ... })` if this pack owns the UUID,
    /// `None` if it does not (the caller continues to the next resolver),
    /// or `Err(...)` on a storage error.
    ///
    /// Must query domain-authoritative tables before mirror tables.
    /// Must NOT filter by namespace. UUID v4 is globally unique; by-ID
    /// resolution is namespace-blind per ADR-007.
    async fn resolve_by_id(
        &self,
        id: uuid::Uuid,
    ) -> Result<Option<crate::Resolved>, crate::RuntimeError>;

    /// Attempt to resolve a UUID including already-soft-deleted records.
    ///
    /// Used by the hard-delete path. Default delegates to `resolve_by_id`;
    /// packs with `deleted_at` columns override this to query without the filter.
    async fn resolve_by_id_including_deleted(
        &self,
        id: uuid::Uuid,
    ) -> Result<Option<crate::Resolved>, crate::RuntimeError> {
        self.resolve_by_id(id).await
    }

    /// Delete a record owned by this pack's private tables.
    ///
    /// `hard` mirrors the `delete` verb's `hard?` argument.
    /// Default behavior for packs with a `deleted_at` column MUST be soft-delete;
    /// `hard=true` performs permanent removal.
    ///
    /// Returns `Ok(Value)` with a `{ deleted: true, id, kind, hard }` body on success.
    /// Returns `Err(RuntimeError::NotFound(...))` if the record does not exist.
    async fn delete_by_id(
        &self,
        id: uuid::Uuid,
        hard: bool,
    ) -> Result<serde_json::Value, crate::RuntimeError>;
}

/// Builder for constructing a `VerbRegistry`.
///
/// Packs are registered here; once `.build()` is called the registry is
/// immutable and cheaply cloneable.
pub struct VerbRegistryBuilder {
    packs: Vec<Box<dyn PackRuntime>>,
    resolvers: Vec<(String, Box<dyn PackByIdResolver>)>,
    gate: GateRef,
    default_namespace: String,
    /// Operator-configured read-visibility set (ADR-007 Rev 4 Rule 3b).
    ///
    /// Threads into `VerbRegistry::visible_namespaces` and is consumed by the
    /// default dispatch path to widen read scope to `['local'] ∪ visible_namespaces`.
    /// Writes remain pinned to `'local'`. An explicit `namespace=` request param
    /// is a precise escape and is not widened by this set. A cloud gate may also
    /// consult the list as policy input at its own layer.
    visible_namespaces: Vec<Namespace>,
    /// Configured actor identity label (ADR-057). When set, dispatch mints tokens
    /// carrying this actor so that `comm.inbox` filters by `to_actor`.
    actor_id: Option<String>,
    /// Optional audit event sink.
    ///
    /// When set, every gate check writes a storage `Event` in addition to the
    /// `tracing::info!` emission. The store is `Arc<dyn EventStore>` so the
    /// registry does not depend on the full `KhiveRuntime` surface — only the
    /// audit-persistence capability is needed here.
    event_store: Option<Arc<dyn EventStore>>,
    /// Optional post-dispatch hook (Issue #158).
    ///
    /// When set, every successful pack dispatch calls `hook.on_dispatch(event)`
    /// with a synthesized Event describing the outcome. Opt-in: when None,
    /// no overhead is incurred.
    dispatch_hook: Option<Arc<dyn DispatchHook>>,
}

impl VerbRegistryBuilder {
    /// Create a builder with no packs, `AllowAllGate`, and the local namespace as default.
    pub fn new() -> Self {
        Self {
            packs: Vec::new(),
            resolvers: Vec::new(),
            gate: std::sync::Arc::new(AllowAllGate),
            default_namespace: Namespace::local().as_str().to_string(),
            visible_namespaces: vec![],
            actor_id: None,
            event_store: None,
            dispatch_hook: None,
        }
    }

    /// Set the operator-configured read-visibility set (ADR-007 Rev 4 Rule 3b).
    ///
    /// On the default (no explicit `namespace=` param) dispatch path, reads fan
    /// out over `['local'] ∪ ns`. Writes remain pinned to `'local'`. An explicit
    /// `namespace=` request parameter is a precise single-namespace escape and
    /// is not widened by this set. A cloud gate may also consult the list as
    /// policy input at its own layer.
    pub fn with_visible_namespaces(&mut self, ns: Vec<Namespace>) -> &mut Self {
        self.visible_namespaces = ns;
        self
    }

    /// Set the configured actor identity label (ADR-057).
    ///
    /// When set, the dispatch path mints tokens carrying this actor so that
    /// `comm.inbox` applies the `to_actor` filter for directed delivery.
    /// When `None` (default), tokens carry `ActorRef::anonymous()` and inbox
    /// falls back to party-line behavior.
    pub fn with_actor_id(&mut self, actor_id: Option<String>) -> &mut Self {
        self.actor_id = actor_id;
        self
    }

    /// Register a pack. The bound `P: Pack + PackRuntime` ensures the pack
    /// declares vocabulary via `Pack` consts alongside runtime dispatch.
    pub fn register<P: khive_types::Pack + PackRuntime + 'static>(&mut self, pack: P) -> &mut Self {
        self.packs.push(Box::new(pack));
        self
    }

    /// Register a boxed pack directly.
    ///
    /// Crate-private: only [`PackRegistry::register_packs`] should call this.
    /// External callers must use the typed [`Self::register`] which enforces the
    /// `Pack + PackRuntime` dual-impl contract at the call site.  Here the
    /// contract is satisfied upstream at the [`PackFactory::create`] site.
    pub(crate) fn register_boxed(&mut self, pack: Box<dyn PackRuntime>) -> &mut Self {
        self.packs.push(pack);
        self
    }

    /// Register a by-ID resolver for a pack that owns private SQL tables.
    ///
    /// Packs that implement `PackByIdResolver` call this during their boot path
    /// so that `get(id)` and `delete(id)` can reach their records.
    pub fn register_resolver(
        &mut self,
        name: impl Into<String>,
        resolver: Box<dyn PackByIdResolver>,
    ) -> &mut Self {
        self.resolvers.push((name.into(), resolver));
        self
    }

    /// Set the authorization gate consulted on every dispatch.
    ///
    /// Defaults to `AllowAllGate` if not set. `Deny` is authoritative — a deny
    /// decision aborts dispatch with `RuntimeError::PermissionDenied`. Gate
    /// infrastructure errors fail open (logged via `tracing::warn!`, dispatch
    /// proceeds).
    pub fn with_gate(&mut self, gate: GateRef) -> &mut Self {
        self.gate = gate;
        self
    }

    /// Set the namespace surfaced to the gate when a verb does not carry an
    /// explicit `namespace` argument. Transports should plumb the runtime's
    /// `default_namespace` so the gate's `input.namespace` always reflects
    /// the operation's true tenant.
    pub fn with_default_namespace(&mut self, ns: impl Into<String>) -> &mut Self {
        self.default_namespace = ns.into();
        self
    }

    /// Set the `EventStore` used to persist audit events.
    ///
    /// When configured, every gate check appends one `Event` (substrate =
    /// `Event`, outcome = `Success` on allow, `Denied` on deny) in addition to
    /// the `tracing::info!` emission that was already present in v0.2.
    ///
    /// Callers that do not set this field continue to use tracing-only emission
    /// (the v0.2 default). There is no behavior change for them.
    pub fn with_event_store(&mut self, store: Arc<dyn EventStore>) -> &mut Self {
        self.event_store = Some(store);
        self
    }

    /// Register a post-dispatch hook (Issue #158).
    ///
    /// When set, every successful pack dispatch calls `hook.on_dispatch(event)`
    /// with a synthesized [`Event`] describing the verb outcome. The hook is
    /// opt-in: registries without a hook incur zero overhead on the dispatch
    /// hot path.
    ///
    /// Brain pack uses this to update its posteriors in real time without
    /// polling the EventStore. Errors from `on_dispatch` are logged via
    /// `tracing::warn!` and never propagated.
    pub fn with_dispatch_hook(&mut self, hook: Arc<dyn DispatchHook>) -> &mut Self {
        self.dispatch_hook = Some(hook);
        self
    }

    /// Consume the builder and produce an immutable, cloneable registry.
    ///
    /// Performs a topological sort of packs using Kahn's algorithm.
    /// Returns an error if any declared dependency is missing from the loaded
    /// pack set, or if a circular dependency is detected.
    pub fn build(self) -> Result<VerbRegistry, RuntimeError> {
        let packs = self.packs;
        let mut name_to_idx: HashMap<&str, usize> = HashMap::with_capacity(packs.len());
        for (idx, pack) in packs.iter().enumerate() {
            if let Some(prev_idx) = name_to_idx.insert(pack.name(), idx) {
                return Err(RuntimeError::PackRedeclared {
                    name: pack.name().to_string(),
                    first_idx: prev_idx,
                    second_idx: idx,
                });
            }
        }

        let mut missing: Vec<MissingPackDependency> = Vec::new();
        let mut indegree = vec![0usize; packs.len()];
        let mut dependents: Vec<Vec<usize>> = vec![Vec::new(); packs.len()];

        for (idx, pack) in packs.iter().enumerate() {
            for &requires in pack.requires() {
                match name_to_idx.get(requires).copied() {
                    Some(dep_idx) => {
                        dependents[dep_idx].push(idx);
                        indegree[idx] += 1;
                    }
                    None => missing.push(MissingPackDependency {
                        from: pack.name().to_string(),
                        requires: requires.to_string(),
                    }),
                }
            }
        }

        if !missing.is_empty() {
            return if missing.len() == 1 {
                Err(RuntimeError::MissingPackDependency(missing.remove(0)))
            } else {
                Err(RuntimeError::MissingPackDependencies(
                    MissingPackDependencies { missing },
                ))
            };
        }

        let mut ready: VecDeque<usize> = indegree
            .iter()
            .enumerate()
            .filter_map(|(idx, degree)| (*degree == 0).then_some(idx))
            .collect();
        let mut ordered_indices = Vec::with_capacity(packs.len());

        while let Some(idx) = ready.pop_front() {
            ordered_indices.push(idx);
            for &dep_idx in &dependents[idx] {
                indegree[dep_idx] -= 1;
                if indegree[dep_idx] == 0 {
                    ready.push_back(dep_idx);
                }
            }
        }

        if ordered_indices.len() != packs.len() {
            let cycle_nodes: HashSet<usize> = indegree
                .iter()
                .enumerate()
                .filter_map(|(idx, degree)| (*degree > 0).then_some(idx))
                .collect();
            let cycle = find_pack_dependency_cycle(&packs, &name_to_idx, &cycle_nodes);
            return Err(RuntimeError::CircularPackDependency(
                CircularPackDependency { cycle },
            ));
        }

        let mut slots: Vec<Option<Box<dyn PackRuntime>>> = packs.into_iter().map(Some).collect();
        let ordered_packs: Vec<Box<dyn PackRuntime>> = ordered_indices
            .into_iter()
            .map(|idx| slots[idx].take().expect("topological index must exist"))
            .collect();

        validate_unique_note_kinds(&ordered_packs)?;
        validate_unique_verb_names(&ordered_packs)?;

        Ok(VerbRegistry {
            packs: Arc::new(ordered_packs),
            resolvers: Arc::new(self.resolvers),
            gate: self.gate,
            default_namespace: self.default_namespace,
            visible_namespaces: self.visible_namespaces,
            actor_id: self.actor_id,
            event_store: self.event_store,
            dispatch_hook: self.dispatch_hook,
        })
    }
}

/// Validate that no two packs declare the same note kind (F073).
///
/// Boot-time duplicate detection prevents pack configuration errors from
/// silently corrupting note kind routing. Returns an error naming the
/// duplicate kind and the two packs that claim it.
fn validate_unique_note_kinds(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
    let mut seen: HashMap<&str, &str> = HashMap::new();
    for pack in packs {
        for &kind in pack.note_kinds() {
            if let Some(first_pack) = seen.insert(kind, pack.name()) {
                return Err(RuntimeError::InvalidInput(format!(
                    "duplicate note kind {kind:?}: claimed by both {first_pack:?} and {:?}",
                    pack.name()
                )));
            }
        }
    }
    Ok(())
}

/// Validate that no two packs declare the same `Visibility::Verb` handler name
/// (F093).
///
/// `Visibility::Subhandler` entries are pack-prefixed by convention and excluded
/// from cross-pack collision detection. Two packs declaring the same subhandler
/// name prefix (e.g. `recall.embed`) would be a pack-authoring error but does not
/// produce a cross-pack routing conflict since only the owning pack dispatches them.
fn validate_unique_verb_names(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
    let mut seen: HashMap<&str, &str> = HashMap::new();
    for pack in packs {
        for handler in pack.handlers() {
            if !matches!(handler.visibility, Visibility::Verb) {
                continue;
            }
            if let Some(first_pack) = seen.insert(handler.name, pack.name()) {
                return Err(RuntimeError::VerbCollision {
                    verb: handler.name.to_string(),
                    first_pack: first_pack.to_string(),
                    second_pack: pack.name().to_string(),
                });
            }
        }
    }
    Ok(())
}

fn find_pack_dependency_cycle(
    packs: &[Box<dyn PackRuntime>],
    name_to_idx: &HashMap<&str, usize>,
    cycle_nodes: &HashSet<usize>,
) -> Vec<String> {
    fn visit(
        idx: usize,
        packs: &[Box<dyn PackRuntime>],
        name_to_idx: &HashMap<&str, usize>,
        cycle_nodes: &HashSet<usize>,
        visiting: &mut Vec<usize>,
        visited: &mut HashSet<usize>,
    ) -> Option<Vec<String>> {
        if let Some(pos) = visiting.iter().position(|&seen| seen == idx) {
            let mut cycle: Vec<String> = visiting[pos..]
                .iter()
                .map(|&i| packs[i].name().to_string())
                .collect();
            cycle.push(packs[idx].name().to_string());
            return Some(cycle);
        }
        if !visited.insert(idx) {
            return None;
        }
        visiting.push(idx);
        for &req in packs[idx].requires() {
            let Some(&dep_idx) = name_to_idx.get(req) else {
                continue;
            };
            if cycle_nodes.contains(&dep_idx) {
                if let Some(cycle) =
                    visit(dep_idx, packs, name_to_idx, cycle_nodes, visiting, visited)
                {
                    return Some(cycle);
                }
            }
        }
        visiting.pop();
        None
    }

    let mut visited = HashSet::new();
    for &idx in cycle_nodes {
        let mut visiting = Vec::new();
        if let Some(cycle) = visit(
            idx,
            packs,
            name_to_idx,
            cycle_nodes,
            &mut visiting,
            &mut visited,
        ) {
            return cycle;
        }
    }
    cycle_nodes
        .iter()
        .map(|&idx| packs[idx].name().to_string())
        .collect()
}

impl Default for VerbRegistryBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Immutable registry that dispatches verb calls to registered packs.
///
/// Clone is cheap (Arc-wrapped). Constructed via `VerbRegistryBuilder`.
#[derive(Clone)]
pub struct VerbRegistry {
    packs: std::sync::Arc<Vec<Box<dyn PackRuntime>>>,
    /// Pack-level by-ID resolvers, in registration order.
    resolvers: std::sync::Arc<Vec<(String, Box<dyn PackByIdResolver>)>>,
    gate: GateRef,
    default_namespace: String,
    /// Operator-configured read-visibility set (ADR-007 Rev 4 Rule 3b).
    ///
    /// On the default (no explicit `namespace=` param) dispatch path, reads fan
    /// out over `['local'] ∪ visible_namespaces`. Writes are unaffected — they
    /// still pin to `'local'`. An explicit `namespace=` request param is a
    /// precise single-namespace escape and is not widened by this set.
    visible_namespaces: Vec<Namespace>,
    /// Configured actor identity label (ADR-057). When `Some`, dispatch mints
    /// tokens carrying this actor so that `comm.inbox` applies the `to_actor`
    /// filter. When `None`, tokens carry `ActorRef::anonymous()` (party-line).
    actor_id: Option<String>,
    /// Audit event sink — `None` means tracing-only (v0.2 default).
    event_store: Option<Arc<dyn EventStore>>,
    /// Post-dispatch hook — `None` means no real-time observation (Issue #158).
    dispatch_hook: Option<Arc<dyn DispatchHook>>,
}

/// Error returned by [`VerbRegistry::apply_schema_plans_with_map`] when two
/// packs on the same backend declare the same auxiliary table (ADR-028 §7).
#[derive(Debug)]
pub struct PackSchemaCollisionError {
    /// First pack to declare the table.
    pub pack_a: &'static str,
    /// Second pack that collides with `pack_a`.
    pub pack_b: &'static str,
    /// Table name or DDL error description.
    pub table: String,
}

impl std::fmt::Display for PackSchemaCollisionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if self.pack_a == self.pack_b {
            write!(
                f,
                "pack schema boot failure for pack {:?}: {}",
                self.pack_a, self.table
            )
        } else {
            write!(
                f,
                "pack schema collision: packs {:?} and {:?} both declare table {:?} \
                 on the same backend — move one pack to a separate backend or rename the table",
                self.pack_a, self.pack_b, self.table
            )
        }
    }
}

impl std::error::Error for PackSchemaCollisionError {}

/// Extract table names from a single DDL statement.
///
/// Handles `CREATE TABLE IF NOT EXISTS`, `CREATE TABLE`, and
/// `CREATE VIRTUAL TABLE IF NOT EXISTS`, `CREATE VIRTUAL TABLE`.
/// Returns an empty Vec when no table name is found (e.g. index DDL).
fn extract_table_names(stmt: &str) -> Vec<String> {
    let normalized = stmt.split_whitespace().collect::<Vec<_>>().join(" ");
    let upper = normalized.to_ascii_uppercase();
    let table_name = if let Some(rest) = upper.strip_prefix("CREATE VIRTUAL TABLE IF NOT EXISTS ") {
        rest.split_whitespace().next()
    } else if let Some(rest) = upper.strip_prefix("CREATE VIRTUAL TABLE ") {
        rest.split_whitespace().next()
    } else if let Some(rest) = upper.strip_prefix("CREATE TABLE IF NOT EXISTS ") {
        rest.split_whitespace().next()
    } else if let Some(rest) = upper.strip_prefix("CREATE TABLE ") {
        rest.split_whitespace().next()
    } else {
        None
    };
    match table_name {
        Some(name) => {
            let clean = name.trim_matches(|c: char| c == '(' || c == ';');
            if clean.is_empty() {
                vec![]
            } else {
                vec![clean.to_ascii_lowercase()]
            }
        }
        None => vec![],
    }
}

impl VerbRegistry {
    /// Return the help schema envelope for a verb (issue #287).
    ///
    /// Walks registered packs for the first matching `HandlerDef` and returns a
    /// structured JSON envelope. Subhandlers carry `callable_via_mcp: false`.
    /// Unknown verbs return `RuntimeError::InvalidInput`. Full shape documented
    /// in `docs/protocol.md` §Request Schema.
    pub fn describe_verb(&self, verb: &str) -> Result<Value, RuntimeError> {
        for pack in self.packs.iter() {
            for handler in pack.handlers().iter() {
                if handler.name == verb {
                    let category = format!("{:?}", handler.category);
                    let params_arr: Vec<Value> = handler
                        .params
                        .iter()
                        .map(|p| {
                            serde_json::json!({
                                "name": p.name,
                                "type": p.param_type,
                                "required": p.required,
                                "description": p.description,
                            })
                        })
                        .collect();
                    // ue-help-introspection C1: subhandlers are not callable via
                    // the MCP request surface.  The help payload must match the
                    // behaviour the dispatch path enforces so that agents who
                    // read `help=true` before probing see accurate availability.
                    if matches!(handler.visibility, Visibility::Subhandler) {
                        return Ok(serde_json::json!({
                            "verb": verb,
                            "pack": pack.name(),
                            "description": handler.description,
                            "category": category,
                            "params": params_arr,
                            "visibility": "internal",
                            "callable_via_mcp": false,
                            "note": "This is an internal subhandler. Calling it via the MCP \
                                     request surface returns permission denied. It can only be \
                                     invoked by internal runtime callers.",
                        }));
                    }
                    return Ok(serde_json::json!({
                        "verb": verb,
                        "pack": pack.name(),
                        "description": handler.description,
                        "category": category,
                        "params": params_arr,
                    }));
                }
            }
        }
        // Only list Verb-visibility handlers so internal subhandlers are not
        // advertised in the unknown-verb error (ue-help-introspection C1 / codex High).
        let available: Vec<&str> = self
            .packs
            .iter()
            .flat_map(|p| p.handlers().iter())
            .filter(|h| matches!(h.visibility, Visibility::Verb))
            .map(|h| h.name)
            .collect();
        Err(RuntimeError::InvalidInput(format!(
            "unknown verb {verb:?}; available: {}",
            available.join(", ")
        )))
    }

    /// Check whether the gate permits writes into `ns`.
    ///
    /// Performs a gate evaluation with verb `"authorize"` before any background
    /// loop is spawned (ADR-056 §6).  Returns `Ok(())` when the gate allows the
    /// namespace, or `Err(RuntimeError::PermissionDenied{..})` when denied.
    /// Gate errors (implementation failures) are surfaced as
    /// `RuntimeError::Internal`.
    pub fn authorize_namespace(&self, ns: Namespace) -> Result<(), RuntimeError> {
        let actor = match self.actor_id.as_deref() {
            Some(id) if !id.trim().is_empty() => ActorRef::new("actor", id),
            _ => ActorRef::anonymous(),
        };
        let req = GateRequest::new(actor, ns, "authorize", serde_json::Value::Null);
        match self.gate.check(&req) {
            Ok(decision) if decision.is_allow() => Ok(()),
            Ok(GateDecision::Deny { reason }) => Err(RuntimeError::PermissionDenied {
                verb: "authorize".to_string(),
                reason,
            }),
            Ok(_) => Err(RuntimeError::PermissionDenied {
                verb: "authorize".to_string(),
                reason: "gate denied".to_string(),
            }),
            Err(e) => Err(RuntimeError::Internal(format!("gate error: {e}"))),
        }
    }

    /// Dispatch a verb to the first pack that handles it.
    ///
    /// Routes through the gate, then invokes the matching pack handler. When
    /// `params["help"] == true`, short-circuits to `describe_verb` with no side effects.
    /// Gate errors are fail-open. Full dispatch flow documented in `docs/protocol.md`.
    pub async fn dispatch(&self, verb: &str, params: Value) -> Result<Value, RuntimeError> {
        // help=true interception (issue #287) — short-circuit before gate/pack.
        if params.get("help").and_then(Value::as_bool) == Some(true) {
            return self.describe_verb(verb);
        }
        // Resolve namespace as an owned String before `params` is moved into
        // pack.dispatch, so the post-dispatch hook can reference it.
        let ns_str: String = params
            .get("namespace")
            .and_then(Value::as_str)
            .map(str::to_string)
            .unwrap_or_else(|| self.default_namespace.clone());
        let ns = Namespace::parse(&ns_str)
            .map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace: {e}")))?;
        // ADR-057: thread the configured actor identity into the gate request so
        // the gate can distinguish human vs agent callers at the dispatch seam.
        // Mirrors the actor resolution used by the token-minting path below.
        let gate_actor = match self.actor_id.as_deref() {
            Some(id) if !id.trim().is_empty() => ActorRef::new("actor", id),
            _ => ActorRef::anonymous(),
        };
        let gate_req = GateRequest::new(gate_actor, ns, verb, params.clone());

        // Consult the gate.
        //
        // - Ok(Allow) → proceed to pack dispatch (tracing + optional EventStore).
        // - Ok(Deny) → emit audit, persist if store configured, return PermissionDenied.
        // - Err(_) → warn via tracing, fail-open (no audit persisted).
        let gate_blocked = match self.gate.check(&gate_req) {
            Ok(decision) => {
                let is_deny = matches!(decision, GateDecision::Deny { .. });

                // Emit audit event via tracing.
                let audit = AuditEvent::from_check(&gate_req, &decision, self.gate.impl_name());
                tracing::info!(
                    audit_event = %serde_json::to_string(&audit)
                        .unwrap_or_else(|_| "{\"error\":\"serialize\"}".into()),
                    "gate.check"
                );

                // Persist to EventStore when configured.
                if let Some(store) = &self.event_store {
                    let outcome = if is_deny {
                        EventOutcome::Denied
                    } else {
                        EventOutcome::Success
                    };
                    let audit_data = serde_json::to_value(&audit).unwrap_or_else(|e| {
                        tracing::warn!(error = %e, "failed to serialize AuditEvent for EventStore");
                        serde_json::Value::Null
                    });
                    let mut storage_event = Event::new(
                        gate_req.namespace.as_str(),
                        verb,
                        EventKind::Audit,
                        SubstrateKind::Event,
                        format!("{}:{}", gate_req.actor.kind, gate_req.actor.id),
                    )
                    .with_outcome(outcome)
                    .with_payload(audit_data);
                    if let Some(target_id) = target_id_from_args(&gate_req.args) {
                        storage_event = storage_event.with_target(target_id);
                    }
                    if let Err(store_err) = store.append_event(storage_event).await {
                        tracing::warn!(
                            verb,
                            error = %store_err,
                            "audit event store write failed (non-fatal)"
                        );
                    }
                }

                if is_deny {
                    let reason = match decision {
                        GateDecision::Deny { reason } => reason,
                        _ => String::new(),
                    };
                    Some(reason)
                } else {
                    None
                }
            }
            Err(err) => {
                // Gate infrastructure failure — fail-open.
                // No decision was produced; no audit event is persisted.
                tracing::warn!(verb, error = %err, "gate check failed (fail-open)");
                None
            }
        };

        // Hard enforcement: Deny is authoritative.
        if let Some(reason) = gate_blocked {
            return Err(RuntimeError::PermissionDenied {
                verb: verb.to_string(),
                reason,
            });
        }

        // Mint the authorized storage token at the dispatch boundary.
        //
        // ADR-007 Rev 4 Rule 0/3/3b: writes pin to `local` by default. Actor
        // identity and config `[actor] id` are attribution and gate-context inputs
        // only — they never route storage. The explicit `namespace=` request param
        // is a precise single-namespace escape (Rule 3): the caller deliberately
        // reads/writes exactly that one set; it is NOT widened by `visible_namespaces`.
        //
        // When actor_id is configured (ADR-057), mint a token carrying that actor
        // label so that comm.inbox applies the to_actor filter for directed delivery.
        // Otherwise, use ActorRef::anonymous() and inbox falls back to party-line.
        let configured_actor = match self.actor_id.as_deref() {
            Some(id) if !id.trim().is_empty() => ActorRef::new("actor", id),
            _ => ActorRef::anonymous(),
        };

        // Rule 3b (Rev 4): on the default (no explicit `namespace=`) path, the read
        // scope widens to `['local'] ∪ self.visible_namespaces`. `'local'` is always
        // included (mint_with_visibility deduplicates). Writes remain pinned to
        // `'local'`. Per-actor distinctions use view-layer tag filters
        // (assignee, actor_id, from/to), not namespace partitions.
        let token = match params.get("namespace").and_then(Value::as_str) {
            Some(ns) => {
                // Rule 3 explicit escape: precise single-namespace scope, read+write. NOT widened.
                let primary = Namespace::parse(ns)
                    .map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace: {e}")))?;
                NamespaceToken::mint_with_visibility(primary, vec![], configured_actor)
            }
            None => {
                // Rule 3b: default path. Write namespace = local; read scope = ['local'] ∪ visible_namespaces.
                let primary = Namespace::local();
                let mut extra_visible = self.visible_namespaces.clone();
                extra_visible.push(Namespace::local()); // 'local' always readable; mint dedups
                NamespaceToken::mint_with_visibility(primary, extra_visible, configured_actor)
            }
        };

        for pack in self.packs.iter() {
            if let Some(handler_def) = pack.handlers().iter().find(|v| v.name == verb) {
                // Strip `namespace` from params before forwarding to packs.
                // The registry has already consumed it to mint the NamespaceToken.
                //
                // Exception: if the handler's own `params` schema declares
                // `"namespace"` as a valid field (e.g. brain.bind, brain.unbind,
                // brain.bindings, brain.resolve), the field is a *business* argument
                // — not a transport routing key — and must be passed through
                // unchanged. Stripping it would silently default the binding to the
                // "*" wildcard, broadening profile scope across namespaces.
                let handler_accepts_namespace =
                    handler_def.params.iter().any(|p| p.name == "namespace");
                let params = if !handler_accepts_namespace {
                    if let Value::Object(mut map) = params {
                        map.remove("namespace");
                        Value::Object(map)
                    } else {
                        params
                    }
                } else {
                    params
                };
                let dispatch_start = Instant::now();
                let result = pack.dispatch(verb, params, self, &token).await;
                let dispatch_us = dispatch_start.elapsed().as_micros() as i64;

                // Post-dispatch hook: fires on success, opt-in (Issue #158).
                if let (Ok(ref ok_val), Some(hook)) = (&result, &self.dispatch_hook) {
                    let mut dispatch_event = Event::new(
                        ns_str.as_str(),
                        verb,
                        EventKind::Audit,
                        SubstrateKind::Event,
                        pack.name(),
                    )
                    .with_outcome(EventOutcome::Success)
                    .with_duration_us(dispatch_us);

                    // For recall verbs: extract the first result's id as
                    // target_id so the brain temporal posterior can observe
                    // real hit/miss and latency (fix for codex P12 Major).
                    if verb == "memory.recall" {
                        let first_note_id = ok_val
                            .as_array()
                            .and_then(|arr| arr.first())
                            .and_then(|v| v.get("id"))
                            .and_then(|v| v.as_str())
                            .and_then(|s| s.parse::<uuid::Uuid>().ok());
                        if let Some(note_id) = first_note_id {
                            dispatch_event = dispatch_event.with_target(note_id);
                        }
                        // No first result → target_id stays None (RecallMiss
                        // in brain's event interpreter).
                    }

                    let dispatch_view = EventView {
                        event: dispatch_event,
                        observations: Vec::new(),
                    };
                    let hook = Arc::clone(hook);
                    hook.on_dispatch(&dispatch_view).await;
                }

                return result;
            }
        }
        // Only list Verb-visibility handlers so internal subhandlers are not
        // advertised in the unknown-verb error (ue-help-introspection C1 / codex High).
        let available: Vec<&str> = self
            .packs
            .iter()
            .flat_map(|p| p.handlers().iter())
            .filter(|h| matches!(h.visibility, Visibility::Verb))
            .map(|h| h.name)
            .collect();
        Err(RuntimeError::InvalidInput(format!(
            "unknown verb {verb:?}; available: {}",
            available.join(", ")
        )))
    }

    /// Registered pack-level by-ID resolvers, in registration order.
    ///
    /// Each element is `(pack_name, resolver)`. The kg `get` and `delete` handlers
    /// iterate this slice to probe pack-private tables when the standard KG
    /// substrates (entity/note/edge/event) return `None` for a given UUID.
    pub fn resolvers(&self) -> &[(String, Box<dyn PackByIdResolver>)] {
        &self.resolvers
    }

    /// Find a kind hook among the registered packs.
    ///
    /// Walks packs in registration order; the first pack that both owns the
    /// kind (declares it in `note_kinds()` or `entity_kinds()`) and returns
    /// a hook from `kind_hook(kind)` wins. Returns `None` if the kind is
    /// unknown to all packs or no owning pack registered a hook.
    pub fn find_kind_hook(&self, kind: &str) -> Option<Arc<dyn KindHook>> {
        for pack in self.packs.iter() {
            let owns = pack.note_kinds().contains(&kind) || pack.entity_kinds().contains(&kind);
            if owns {
                if let Some(hook) = pack.kind_hook(kind) {
                    return Some(hook);
                }
            }
        }
        None
    }

    /// All MCP-exposed handlers across all registered packs (`Visibility::Verb` only).
    ///
    /// Subhandlers (`Visibility::Subhandler`) are excluded — they are internal
    /// pipeline steps not surfaced on the MCP wire (F118). Returned with `'static`
    /// lifetime since pack handlers are `&'static [HandlerDef]` constants.
    pub fn all_verbs(&self) -> Vec<&'static HandlerDef> {
        self.packs
            .iter()
            .flat_map(|p| p.handlers().iter())
            .filter(|h| matches!(h.visibility, Visibility::Verb))
            .collect()
    }

    /// All MCP-exposed handlers paired with the name of the pack that owns them
    /// (`Visibility::Verb` only).
    ///
    /// Subhandlers (`Visibility::Subhandler`) are excluded from the MCP catalog
    /// (F118-F123). Use `all_handlers_with_names` when internal handlers must
    /// also be enumerated (e.g. runtime introspection).
    pub fn all_verbs_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
        self.packs
            .iter()
            .flat_map(|p| p.handlers().iter().map(move |v| (p.name(), v)))
            .filter(|(_, h)| matches!(h.visibility, Visibility::Verb))
            .collect()
    }

    /// All handler definitions across all registered packs, including subhandlers.
    ///
    /// Unlike `all_verbs`, this includes `Visibility::Subhandler` entries. Useful
    /// for runtime introspection (e.g. `list_handlers`) and tooling that needs
    /// the complete handler surface.
    pub fn all_handlers_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
        self.packs
            .iter()
            .flat_map(|p| p.handlers().iter().map(move |v| (p.name(), v)))
            .collect()
    }

    /// Merged set of note kinds across all registered packs (deduplicated,
    /// first-seen order preserved).
    pub fn all_note_kinds(&self) -> Vec<&'static str> {
        let mut seen = std::collections::HashSet::new();
        self.packs
            .iter()
            .flat_map(|p| p.note_kinds().iter().copied())
            .filter(|k| seen.insert(*k))
            .collect()
    }

    /// Merged set of entity kinds across all registered packs (deduplicated,
    /// first-seen order preserved).
    pub fn all_entity_kinds(&self) -> Vec<&'static str> {
        let mut seen = std::collections::HashSet::new();
        self.packs
            .iter()
            .flat_map(|p| p.entity_kinds().iter().copied())
            .filter(|k| seen.insert(*k))
            .collect()
    }

    /// Names of packs in topological load order.
    pub fn pack_names(&self) -> Vec<&str> {
        self.packs.iter().map(|p| p.name()).collect()
    }

    /// Declared dependencies for a registered pack.
    pub fn pack_requires(&self, name: &str) -> Option<&'static [&'static str]> {
        self.packs
            .iter()
            .find(|p| p.name() == name)
            .map(|p| p.requires())
    }

    /// Note kinds owned by a specific registered pack.
    ///
    /// Returns `None` if no pack with `name` is registered. The slice is
    /// the pack's `NOTE_KINDS` constant — `'static` lifetime, no allocation.
    pub fn pack_note_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
        self.packs
            .iter()
            .find(|p| p.name() == name)
            .map(|p| p.note_kinds())
    }

    /// Entity kinds owned by a specific registered pack.
    ///
    /// Returns `None` if no pack with `name` is registered. The slice is
    /// the pack's `ENTITY_KINDS` constant — `'static` lifetime, no allocation.
    pub fn pack_entity_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
        self.packs
            .iter()
            .find(|p| p.name() == name)
            .map(|p| p.entity_kinds())
    }

    /// Handlers declared by a specific registered pack.
    ///
    /// Returns `None` if no pack with `name` is registered. Each `HandlerDef`
    /// carries name + description + visibility — sufficient for introspection clients.
    pub fn pack_verbs(&self, name: &str) -> Option<&'static [HandlerDef]> {
        self.packs
            .iter()
            .find(|p| p.name() == name)
            .map(|p| p.handlers())
    }

    /// All pack-declared edge endpoint rules across registered packs.
    ///
    /// Order follows topological pack registration; duplicates are *not* deduplicated —
    /// validation only checks membership, and an exact-duplicate rule is a
    /// harmless restatement.
    pub fn all_edge_rules(&self) -> Vec<EdgeEndpointRule> {
        self.packs
            .iter()
            .flat_map(|p| p.edge_rules().iter().copied())
            .collect()
    }

    /// Collect all `NoteKindSpec` declarations from every loaded pack.
    ///
    /// Used by the runtime for lifecycle introspection and future enforcement.
    pub fn all_note_kind_specs(&self) -> Vec<&'static NoteKindSpec> {
        self.packs
            .iter()
            .flat_map(|p| p.note_kind_specs().iter())
            .collect()
    }

    /// All pack-contributed validation rules across registered packs.
    ///
    /// Returns references into the pack-owned `'static` slices — no allocation
    /// beyond the outer `Vec`. Rule IDs are namespaced by pack; callers can
    /// group by `rule.id.split_once('/')` to attribute rules to their packs.
    pub fn all_validation_rules(&self) -> Vec<&'static ValidationRule> {
        self.packs
            .iter()
            .flat_map(|p| p.validation_rules().iter())
            .collect()
    }

    /// Pack-auxiliary schema plans for all registered packs.
    ///
    /// Returns one `SchemaPlan` per pack. Callers (typically the runtime
    /// bootstrap) apply each plan to the pack's assigned backend. Empty plans
    /// are included so the caller can iterate uniformly; callers that want to
    /// skip empty plans should check `plan.is_empty()`.
    pub fn all_schema_plans(&self) -> Vec<SchemaPlan> {
        self.packs.iter().map(|p| p.schema_plan()).collect()
    }

    /// Invoke `PackRuntime::register_embedders` on every registered pack.
    ///
    /// Called by the transport during startup, after the registry is built and
    /// before the first verb dispatch, so that custom embedding providers
    /// contributed by packs are reachable via `KhiveRuntime::embedder(name)`.
    ///
    /// Packs whose `register_embedders` is the default no-op pay no overhead.
    /// The method is idempotent when the underlying registry uses last-wins
    /// semantics for duplicate provider names.
    pub fn call_register_embedders(&self, runtime: &KhiveRuntime) {
        for pack in self.packs.iter() {
            pack.register_embedders(runtime);
        }
    }

    /// Invoke `PackRuntime::register_entity_type_validator` on every registered pack.
    ///
    /// Called by the transport during startup, after the registry is built and
    /// before the first verb dispatch, so that entity-type validation at the
    /// runtime layer is active for all write paths including direct `create_many`
    /// callers that bypass the handler layer.
    ///
    /// Packs whose `register_entity_type_validator` is the default no-op pay
    /// no overhead.
    pub fn call_register_entity_type_validators(&self, runtime: &KhiveRuntime) {
        for pack in self.packs.iter() {
            pack.register_entity_type_validator(runtime);
        }
    }

    /// Invoke `PackRuntime::warm` on every registered pack.
    /// Called by the daemon at boot (in a background task) so expensive in-memory
    /// state (ANN indexes) is pre-loaded without blocking request serving.
    pub async fn call_warm_all(&self) {
        for pack in self.packs.iter() {
            pack.warm().await;
        }
    }

    /// Resolve the presentation policy for a verb name.
    ///
    /// Walks all registered handlers (including subhandlers) for the first
    /// matching name and returns its declared [`VerbPresentationPolicy`].
    /// Returns `Standard` for unknown verbs — unknown verbs will fail at
    /// dispatch anyway, so the fallback here is safe.
    pub fn presentation_policy_for(&self, verb: &str) -> khive_types::VerbPresentationPolicy {
        for pack in self.packs.iter() {
            if let Some(handler) = pack.handlers().iter().find(|h| h.name == verb) {
                return handler.presentation_policy();
            }
        }
        khive_types::VerbPresentationPolicy::Standard
    }

    /// Returns `true` if the named verb exists and is tagged
    /// `Visibility::Subhandler` (internal / operator-only).
    ///
    /// Used by the MCP server to gate subhandler invocation at the wire
    /// boundary without blocking internal callers that invoke the same verbs
    /// through the runtime directly.
    pub fn is_subhandler_verb(&self, verb: &str) -> bool {
        for pack in self.packs.iter() {
            if let Some(handler) = pack.handlers().iter().find(|h| h.name == verb) {
                return matches!(handler.visibility, Visibility::Subhandler);
            }
        }
        false
    }

    /// Apply all non-empty pack-auxiliary schema plans to the given backend
    /// (c12 startup application).
    ///
    /// This is the centralized startup hook that replaced the previous lazy
    /// per-pack self-bootstrap pattern. Each pack's `SchemaPlan` carries
    /// idempotent `CREATE TABLE IF NOT EXISTS` DDL; calling this more than once
    /// is safe. Empty plans are skipped.
    ///
    /// Errors from individual plans are logged via `tracing::warn!` and not
    /// propagated so that a single pack's schema failure does not prevent the
    /// rest from loading. Callers that need hard-failure semantics should call
    /// `all_schema_plans()` and apply each plan individually.
    pub fn apply_schema_plans(&self, backend: &khive_db::StorageBackend) {
        for plan in self.all_schema_plans() {
            if plan.is_empty() {
                continue;
            }
            if let Err(e) = backend.apply_pack_ddl_statements(plan.statements) {
                tracing::warn!(
                    pack = plan.pack,
                    error = %e,
                    "failed to apply pack schema plan at startup (non-fatal)"
                );
            }
        }
    }

    /// Pack-auxiliary schema plans with their owning pack names.
    ///
    /// Returns `(pack_name, SchemaPlan)` pairs for every registered pack.
    /// Used by the multi-backend boot path to apply each plan to the pack's
    /// assigned backend rather than a single shared backend.
    pub fn all_schema_plans_named(&self) -> Vec<(&'static str, SchemaPlan)> {
        self.packs
            .iter()
            .map(|p| {
                let plan = p.schema_plan();
                (plan.pack, plan)
            })
            .collect()
    }

    /// Apply pack-auxiliary schema plans using a per-pack backend map.
    ///
    /// For each `(pack_name, plan)` returned by `all_schema_plans_named()`,
    /// applies the plan to `backend_for_pack[pack_name]` when present,
    /// falling back to `default_backend` for any pack not in the map.
    ///
    /// Returns an error when two packs on the same backend declare the same
    /// auxiliary table (ADR-028 §7 collision policy: boot failure naming both
    /// packs and the conflicting table).
    ///
    /// This is the multi-backend boot path (ADR-028). Single-backend callers
    /// should continue using [`Self::apply_schema_plans`].
    pub fn apply_schema_plans_with_map(
        &self,
        backend_for_pack: &HashMap<&str, &khive_db::StorageBackend>,
        default_backend: &khive_db::StorageBackend,
    ) -> Result<(), crate::PackSchemaCollisionError> {
        // Track which pack first claimed each table on each backend.
        // Backend identity is the raw pointer of the underlying connection pool Arc.
        let mut claimed: HashMap<(*const (), String), &'static str> = HashMap::new();

        for (pack_name, plan) in self.all_schema_plans_named() {
            if plan.is_empty() {
                continue;
            }
            let backend = backend_for_pack
                .get(pack_name)
                .copied()
                .unwrap_or(default_backend);
            let backend_ptr = std::sync::Arc::as_ptr(&backend.pool_arc()) as *const ();

            // Pre-scan DDL for table names and detect collisions before applying.
            for stmt in plan.statements {
                for table_name in extract_table_names(stmt) {
                    let key = (backend_ptr, table_name.clone());
                    match claimed.entry(key) {
                        std::collections::hash_map::Entry::Vacant(e) => {
                            e.insert(pack_name);
                        }
                        std::collections::hash_map::Entry::Occupied(e) => {
                            let prior_pack = *e.get();
                            return Err(crate::PackSchemaCollisionError {
                                pack_a: prior_pack,
                                pack_b: pack_name,
                                table: table_name,
                            });
                        }
                    }
                }
            }

            backend
                .apply_pack_ddl_statements(plan.statements)
                .map_err(|e| crate::PackSchemaCollisionError {
                    pack_a: pack_name,
                    pack_b: pack_name,
                    table: format!("DDL error: {e}"),
                })?;
        }
        Ok(())
    }
}

// ── Inventory-based dynamic pack loading ────────────────────────────────────

/// Factory for creating pack instances registered via `inventory` at link time.
/// Each pack crate submits a `&'static dyn PackFactory` wrapped in a
/// [`PackRegistration`]; the binary's linker collects them all into a single
/// slice iterable at runtime.
///
/// Implementors must be `Send + Sync + 'static` because the registry is built
/// once and shared across async tasks.
pub trait PackFactory: Send + Sync + 'static {
    /// Canonical lowercase name for this pack (e.g. `"kg"`, `"gtd"`).
    fn name(&self) -> &'static str;

    /// Names of packs that must be loaded before this one.
    ///
    /// Defaults to empty so pack crates that have no dependencies compile
    /// without changes. [`PackRegistry::register_packs`] validates that every
    /// name listed here is present in the caller's explicit pack list — absent
    /// dependencies are a boot error, not silently auto-added.
    fn requires(&self) -> &'static [&'static str] {
        &[]
    }

    /// Create a new pack instance for the given runtime.
    fn create(&self, runtime: KhiveRuntime) -> Box<dyn PackRuntime>;

    /// Optionally create a `PackByIdResolver` for this pack.
    ///
    /// Packs that own private SQL tables implement this to hook into
    /// `get(id)` and `delete(id)`. Defaults to `None` so existing packs
    /// compile without changes.
    fn create_resolver(&self, _runtime: KhiveRuntime) -> Option<Box<dyn PackByIdResolver>> {
        None
    }
}

/// Newtype wrapper collected by `inventory` so pack crates can submit
/// `&'static dyn PackFactory` references without the type-ascription syntax
/// that `inventory::submit!` does not support for bare trait-object references.
pub struct PackRegistration(pub &'static dyn PackFactory);

inventory::collect!(PackRegistration);

/// Error returned by [`PackRegistry::register_packs`] when boot validation fails.
#[derive(Debug)]
pub enum PackLoadError {
    /// The requested pack name was not found in the inventory.
    UnknownPack(String),
    /// A pack was requested but a declared dependency is absent from the list.
    MissingDependency {
        /// The pack that declared the dependency.
        pack: String,
        /// The dependency that is missing from the requested pack list.
        dep: String,
    },
}

impl std::fmt::Display for PackLoadError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            PackLoadError::UnknownPack(name) => write!(f, "unknown pack {name:?}"),
            PackLoadError::MissingDependency { pack, dep } => write!(
                f,
                "pack {pack:?} requires {dep:?}, which is not in the requested pack list; \
                 add --pack {dep} before --pack {pack}"
            ),
        }
    }
}

impl std::error::Error for PackLoadError {}

/// Registry of pack factories discovered via `inventory` at link time.
///
/// No instance is needed — all methods are associated functions that walk the
/// globally-collected [`PackRegistration`] slice.
pub struct PackRegistry;

impl PackRegistry {
    /// Names of all pack factories discovered via `inventory`.
    pub fn discovered_names() -> Vec<&'static str> {
        inventory::iter::<PackRegistration>
            .into_iter()
            .map(|r| r.0.name())
            .collect()
    }

    /// Register the named packs into `builder` using the supplied `runtime`.
    ///
    /// Validates the explicit pack list against `PackFactory::requires()` —
    /// if any requested pack declares a dependency that is absent from `names`,
    /// registration fails (missing dependency is a boot error, not silently
    /// auto-added). Callers must include all required packs explicitly.
    ///
    /// The [`VerbRegistryBuilder::build`] topo-sort enforces correct load order.
    ///
    /// Returns `Ok(())` when all names are recognised and all declared
    /// dependencies are satisfied; returns `Err(PackLoadError)` with a
    /// distinct variant for unknown pack vs missing dependency.
    pub fn register_packs(
        names: &[String],
        runtime: KhiveRuntime,
        builder: &mut VerbRegistryBuilder,
    ) -> Result<(), PackLoadError> {
        // Build a name→factory index once.
        let all: Vec<&'static dyn PackFactory> = inventory::iter::<PackRegistration>
            .into_iter()
            .map(|r| r.0)
            .collect();
        let factory_for = |name: &str| -> Option<&'static dyn PackFactory> {
            all.iter().copied().find(|f| f.name() == name)
        };

        // Validate that every requested name is a known factory.
        let requested: std::collections::HashSet<&str> = names.iter().map(String::as_str).collect();
        for name in names {
            factory_for(name.as_str()).ok_or_else(|| PackLoadError::UnknownPack(name.clone()))?;
        }

        // Validate that all requires() dependencies are explicitly present in
        // the requested set. Missing dep → boot error, not auto-add.
        for name in names {
            let factory = factory_for(name.as_str()).unwrap(); // validated above
            for &dep in factory.requires() {
                if !requested.contains(dep) {
                    return Err(PackLoadError::MissingDependency {
                        pack: name.clone(),
                        dep: dep.to_string(),
                    });
                }
            }
        }

        // Register every requested pack; VerbRegistryBuilder::build()
        // performs the topo-sort, so insertion order here does not matter.
        for name in names {
            let factory = factory_for(name.as_str()).unwrap(); // validated above
            builder.register_boxed(factory.create(runtime.clone()));
            if let Some(resolver) = factory.create_resolver(runtime.clone()) {
                builder.register_resolver(name.clone(), resolver);
            }
        }

        Ok(())
    }

    /// Register the named packs into `builder`, routing each pack to its own runtime.
    ///
    /// `runtimes` maps pack name → `KhiveRuntime` (one per backend assignment).
    /// `default_runtime` is used for any pack whose name is not in `runtimes`.
    /// The validation logic (unknown pack, missing dependency) is identical to
    /// [`PackRegistry::register_packs`].
    ///
    /// This is the multi-backend boot path (ADR-028). Single-backend callers
    /// should continue using [`PackRegistry::register_packs`].
    pub fn register_packs_with_runtimes(
        names: &[String],
        runtimes: &HashMap<String, KhiveRuntime>,
        default_runtime: &KhiveRuntime,
        builder: &mut VerbRegistryBuilder,
    ) -> Result<(), PackLoadError> {
        let all: Vec<&'static dyn PackFactory> = inventory::iter::<PackRegistration>
            .into_iter()
            .map(|r| r.0)
            .collect();
        let factory_for = |name: &str| -> Option<&'static dyn PackFactory> {
            all.iter().copied().find(|f| f.name() == name)
        };

        let requested: std::collections::HashSet<&str> = names.iter().map(String::as_str).collect();
        for name in names {
            factory_for(name.as_str()).ok_or_else(|| PackLoadError::UnknownPack(name.clone()))?;
        }

        for name in names {
            let factory = factory_for(name.as_str()).unwrap();
            for &dep in factory.requires() {
                if !requested.contains(dep) {
                    return Err(PackLoadError::MissingDependency {
                        pack: name.clone(),
                        dep: dep.to_string(),
                    });
                }
            }
        }

        for name in names {
            let factory = factory_for(name.as_str()).unwrap();
            let runtime = runtimes
                .get(name.as_str())
                .cloned()
                .unwrap_or_else(|| default_runtime.clone());
            builder.register_boxed(factory.create(runtime.clone()));
            if let Some(resolver) = factory.create_resolver(runtime) {
                builder.register_resolver(name.clone(), resolver);
            }
        }

        Ok(())
    }
}

fn target_id_from_args(args: &serde_json::Value) -> Option<uuid::Uuid> {
    args.get("target_id")
        .and_then(serde_json::Value::as_str)
        .and_then(|s| s.parse::<uuid::Uuid>().ok())
}

// INLINE TEST JUSTIFICATION: tests here exercise VerbRegistry collision detection,
// gate enforcement, and dispatch ordering that depend on direct access to the
// registry's private `packs` Vec and gate field. Moving them to tests/ would
// require pub-exporting registry internals. Broad behavioral dispatch tests
// live in tests/integration.rs.
#[cfg(test)]
mod tests {
    use super::*;
    use khive_types::Pack;

    struct AlphaPack;

    impl Pack for AlphaPack {
        const NAME: &'static str = "alpha";
        const NOTE_KINDS: &'static [&'static str] = &["memo", "log"];
        const ENTITY_KINDS: &'static [&'static str] = &["widget"];
        const HANDLERS: &'static [HandlerDef] = &[
            HandlerDef {
                name: "create",
                description: "create a widget",
                visibility: Visibility::Verb,
                category: VerbCategory::Commissive,
                params: &[],
            },
            HandlerDef {
                name: "list",
                description: "list widgets",
                visibility: Visibility::Verb,
                category: VerbCategory::Assertive,
                params: &[],
            },
        ];
    }

    #[async_trait]
    impl PackRuntime for AlphaPack {
        fn name(&self) -> &str {
            AlphaPack::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            AlphaPack::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            AlphaPack::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            AlphaPack::HANDLERS
        }
        async fn dispatch(
            &self,
            verb: &str,
            _params: Value,
            _registry: &VerbRegistry,
            _token: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Ok(serde_json::json!({ "pack": "alpha", "verb": verb }))
        }
    }

    struct BetaPack;

    impl Pack for BetaPack {
        const NAME: &'static str = "beta";
        const NOTE_KINDS: &'static [&'static str] = &["alert"];
        const ENTITY_KINDS: &'static [&'static str] = &["widget", "gadget"];
        const HANDLERS: &'static [HandlerDef] = &[
            HandlerDef {
                name: "notify",
                description: "send alert",
                visibility: Visibility::Verb,
                category: VerbCategory::Commissive,
                params: &[],
            },
            // "create" is Subhandler so it does NOT collide with AlphaPack's
            // Verb-visibility "create" — subhandlers are pack-internal and
            // excluded from cross-pack collision detection.
            HandlerDef {
                name: "create",
                description: "beta internal create (subhandler)",
                visibility: Visibility::Subhandler,
                category: VerbCategory::Commissive,
                params: &[],
            },
        ];
    }

    /// Build a registry with AlphaPack + BetaPack.
    ///
    /// BetaPack's `create` is Subhandler so there is no Verb-visibility
    /// collision with AlphaPack's `create` Verb. Tests that need a collision
    /// use `build_colliding_registry()` instead.
    fn build_registry() -> VerbRegistry {
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.register(BetaPack);
        builder.build().expect("registry builds without collision")
    }

    /// Build a registry with two packs that declare the same Verb-visibility
    /// handler — used to test that `VerbCollision` is raised at build time.
    struct CollidingPack;

    impl Pack for CollidingPack {
        const NAME: &'static str = "colliding";
        const NOTE_KINDS: &'static [&'static str] = &[];
        const ENTITY_KINDS: &'static [&'static str] = &[];
        const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
            name: "create",
            description: "duplicate Verb-visibility create",
            visibility: Visibility::Verb,
            category: VerbCategory::Commissive,
            params: &[],
        }];
    }

    #[async_trait]
    impl PackRuntime for CollidingPack {
        fn name(&self) -> &str {
            Self::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            Self::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            Self::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            Self::HANDLERS
        }
        async fn dispatch(
            &self,
            verb: &str,
            _params: Value,
            _registry: &VerbRegistry,
            _token: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Ok(serde_json::json!({ "pack": "colliding", "verb": verb }))
        }
    }

    #[async_trait]
    impl PackRuntime for BetaPack {
        fn name(&self) -> &str {
            BetaPack::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            BetaPack::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            BetaPack::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            BetaPack::HANDLERS
        }
        async fn dispatch(
            &self,
            verb: &str,
            _params: Value,
            _registry: &VerbRegistry,
            _token: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Ok(serde_json::json!({ "pack": "beta", "verb": verb }))
        }
    }

    #[tokio::test]
    async fn dispatch_routes_to_correct_pack() {
        let reg = build_registry();

        let res = reg.dispatch("list", Value::Null).await.unwrap();
        assert_eq!(res["pack"], "alpha");

        let res = reg.dispatch("notify", Value::Null).await.unwrap();
        assert_eq!(res["pack"], "beta");
    }

    /// F093/F094: two packs declaring the same `Visibility::Verb` handler must be
    /// rejected at build time — the old "first registered wins" behaviour is
    /// replaced by a boot error.
    #[test]
    fn verb_collision_is_boot_time_error() {
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.register(CollidingPack);
        let err = builder
            .build()
            .err()
            .expect("duplicate Verb-visibility handler must be rejected at build time");
        assert!(
            matches!(err, RuntimeError::VerbCollision { ref verb, .. } if verb == "create"),
            "expected VerbCollision for 'create', got {err:?}"
        );
        let msg = err.to_string();
        assert!(
            msg.contains("create"),
            "error must name the colliding verb: {msg}"
        );
        assert!(
            msg.contains("alpha") || msg.contains("colliding"),
            "error must name one of the conflicting packs: {msg}"
        );
    }

    /// Subhandler-visibility handlers with the same name across packs are NOT
    /// a collision — they are pack-internal and excluded from cross-pack
    /// collision detection.
    #[test]
    fn subhandler_same_name_across_packs_is_not_a_collision() {
        struct SubhandlerPack;
        impl Pack for SubhandlerPack {
            const NAME: &'static str = "subhandler_pack";
            const NOTE_KINDS: &'static [&'static str] = &[];
            const ENTITY_KINDS: &'static [&'static str] = &[];
            const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
                name: "create",
                description: "internal create",
                visibility: Visibility::Subhandler,
                category: VerbCategory::Commissive,
                params: &[],
            }];
        }
        #[async_trait]
        impl PackRuntime for SubhandlerPack {
            fn name(&self) -> &str {
                Self::NAME
            }
            fn note_kinds(&self) -> &'static [&'static str] {
                Self::NOTE_KINDS
            }
            fn entity_kinds(&self) -> &'static [&'static str] {
                Self::ENTITY_KINDS
            }
            fn handlers(&self) -> &'static [HandlerDef] {
                Self::HANDLERS
            }
            async fn dispatch(
                &self,
                verb: &str,
                _: Value,
                _: &VerbRegistry,
                _: &NamespaceToken,
            ) -> Result<Value, RuntimeError> {
                Ok(serde_json::json!({"pack": "subhandler_pack", "verb": verb}))
            }
        }
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack); // AlphaPack has Verb "create"
        builder.register(SubhandlerPack); // SubhandlerPack has Subhandler "create" — no collision
        builder
            .build()
            .expect("subhandler same name must NOT be a collision");
    }

    #[tokio::test]
    async fn dispatch_unknown_verb_returns_error() {
        let reg = build_registry();

        let err = reg.dispatch("explode", Value::Null).await.unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("explode"));
        assert!(msg.contains("create"));
    }

    /// `all_verbs` returns only `Visibility::Verb` entries (F118).
    ///
    /// BetaPack's `create` is `Visibility::Subhandler` — it must NOT appear
    /// in `all_verbs()` even though it has the same name as a Verb in AlphaPack.
    #[test]
    fn all_verbs_aggregates_across_packs_excludes_subhandlers() {
        let reg = build_registry();
        let verbs: Vec<&str> = reg.all_verbs().iter().map(|v| v.name).collect();
        // BetaPack's "create" (Subhandler) is absent; only Verb-visibility entries appear.
        assert_eq!(verbs, vec!["create", "list", "notify"]);
    }

    #[test]
    fn all_verbs_with_names_pairs_pack_name_excludes_subhandlers() {
        let reg = build_registry();
        let pairs: Vec<(&str, &str)> = reg
            .all_verbs_with_names()
            .iter()
            .map(|(pack, v)| (*pack, v.name))
            .collect();
        // BetaPack's "create" is Subhandler and must NOT appear here.
        assert_eq!(
            pairs,
            vec![("alpha", "create"), ("alpha", "list"), ("beta", "notify"),]
        );
    }

    #[test]
    fn all_handlers_with_names_includes_subhandlers() {
        let reg = build_registry();
        let pairs: Vec<(&str, &str)> = reg
            .all_handlers_with_names()
            .iter()
            .map(|(pack, v)| (*pack, v.name))
            .collect();
        // BetaPack's Subhandler "create" IS present in the full handler list.
        assert_eq!(
            pairs,
            vec![
                ("alpha", "create"),
                ("alpha", "list"),
                ("beta", "notify"),
                ("beta", "create"),
            ]
        );
    }

    #[test]
    fn note_kinds_are_ordered() {
        let reg = build_registry();
        let kinds = reg.all_note_kinds();
        assert_eq!(kinds, vec!["memo", "log", "alert"]);
    }

    #[test]
    fn note_kind_duplicate_rejected_at_build_time() {
        struct DupPack;

        impl khive_types::Pack for DupPack {
            const NAME: &'static str = "dup";
            // "memo" is already declared by AlphaPack — must be rejected at build.
            const NOTE_KINDS: &'static [&'static str] = &["memo"];
            const ENTITY_KINDS: &'static [&'static str] = &[];
            const HANDLERS: &'static [HandlerDef] = &[];
        }

        #[async_trait]
        impl PackRuntime for DupPack {
            fn name(&self) -> &str {
                Self::NAME
            }
            fn note_kinds(&self) -> &'static [&'static str] {
                Self::NOTE_KINDS
            }
            fn entity_kinds(&self) -> &'static [&'static str] {
                Self::ENTITY_KINDS
            }
            fn handlers(&self) -> &'static [HandlerDef] {
                Self::HANDLERS
            }
            async fn dispatch(
                &self,
                _verb: &str,
                _params: Value,
                _registry: &VerbRegistry,
                _token: &NamespaceToken,
            ) -> Result<Value, RuntimeError> {
                Ok(Value::Null)
            }
        }

        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.register(DupPack);
        let err = builder
            .build()
            .err()
            .expect("duplicate note kind must be rejected");
        let msg = err.to_string();
        assert!(
            msg.contains("memo"),
            "error must name the duplicate kind: {msg}"
        );
        assert!(
            msg.contains("alpha") || msg.contains("dup"),
            "error must name one of the conflicting packs: {msg}"
        );
    }

    #[test]
    fn entity_kinds_are_deduplicated() {
        let reg = build_registry();
        let kinds = reg.all_entity_kinds();
        assert_eq!(kinds, vec!["widget", "gadget"]);
    }

    // ---- Gate wiring ----

    use khive_gate::{Gate, GateError};
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;

    #[derive(Default, Debug)]
    struct CountingGate {
        calls: AtomicUsize,
        deny_verb: Option<&'static str>,
    }

    impl Gate for CountingGate {
        fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            if Some(req.verb.as_str()) == self.deny_verb {
                Ok(GateDecision::deny(format!("test deny for {}", req.verb)))
            } else {
                Ok(GateDecision::allow())
            }
        }
    }

    #[tokio::test]
    async fn dispatch_consults_the_gate() {
        let gate = Arc::new(CountingGate::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", Value::Null).await.unwrap();
        reg.dispatch("create", Value::Null).await.unwrap();
        assert_eq!(
            gate.calls.load(Ordering::SeqCst),
            2,
            "gate should be consulted once per dispatch"
        );
    }

    #[tokio::test]
    async fn dispatch_returns_permission_denied_on_deny_v03() {
        let gate = Arc::new(CountingGate {
            calls: AtomicUsize::new(0),
            deny_verb: Some("create"),
        });
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        let reg = builder.build().expect("registry builds");

        // Gate denies — dispatch now returns PermissionDenied (hard enforcement).
        let err = reg.dispatch("create", Value::Null).await.unwrap_err();
        assert!(
            matches!(err, RuntimeError::PermissionDenied { ref verb, .. } if verb == "create"),
            "expected PermissionDenied, got {err:?}"
        );
        let msg = err.to_string();
        assert!(
            msg.contains("create"),
            "error message must name the verb: {msg}"
        );
        assert!(
            msg.contains("test deny for create"),
            "error message must carry the deny reason: {msg}"
        );
        assert_eq!(gate.calls.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn dispatch_allow_verb_succeeds_even_with_deny_gate_for_other_verb() {
        // Deny only "create" — "list" must still work.
        let gate = Arc::new(CountingGate {
            calls: AtomicUsize::new(0),
            deny_verb: Some("create"),
        });
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        let reg = builder.build().expect("registry builds");

        let res = reg.dispatch("list", Value::Null).await.unwrap();
        assert_eq!(res["pack"], "alpha");
    }

    #[tokio::test]
    async fn dispatch_uses_allow_all_gate_by_default() {
        // No `with_gate` call — builder should use `AllowAllGate` so dispatch works.
        let reg = build_registry();
        let res = reg.dispatch("list", Value::Null).await.unwrap();
        assert_eq!(res["pack"], "alpha");
    }

    // Captures the namespace each call sees so we can assert what the gate
    // actually receives — codex round-1 caught us hard-wiring `default_ns()`.
    #[derive(Default, Debug)]
    struct NamespaceCapturingGate {
        seen: std::sync::Mutex<Vec<String>>,
    }

    impl Gate for NamespaceCapturingGate {
        fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
            self.seen
                .lock()
                .unwrap()
                .push(req.namespace.as_str().to_string());
            Ok(GateDecision::allow())
        }
    }

    #[tokio::test]
    async fn dispatch_propagates_params_namespace_to_gate() {
        let gate = Arc::new(NamespaceCapturingGate::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        builder.with_default_namespace("tenant-x");
        let reg = builder.build().expect("registry builds");

        // Explicit namespace in params wins.
        reg.dispatch("list", serde_json::json!({"namespace": "tenant-y"}))
            .await
            .unwrap();
        // Missing namespace → registry default.
        reg.dispatch("list", Value::Null).await.unwrap();
        // Empty string is rejected: Namespace::parse("") fails → InvalidInput error.
        let err = reg
            .dispatch("list", serde_json::json!({"namespace": ""}))
            .await
            .unwrap_err();
        assert!(
            matches!(err, RuntimeError::InvalidInput(_)),
            "empty namespace must return InvalidInput, got {err:?}"
        );

        let seen = gate.seen.lock().unwrap().clone();
        assert_eq!(seen, vec!["tenant-y", "tenant-x"]);
    }

    #[tokio::test]
    async fn dispatch_falls_back_to_local_when_no_default_set() {
        // Builder default mirrors `Namespace::default_ns()`.
        let gate = Arc::new(NamespaceCapturingGate::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", Value::Null).await.unwrap();
        let seen = gate.seen.lock().unwrap().clone();
        assert_eq!(seen, vec!["local"]);
    }

    // ---- Audit event emission ----

    use khive_gate::{AuditDecision, AuditEvent, Obligation};

    /// A gate that records every audit event emitted via from_check.
    #[derive(Default, Debug)]
    struct AuditCapturingGate {
        events: std::sync::Mutex<Vec<AuditEvent>>,
        deny_verb: Option<&'static str>,
    }

    impl Gate for AuditCapturingGate {
        fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
            let decision = if Some(req.verb.as_str()) == self.deny_verb {
                GateDecision::deny("test deny")
            } else {
                GateDecision::allow_with(vec![Obligation::Audit {
                    tag: format!("{}.check", req.verb),
                }])
            };
            // Capture what dispatch will also emit.
            let ev = AuditEvent::from_check(req, &decision, self.impl_name());
            self.events.lock().unwrap().push(ev);
            Ok(decision)
        }

        fn impl_name(&self) -> &'static str {
            "AuditCapturingGate"
        }
    }

    #[tokio::test]
    async fn dispatch_emits_one_audit_event_per_call() {
        let gate = Arc::new(AuditCapturingGate::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", Value::Null).await.unwrap();
        reg.dispatch("create", Value::Null).await.unwrap();

        let evs = gate.events.lock().unwrap();
        assert_eq!(evs.len(), 2, "exactly one audit event per dispatch call");
    }

    #[tokio::test]
    async fn dispatch_audit_event_allow_carries_obligations() {
        let gate = Arc::new(AuditCapturingGate::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", Value::Null).await.unwrap();

        let evs = gate.events.lock().unwrap();
        let ev = &evs[0];
        assert_eq!(ev.verb, "list");
        assert_eq!(ev.decision, AuditDecision::Allow);
        assert!(ev.deny_reason.is_none());
        assert_eq!(ev.obligations.len(), 1);
        assert_eq!(ev.gate_impl, "AuditCapturingGate");
    }

    #[tokio::test]
    async fn dispatch_audit_event_deny_carries_reason() {
        let gate = Arc::new(AuditCapturingGate {
            events: Default::default(),
            deny_verb: Some("create"),
        });
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        let reg = builder.build().expect("registry builds");

        // Gate denies — dispatch returns PermissionDenied (hard enforcement).
        // The audit event is still recorded (captured inside the gate impl).
        let err = reg.dispatch("create", Value::Null).await.unwrap_err();
        assert!(matches!(err, RuntimeError::PermissionDenied { .. }));

        let evs = gate.events.lock().unwrap();
        let ev = &evs[0];
        assert_eq!(ev.verb, "create");
        assert_eq!(ev.decision, AuditDecision::Deny);
        assert_eq!(ev.deny_reason.as_deref(), Some("test deny"));
        assert!(ev.obligations.is_empty());
    }

    #[tokio::test]
    async fn dispatch_audit_event_fields_match_gate_request() {
        let gate = Arc::new(AuditCapturingGate::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        builder.with_default_namespace("tenant-z");
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", serde_json::json!({"namespace": "tenant-q"}))
            .await
            .unwrap();

        let evs = gate.events.lock().unwrap();
        let ev = &evs[0];
        // Namespace from params wins.
        assert_eq!(ev.namespace, "tenant-q");
        assert_eq!(ev.verb, "list");
        assert_eq!(ev.actor.kind, "anonymous");
    }

    // ---- Actor attribution threading into gate request (ADR-057) ----

    /// A gate spy that captures the raw `GateRequest` it receives.
    #[derive(Default, Debug)]
    struct ActorCapturingGate {
        requests: std::sync::Mutex<Vec<GateRequest>>,
    }

    impl Gate for ActorCapturingGate {
        fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
            self.requests.lock().unwrap().push(req.clone());
            Ok(GateDecision::allow())
        }
    }

    /// When `actor_id` is configured, the gate request carries that actor, not
    /// anonymous. This exercises the ADR-057 attribution fix: the gate can
    /// distinguish an agent caller from an unauthenticated caller.
    #[tokio::test]
    async fn gate_request_carries_configured_actor_when_actor_id_is_set() {
        let gate = Arc::new(ActorCapturingGate::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        builder.with_actor_id(Some("team-abc:implementer".to_string()));
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", Value::Null).await.unwrap();

        let reqs = gate.requests.lock().unwrap();
        assert_eq!(reqs.len(), 1);
        let req = &reqs[0];
        assert_eq!(
            req.actor.kind, "actor",
            "gate request must carry kind='actor' when actor_id is configured"
        );
        assert_eq!(
            req.actor.id, "team-abc:implementer",
            "gate request must carry the configured actor id"
        );
    }

    /// When no `actor_id` is configured, the gate request still receives the
    /// anonymous actor (no regression to the party-line default).
    #[tokio::test]
    async fn gate_request_carries_anonymous_when_no_actor_id_configured() {
        let gate = Arc::new(ActorCapturingGate::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate.clone());
        // actor_id left at default (None).
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", Value::Null).await.unwrap();

        let reqs = gate.requests.lock().unwrap();
        assert_eq!(reqs.len(), 1);
        let req = &reqs[0];
        assert_eq!(
            req.actor.kind, "anonymous",
            "gate request must carry anonymous actor when no actor_id is configured"
        );
        assert_eq!(req.actor.id, "local");
    }

    // ---- Rego gate: fail-closed end-to-end (issue #30) ----

    /// A `RegoGate` whose policy lacks the named entrypoint rule must cause
    /// `VerbRegistry::dispatch` to return `RuntimeError::PermissionDenied` —
    /// never to proceed to the pack handler.
    ///
    /// This is the runtime-level assertion for the fix to security issue #30.
    /// `RegoGate::check` converts all evaluation failures (missing rule,
    /// undefined result, serialization error, poisoned engine) to
    /// `Ok(GateDecision::Deny)`, so dispatch is blocked. The runtime's
    /// fail-open `Err(_)` branch remains for non-evaluation gate errors
    /// (e.g. infrastructure faults from other `Gate` implementations).
    #[tokio::test]
    async fn rego_gate_missing_entrypoint_returns_permission_denied() {
        use khive_gate_rego::RegoGate;

        // Policy defines `verdict` but NOT `data.khive.gate.decision` (the
        // default entrypoint).  Construction succeeds — from_policy_str does
        // not validate the default entrypoint.  check() must convert the
        // missing-rule evaluation error to Ok(Deny) so the runtime denies
        // the request rather than treating the Err as a fail-open signal.
        let policy = r#"
            package khive.gate
            import rego.v1
            verdict := "allow"
        "#;
        let gate = Arc::new(RegoGate::from_policy_str(policy).expect("policy compiles"));

        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(gate);
        let reg = builder.build().expect("registry builds");

        let err = reg.dispatch("create", Value::Null).await.unwrap_err();
        assert!(
            matches!(err, RuntimeError::PermissionDenied { ref verb, .. } if verb == "create"),
            "expected PermissionDenied for missing rego entrypoint, got {err:?}"
        );
    }

    // ---- Audit tracing emission ----
    //
    // The AuditCapturingGate tests above prove that AuditEvent::from_check is
    // called with the right inputs, but they observe the event *inside* the
    // gate impl — they would still pass if dispatch's
    // `tracing::info!(audit_event = ..., "gate.check")` were deleted or
    // renamed. The tests below install a capture Layer and assert on the
    // actual tracing event surfaced from dispatch. This locks the public
    // observability contract: one `gate.check` info event per dispatch,
    // carrying an `audit_event` field that round-trips back to an `AuditEvent`.

    use std::sync::{Mutex as StdMutex, Once, OnceLock};

    use serial_test::serial;
    use tracing::field::{Field, Visit};

    #[derive(Clone, Debug, Default)]
    struct CapturedEvent {
        message: Option<String>,
        audit_event: Option<String>,
    }

    #[derive(Default)]
    struct CapturedEventVisitor(CapturedEvent);

    impl Visit for CapturedEventVisitor {
        fn record_str(&mut self, field: &Field, value: &str) {
            match field.name() {
                "message" => self.0.message = Some(value.to_string()),
                "audit_event" => self.0.audit_event = Some(value.to_string()),
                _ => {}
            }
        }

        fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
            // `tracing::info!(audit_event = %expr, "msg")` records via the
            // Display-wrapped Debug path, so we receive the JSON string here.
            // `"msg"` literal records as a `message` field via `record_debug`
            // with a quoted Debug representation; strip the surrounding quotes
            // so the captured message matches the source.
            let formatted = format!("{value:?}");
            let cleaned = formatted
                .trim_start_matches('"')
                .trim_end_matches('"')
                .to_string();
            match field.name() {
                "message" => self.0.message = Some(cleaned),
                "audit_event" => self.0.audit_event = Some(cleaned),
                _ => {}
            }
        }
    }

    /// Minimal `tracing::Subscriber` that captures events into a shared vec.
    ///
    /// Implemented directly (without `tracing_subscriber::registry()` layering)
    /// to avoid the layer machinery that can cause thread-local dispatch to be
    /// bypassed when the registry's internal global state is initialised by
    /// another subscriber in the same test binary.
    ///
    /// Isolation across concurrent tests is handled at the dispatcher level by
    /// `tracing::dispatcher::with_default`, which installs this subscriber
    /// as the thread-local default for the duration of the test closure.
    /// Other threads (e.g. `#[tokio::test]` pool workers) emit through their
    /// own (typically NoSubscriber) dispatchers and never reach this instance.
    struct CaptureSubscriber {
        events: Arc<StdMutex<Vec<CapturedEvent>>>,
    }

    impl CaptureSubscriber {
        fn new(events: Arc<StdMutex<Vec<CapturedEvent>>>) -> Self {
            Self { events }
        }
    }

    impl tracing::Subscriber for CaptureSubscriber {
        fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
            true
        }
        fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
            tracing::span::Id::from_u64(1)
        }
        fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
        fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
        fn event(&self, event: &tracing::Event<'_>) {
            let mut visitor = CapturedEventVisitor::default();
            event.record(&mut visitor);
            self.events.lock().unwrap().push(visitor.0);
        }
        fn enter(&self, _: &tracing::span::Id) {}
        fn exit(&self, _: &tracing::span::Id) {}
    }

    /// Global capture buffer for the tracing tests.
    ///
    /// The subscriber is installed exactly once via `set_global_default`
    /// (thread-local dispatchers via `with_default` proved unreliable when
    /// other tests in the binary configure their own dispatchers in parallel —
    /// the global state interacted unpredictably and events were lost).
    ///
    /// Each test that uses this buffer is `#[serial]`, so only one
    /// runs at a time. The buffer is cleared at the start of each capture call.
    static GLOBAL_CAPTURE: OnceLock<Arc<StdMutex<Vec<CapturedEvent>>>> = OnceLock::new();
    static GLOBAL_INIT: Once = Once::new();

    fn global_capture() -> Arc<StdMutex<Vec<CapturedEvent>>> {
        GLOBAL_INIT.call_once(|| {
            let buffer = Arc::new(StdMutex::new(Vec::new()));
            let subscriber = CaptureSubscriber::new(Arc::clone(&buffer));
            // Ignore error: if another subscriber is already set globally, our
            // subscriber installation fails, but the buffer will simply stay
            // empty and tests will fail with a clear "got 0 events" message
            // rather than a silent corruption.
            let _ = tracing::subscriber::set_global_default(subscriber);
            let _ = GLOBAL_CAPTURE.set(buffer);
        });
        Arc::clone(GLOBAL_CAPTURE.get().expect("global capture initialized"))
    }

    /// Run an async block under the global capture subscriber and return
    /// the events emitted during the run. Clears the buffer at the start.
    ///
    /// Callers MUST be `#[serial]` to prevent concurrent buffer pollution.
    fn capture_dispatch_events<Fut>(future: Fut) -> Vec<CapturedEvent>
    where
        Fut: std::future::Future<Output = ()>,
    {
        let buffer = global_capture();
        buffer.lock().unwrap().clear();

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("build current-thread tokio runtime");
        rt.block_on(future);

        let result = buffer.lock().unwrap().clone();
        result
    }

    /// Pull every captured event whose `message` matches `"gate.check"` AND
    /// whose audit_event JSON declares the expected `gate_impl` name.
    ///
    /// Filtering by `gate_impl` lets concurrent tests in the same binary
    /// emit their own gate.check events into the global capture buffer
    /// without polluting each others' counts.
    fn gate_check_events_for(events: &[CapturedEvent], gate_impl: &str) -> Vec<CapturedEvent> {
        events
            .iter()
            .filter(|e| e.message.as_deref() == Some("gate.check"))
            .filter(|e| {
                e.audit_event
                    .as_deref()
                    .and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok())
                    .and_then(|v| {
                        v.get("gate_impl")
                            .and_then(|g| g.as_str().map(|s| s.to_string()))
                    })
                    .as_deref()
                    == Some(gate_impl)
            })
            .cloned()
            .collect()
    }

    #[test]
    #[serial]
    fn dispatch_tracing_emits_one_gate_check_event_on_allow() {
        #[derive(Debug)]
        struct TracingAllowGate;
        impl Gate for TracingAllowGate {
            fn check(&self, _: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::allow())
            }
            fn impl_name(&self) -> &'static str {
                "TracingAllowGate"
            }
        }

        let events = capture_dispatch_events(async {
            let mut builder = VerbRegistryBuilder::new();
            builder.register(AlphaPack);
            builder.with_gate(Arc::new(TracingAllowGate));
            builder.with_default_namespace("tenant-default");
            let reg = builder.build().expect("registry builds");
            reg.dispatch("list", serde_json::json!({"namespace": "tenant-q"}))
                .await
                .unwrap();
        });

        let gate_events = gate_check_events_for(&events, "TracingAllowGate");
        assert_eq!(
            gate_events.len(),
            1,
            "exactly one gate.check tracing event per dispatch (allow); got {gate_events:?}"
        );
        let payload = gate_events[0]
            .audit_event
            .as_ref()
            .expect("gate.check event must carry an audit_event field");
        let audit: khive_gate::AuditEvent =
            serde_json::from_str(payload).expect("audit_event payload must decode to AuditEvent");
        assert_eq!(audit.decision, AuditDecision::Allow);
        assert_eq!(audit.verb, "list");
        assert_eq!(audit.namespace, "tenant-q");
        assert_eq!(audit.gate_impl, "TracingAllowGate");
        assert!(
            audit.deny_reason.is_none(),
            "deny_reason must be None on Allow"
        );
    }

    // ---- Hard enforcement + EventStore persistence ----

    use crate::runtime::NamespaceToken;
    use async_trait::async_trait;
    use khive_storage::{
        BatchWriteSummary, Event, EventFilter, EventStore, Page, PageRequest, SubstrateKind,
    };
    use khive_types::EventOutcome;

    /// In-memory EventStore for unit tests — avoids file-backed SQLite.
    #[derive(Default, Debug)]
    struct MemoryEventStore {
        events: std::sync::Mutex<Vec<Event>>,
    }

    #[async_trait]
    impl EventStore for MemoryEventStore {
        async fn append_event(&self, event: Event) -> khive_storage::StorageResult<()> {
            self.events.lock().unwrap().push(event);
            Ok(())
        }
        async fn append_events(
            &self,
            events: Vec<Event>,
        ) -> khive_storage::StorageResult<BatchWriteSummary> {
            let attempted = events.len() as u64;
            let affected = attempted;
            self.events.lock().unwrap().extend(events);
            Ok(BatchWriteSummary {
                attempted,
                affected,
                failed: 0,
                first_error: String::new(),
            })
        }
        async fn get_event(&self, id: uuid::Uuid) -> khive_storage::StorageResult<Option<Event>> {
            Ok(self
                .events
                .lock()
                .unwrap()
                .iter()
                .find(|e| e.id == id)
                .cloned())
        }
        async fn query_events(
            &self,
            _filter: EventFilter,
            _page: PageRequest,
        ) -> khive_storage::StorageResult<Page<Event>> {
            let items = self.events.lock().unwrap().clone();
            let total = items.len() as u64;
            Ok(Page {
                items,
                total: Some(total),
            })
        }
        async fn count_events(&self, _filter: EventFilter) -> khive_storage::StorageResult<u64> {
            Ok(self.events.lock().unwrap().len() as u64)
        }
    }

    #[tokio::test]
    async fn allow_all_gate_default_remains_backward_compatible() {
        // No gate set — AllowAllGate is the default. Dispatch must succeed.
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        let reg = builder.build().expect("registry builds");

        let res = reg.dispatch("list", Value::Null).await.unwrap();
        assert_eq!(
            res["pack"], "alpha",
            "AllowAllGate must allow every verb — backward compat guarantee"
        );
        let res = reg.dispatch("create", Value::Null).await.unwrap();
        assert_eq!(res["pack"], "alpha");
    }

    #[tokio::test]
    async fn deny_gate_returns_permission_denied_pack_never_invoked() {
        #[derive(Debug)]
        struct AlwaysDenyGate;
        impl Gate for AlwaysDenyGate {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::deny("test: always deny"))
            }
        }

        // Track whether dispatch was ever invoked on the pack.
        #[derive(Debug)]
        struct TrackedPack {
            invoked: Arc<AtomicUsize>,
        }

        impl khive_types::Pack for TrackedPack {
            const NAME: &'static str = "tracked";
            const NOTE_KINDS: &'static [&'static str] = &[];
            const ENTITY_KINDS: &'static [&'static str] = &[];
            const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
                name: "guarded",
                description: "a guarded verb",
                visibility: Visibility::Verb,
                category: VerbCategory::Assertive,
                params: &[],
            }];
        }

        #[async_trait]
        impl PackRuntime for TrackedPack {
            fn name(&self) -> &str {
                Self::NAME
            }
            fn note_kinds(&self) -> &'static [&'static str] {
                Self::NOTE_KINDS
            }
            fn entity_kinds(&self) -> &'static [&'static str] {
                Self::ENTITY_KINDS
            }
            fn handlers(&self) -> &'static [HandlerDef] {
                Self::HANDLERS
            }
            async fn dispatch(
                &self,
                _verb: &str,
                _params: Value,
                _registry: &VerbRegistry,
                _token: &NamespaceToken,
            ) -> Result<Value, RuntimeError> {
                self.invoked.fetch_add(1, Ordering::SeqCst);
                Ok(serde_json::json!({"invoked": true}))
            }
        }

        let invoked = Arc::new(AtomicUsize::new(0));
        let mut builder = VerbRegistryBuilder::new();
        builder.register(TrackedPack {
            invoked: invoked.clone(),
        });
        builder.with_gate(Arc::new(AlwaysDenyGate));
        let reg = builder.build().expect("registry builds");

        let err = reg.dispatch("guarded", Value::Null).await.unwrap_err();
        assert!(
            matches!(err, RuntimeError::PermissionDenied { ref verb, ref reason } if verb == "guarded" && reason.contains("always deny")),
            "expected PermissionDenied with verb=guarded and reason, got: {err:?}"
        );
        assert_eq!(
            invoked.load(Ordering::SeqCst),
            0,
            "pack dispatch MUST NOT be invoked when gate denies"
        );
    }

    #[tokio::test]
    async fn audit_event_persists_to_event_store_on_allow() {
        let store = Arc::new(MemoryEventStore::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_event_store(store.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
            .await
            .unwrap();

        let count = store.count_events(EventFilter::default()).await.unwrap();
        assert_eq!(count, 1, "one audit event persisted to EventStore on allow");

        let page = store
            .query_events(
                EventFilter::default(),
                PageRequest {
                    limit: 10,
                    offset: 0,
                },
            )
            .await
            .unwrap();
        let ev = &page.items[0];
        assert_eq!(ev.verb, "list");
        assert_eq!(ev.namespace, "test-ns");
        assert_eq!(ev.substrate, SubstrateKind::Event);
        assert_eq!(ev.outcome, EventOutcome::Success);
    }

    #[tokio::test]
    async fn audit_event_persists_to_event_store_on_deny() {
        #[derive(Debug)]
        struct AlwaysDenyGate;
        impl Gate for AlwaysDenyGate {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::deny("denied by test"))
            }
        }

        let store = Arc::new(MemoryEventStore::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(Arc::new(AlwaysDenyGate));
        builder.with_event_store(store.clone());
        let reg = builder.build().expect("registry builds");

        // Hard enforce → PermissionDenied returned.
        let err = reg
            .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
            .await
            .unwrap_err();
        assert!(matches!(err, RuntimeError::PermissionDenied { .. }));

        let count = store.count_events(EventFilter::default()).await.unwrap();
        assert_eq!(count, 1, "one audit event persisted to EventStore on deny");

        let page = store
            .query_events(
                EventFilter::default(),
                PageRequest {
                    limit: 10,
                    offset: 0,
                },
            )
            .await
            .unwrap();
        let ev = &page.items[0];
        assert_eq!(ev.verb, "list");
        assert_eq!(ev.outcome, EventOutcome::Denied);
    }

    #[tokio::test]
    async fn gate_error_does_not_persist_to_event_store() {
        #[derive(Debug)]
        struct FailingGate;
        impl Gate for FailingGate {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, khive_gate::GateError> {
                Err(khive_gate::GateError::Internal("gate broken".into()))
            }
        }

        let store = Arc::new(MemoryEventStore::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(Arc::new(FailingGate));
        builder.with_event_store(store.clone());
        let reg = builder.build().expect("registry builds");

        // Gate Err → fail-open, dispatch proceeds.
        let res = reg.dispatch("list", Value::Null).await.unwrap();
        assert_eq!(
            res["pack"], "alpha",
            "gate error must fail-open, not block dispatch"
        );

        let count = store.count_events(EventFilter::default()).await.unwrap();
        assert_eq!(
            count, 0,
            "gate infrastructure error must NOT produce an audit event in EventStore"
        );
    }

    #[tokio::test]
    async fn no_event_store_configured_tracing_only() {
        // When no event_store is configured, dispatch must succeed without error.
        // (The tracing path is exercised in the tracing tests above; here we just
        // verify the absence of event_store does not break dispatch.)
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        let reg = builder.build().expect("registry builds");

        let res = reg.dispatch("list", Value::Null).await.unwrap();
        assert_eq!(res["pack"], "alpha");
    }

    #[test]
    #[serial]
    fn dispatch_tracing_emits_gate_check_event_with_deny_payload() {
        #[derive(Debug)]
        struct TracingDenyGate;
        impl Gate for TracingDenyGate {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::deny("denied by test gate"))
            }
            fn impl_name(&self) -> &'static str {
                "TracingDenyGate"
            }
        }

        let events = capture_dispatch_events(async {
            let mut builder = VerbRegistryBuilder::new();
            builder.register(AlphaPack);
            builder.with_gate(Arc::new(TracingDenyGate));
            let reg = builder.build().expect("registry builds");
            // Hard enforcement — dispatch returns PermissionDenied on Deny.
            // The tracing audit event is still emitted before the error is returned.
            let _ = reg.dispatch("create", serde_json::Value::Null).await;
        });

        let gate_events = gate_check_events_for(&events, "TracingDenyGate");
        assert_eq!(
            gate_events.len(),
            1,
            "exactly one gate.check tracing event per dispatch (deny); got {gate_events:?}"
        );
        let payload = gate_events[0]
            .audit_event
            .as_ref()
            .expect("gate.check event must carry an audit_event field on Deny");
        let audit: khive_gate::AuditEvent =
            serde_json::from_str(payload).expect("audit_event payload must decode to AuditEvent");
        assert_eq!(audit.decision, AuditDecision::Deny);
        assert_eq!(audit.deny_reason.as_deref(), Some("denied by test gate"));
        assert_eq!(audit.gate_impl, "TracingDenyGate");
        // Wire-shape rule: obligations is always serialized as an array, empty
        // on Deny. Round-trip back through serde_json::Value to confirm the
        // field exists on the wire and is `[]`, not missing.
        let payload_json: serde_json::Value =
            serde_json::from_str(payload).expect("payload must be valid JSON");
        assert_eq!(
            payload_json["obligations"],
            serde_json::Value::Array(Vec::new()),
            "obligations must be `[]` on Deny on the tracing payload, not omitted"
        );
    }

    // ---- EventStore audit envelope round-trip ----
    //
    // Codex review finding (Major #1): EventStore was persisting a summary
    // Event without the full AuditEvent fields (deny_reason, gate_impl,
    // obligations). This test verifies the complete envelope survives
    // append_event → query_events.

    #[tokio::test]
    async fn audit_envelope_round_trips_deny_reason_and_gate_impl_through_event_store() {
        #[derive(Debug)]
        struct DenyGateWithName;
        impl Gate for DenyGateWithName {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::deny("policy: write forbidden for anon"))
            }
            fn impl_name(&self) -> &'static str {
                "DenyGateWithName"
            }
        }

        let store = Arc::new(MemoryEventStore::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(Arc::new(DenyGateWithName));
        builder.with_event_store(store.clone());
        let reg = builder.build().expect("registry builds");

        // Dispatch is denied — PermissionDenied returned.
        let err = reg
            .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
            .await
            .unwrap_err();
        assert!(
            matches!(err, RuntimeError::PermissionDenied { .. }),
            "expected PermissionDenied, got {err:?}"
        );

        // Exactly one event in the store.
        let page = store
            .query_events(
                EventFilter::default(),
                PageRequest {
                    limit: 10,
                    offset: 0,
                },
            )
            .await
            .unwrap();
        assert_eq!(
            page.items.len(),
            1,
            "one audit event must be persisted on deny"
        );

        let ev = &page.items[0];
        assert_eq!(ev.outcome, EventOutcome::Denied);

        // The payload field must hold the full AuditEvent envelope.
        let data = &ev.payload;

        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
            .expect("Event.payload must deserialize to AuditEvent");

        assert_eq!(
            audit.deny_reason.as_deref(),
            Some("policy: write forbidden for anon"),
            "deny_reason must be preserved through EventStore"
        );
        assert_eq!(
            audit.gate_impl, "DenyGateWithName",
            "gate_impl must be preserved through EventStore"
        );
        assert_eq!(
            audit.decision,
            khive_gate::AuditDecision::Deny,
            "decision field must be preserved through EventStore"
        );
    }

    #[tokio::test]
    async fn audit_envelope_round_trips_obligations_through_event_store() {
        use khive_gate::Obligation;

        #[derive(Debug)]
        struct ObligationGate;
        impl Gate for ObligationGate {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::allow_with(vec![Obligation::Audit {
                    tag: "billing.meter".into(),
                }]))
            }
            fn impl_name(&self) -> &'static str {
                "ObligationGate"
            }
        }

        let store = Arc::new(MemoryEventStore::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(Arc::new(ObligationGate));
        builder.with_event_store(store.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
            .await
            .unwrap();

        let page = store
            .query_events(
                EventFilter::default(),
                PageRequest {
                    limit: 10,
                    offset: 0,
                },
            )
            .await
            .unwrap();
        assert_eq!(page.items.len(), 1);

        let ev = &page.items[0];
        assert_eq!(ev.outcome, EventOutcome::Success);

        let data = &ev.payload;

        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
            .expect("Event.payload must deserialize to AuditEvent");

        assert_eq!(audit.gate_impl, "ObligationGate");
        assert_eq!(
            audit.obligations.len(),
            1,
            "obligations must be preserved through EventStore"
        );
        match &audit.obligations[0] {
            Obligation::Audit { tag } => assert_eq!(tag, "billing.meter"),
            other => panic!("expected Audit obligation, got {other:?}"),
        }
    }

    // ---- SQL-backed audit envelope round-trip (codex r2) ----
    //
    // The two tests above use MemoryEventStore (no serialization). This test
    // wires the production SqlEventStore via KhiveRuntime::memory() to verify
    // that the full AuditEvent envelope survives the SQL text→parse round-trip
    // (Event.data is stored as TEXT and parsed back on read).

    #[tokio::test]
    async fn sql_backed_audit_envelope_round_trips_deny_reason_gate_impl_and_obligations() {
        #[derive(Debug)]
        struct SqlTestDenyGate;
        impl Gate for SqlTestDenyGate {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::deny("sql-path: write denied"))
            }
            fn impl_name(&self) -> &'static str {
                "SqlTestDenyGate"
            }
        }

        // KhiveRuntime::memory() creates an in-memory SQLite pool (is_file_backed=false).
        // events_for_namespace ensures the events schema and returns a SqlEventStore
        // scoped to "test-ns". The pool is shared so reads and writes see the same data.
        let rt = KhiveRuntime::memory().expect("in-memory runtime");
        let test_tok = NamespaceToken::for_namespace(Namespace::parse("test-ns").unwrap());
        let sql_store = rt
            .events(&test_tok)
            .expect("events_for_namespace must succeed");

        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(Arc::new(SqlTestDenyGate));
        builder.with_event_store(sql_store.clone());
        let reg = builder.build().expect("registry builds");

        // Dispatch is denied — PermissionDenied returned.
        let err = reg
            .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
            .await
            .unwrap_err();
        assert!(
            matches!(err, RuntimeError::PermissionDenied { .. }),
            "expected PermissionDenied, got {err:?}"
        );

        // Query via the same SqlEventStore — this is the SQL read path.
        let page = sql_store
            .query_events(
                EventFilter::default(),
                PageRequest {
                    limit: 10,
                    offset: 0,
                },
            )
            .await
            .unwrap();
        assert_eq!(
            page.items.len(),
            1,
            "one audit event must be persisted on deny through SqlEventStore"
        );

        let ev = &page.items[0];
        assert_eq!(ev.outcome, EventOutcome::Denied);

        // Event.payload must hold the full AuditEvent serialized as JSON text and
        // parsed back. If the SQL path was lossy, this deserialization would fail
        // or the field assertions below would fail.
        let data = &ev.payload;

        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
            .expect("Event.payload must deserialize to AuditEvent after SQL round-trip");

        assert_eq!(
            audit.deny_reason.as_deref(),
            Some("sql-path: write denied"),
            "deny_reason must survive the SQL text round-trip"
        );
        assert_eq!(
            audit.gate_impl, "SqlTestDenyGate",
            "gate_impl must survive the SQL text round-trip"
        );
        assert_eq!(
            audit.decision,
            khive_gate::AuditDecision::Deny,
            "decision field must survive the SQL text round-trip"
        );
        // obligations is [] on a Deny gate (no obligations returned).
        // Verify the field is present and empty after SQL round-trip.
        assert!(
            audit.obligations.is_empty(),
            "obligations must be preserved as empty [] through SQL round-trip"
        );
    }

    // ---- SQL-backed audit envelope: non-empty obligations survive round-trip ----
    //
    // Codex r3 identified a blind spot: the deny-path SQL test above only
    // asserts obligations == [], which passes even if the SQL path drops the
    // field entirely (AuditEvent.obligations has #[serde(default)]).
    //
    // This test installs an allow-path gate that returns a non-empty obligations
    // vec. After dispatch, the same SqlEventStore is queried and both layers are
    // checked:
    //   1. Raw Event.data["obligations"] is a non-empty JSON array.
    //   2. Deserialized AuditEvent.obligations[0] matches the expected variant.
    #[tokio::test]
    async fn sql_backed_audit_envelope_round_trips_non_empty_obligations() {
        use khive_gate::Obligation;

        #[derive(Debug)]
        struct SqlTestAllowWithObligationGate;
        impl Gate for SqlTestAllowWithObligationGate {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::allow_with(vec![Obligation::Audit {
                    tag: "sql-path-billing.meter".into(),
                }]))
            }
            fn impl_name(&self) -> &'static str {
                "SqlTestAllowWithObligationGate"
            }
        }

        let rt = KhiveRuntime::memory().expect("in-memory runtime");
        let test_tok = NamespaceToken::for_namespace(Namespace::parse("test-ns").unwrap());
        let sql_store = rt
            .events(&test_tok)
            .expect("events_for_namespace must succeed");

        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_gate(Arc::new(SqlTestAllowWithObligationGate));
        builder.with_event_store(sql_store.clone());
        let reg = builder.build().expect("registry builds");

        // Dispatch succeeds — the gate allows with obligations.
        reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
            .await
            .expect("dispatch must succeed when gate allows");

        // Query via the same SqlEventStore — this is the SQL read path.
        let page = sql_store
            .query_events(
                EventFilter::default(),
                PageRequest {
                    limit: 10,
                    offset: 0,
                },
            )
            .await
            .unwrap();
        assert_eq!(
            page.items.len(),
            1,
            "one audit event must be persisted on allow through SqlEventStore"
        );

        let ev = &page.items[0];
        assert_eq!(ev.outcome, EventOutcome::Success);

        let data = &ev.payload;

        // Layer 1: raw JSON check — obligations must be a non-empty array in
        // the persisted TEXT. If the SQL path dropped the field, the default
        // #[serde(default)] would silently deserialize it to [], so we verify
        // the raw JSON before deserializing.
        let obligations_raw = data
            .get("obligations")
            .expect("Event.data JSON must contain 'obligations' key");
        let obligations_arr = obligations_raw
            .as_array()
            .expect("'obligations' must be a JSON array");
        assert!(
            !obligations_arr.is_empty(),
            "raw Event.data['obligations'] must be non-empty after SQL round-trip"
        );

        // Layer 2: deserialized AuditEvent check — the obligation variant and
        // payload must survive the text round-trip faithfully.
        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
            .expect("Event.data must deserialize to AuditEvent after SQL round-trip");

        assert_eq!(
            audit.gate_impl, "SqlTestAllowWithObligationGate",
            "gate_impl must survive the SQL text round-trip"
        );
        assert_eq!(
            audit.decision,
            khive_gate::AuditDecision::Allow,
            "decision field must survive the SQL text round-trip"
        );
        assert_eq!(
            audit.obligations.len(),
            1,
            "obligations must be non-empty after SQL round-trip (not silently defaulted to [])"
        );
        match &audit.obligations[0] {
            Obligation::Audit { tag } => assert_eq!(
                tag, "sql-path-billing.meter",
                "Audit obligation tag must survive the SQL text round-trip"
            ),
            other => panic!("expected Audit obligation, got {other:?}"),
        }
    }

    // ---- Audit payload shape for 'create' verb dispatch ----
    //
    // The previous audit tests verify the envelope shape for the 'list' verb.
    // This test dispatches 'create' (matching the create_note + annotates path)
    // and verifies that ev.verb, ev.outcome, and ev.data all round-trip correctly
    // through the EventStore. Ensures the wire shape is independent of which verb
    // triggers the gate check.
    #[tokio::test]
    async fn audit_event_payload_shape_for_create_verb() {
        let store = Arc::new(MemoryEventStore::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_event_store(store.clone());
        builder.with_default_namespace("test-ns");
        let reg = builder.build().expect("registry builds");

        // Dispatch 'create' — AlphaPack returns a stub value; what matters is
        // the EventStore entry emitted by the registry's gate-check path.
        reg.dispatch("create", serde_json::json!({"namespace": "test-ns"}))
            .await
            .unwrap();

        let count = store.count_events(EventFilter::default()).await.unwrap();
        assert_eq!(count, 1, "exactly one audit event for one dispatch");

        let page = store
            .query_events(
                EventFilter::default(),
                PageRequest {
                    limit: 10,
                    offset: 0,
                },
            )
            .await
            .unwrap();
        let ev = &page.items[0];

        // Top-level Event fields.
        assert_eq!(ev.verb, "create", "ev.verb must be the dispatched verb");
        assert_eq!(
            ev.outcome,
            EventOutcome::Success,
            "ev.outcome must be Success on allow"
        );
        assert_eq!(
            ev.namespace, "test-ns",
            "ev.namespace must match the dispatch namespace"
        );

        // ev.payload must hold the full AuditEvent envelope.
        let data = &ev.payload;

        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
            .expect("ev.payload must deserialize to AuditEvent");

        assert_eq!(
            audit.decision,
            khive_gate::AuditDecision::Allow,
            "AuditEvent.decision must be Allow"
        );
        assert_eq!(audit.verb, "create", "AuditEvent.verb must be 'create'");
        assert_eq!(
            audit.namespace, "test-ns",
            "AuditEvent.namespace must be preserved"
        );
        assert_eq!(
            audit.gate_impl, "AllowAllGate",
            "AuditEvent.gate_impl must name the gate implementation"
        );
        assert!(
            audit.deny_reason.is_none(),
            "AuditEvent.deny_reason must be None on Allow"
        );
        // Wire-shape check: obligations serializes as [] on AllowAllGate.
        let payload_json: serde_json::Value =
            serde_json::from_value(data.clone()).expect("data must be valid JSON");
        assert_eq!(
            payload_json["obligations"],
            serde_json::Value::Array(Vec::new()),
            "obligations must be [] on AllowAllGate"
        );
    }

    // #282: registry audit event must carry target_id when dispatch params include it.
    #[tokio::test]
    async fn audit_event_threads_target_id_from_dispatch_args() {
        let store = Arc::new(MemoryEventStore::default());
        let target = uuid::Uuid::new_v4();
        let mut builder = VerbRegistryBuilder::new();
        builder.register(AlphaPack);
        builder.with_event_store(store.clone());
        builder.with_default_namespace("test-ns");
        let reg = builder.build().expect("registry builds");

        reg.dispatch(
            "create",
            serde_json::json!({"namespace": "test-ns", "target_id": target}),
        )
        .await
        .unwrap();

        let page = store
            .query_events(
                EventFilter::default(),
                PageRequest {
                    offset: 0,
                    limit: 10,
                },
            )
            .await
            .unwrap();
        assert_eq!(
            page.items[0].target_id,
            Some(target),
            "#282: audit event must carry target_id from dispatch params"
        );
    }
}

// ---- Inter-pack dependency checking ----

#[cfg(test)]
mod dep_tests {
    use super::*;
    use async_trait::async_trait;
    use khive_types::Pack;
    use serde_json::Value;

    struct KgDepPack;
    struct MemoryDepPack;
    struct ADepPack;
    struct BDepPack;

    impl Pack for KgDepPack {
        const NAME: &'static str = "kg_dep";
        const NOTE_KINDS: &'static [&'static str] = &["observation"];
        const ENTITY_KINDS: &'static [&'static str] = &["concept"];
        const HANDLERS: &'static [HandlerDef] = &[];
    }

    impl Pack for MemoryDepPack {
        const NAME: &'static str = "memory_dep";
        const NOTE_KINDS: &'static [&'static str] = &["memory"];
        const ENTITY_KINDS: &'static [&'static str] = &[];
        const HANDLERS: &'static [HandlerDef] = &[];
        const REQUIRES: &'static [&'static str] = &["kg_dep"];
    }

    impl Pack for ADepPack {
        const NAME: &'static str = "pack_a";
        const NOTE_KINDS: &'static [&'static str] = &[];
        const ENTITY_KINDS: &'static [&'static str] = &[];
        const HANDLERS: &'static [HandlerDef] = &[];
        const REQUIRES: &'static [&'static str] = &["pack_b"];
    }

    impl Pack for BDepPack {
        const NAME: &'static str = "pack_b";
        const NOTE_KINDS: &'static [&'static str] = &[];
        const ENTITY_KINDS: &'static [&'static str] = &[];
        const HANDLERS: &'static [HandlerDef] = &[];
        const REQUIRES: &'static [&'static str] = &["pack_a"];
    }

    #[async_trait]
    impl PackRuntime for KgDepPack {
        fn name(&self) -> &str {
            Self::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            Self::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            Self::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            Self::HANDLERS
        }
        async fn dispatch(
            &self,
            verb: &str,
            _: Value,
            _: &VerbRegistry,
            _: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Err(RuntimeError::InvalidInput(format!(
                "KgDepPack has no verbs: {verb}"
            )))
        }
    }

    #[async_trait]
    impl PackRuntime for MemoryDepPack {
        fn name(&self) -> &str {
            Self::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            Self::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            Self::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            Self::HANDLERS
        }
        fn requires(&self) -> &'static [&'static str] {
            Self::REQUIRES
        }
        async fn dispatch(
            &self,
            verb: &str,
            _: Value,
            _: &VerbRegistry,
            _: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Err(RuntimeError::InvalidInput(format!(
                "MemoryDepPack has no verbs: {verb}"
            )))
        }
    }

    #[async_trait]
    impl PackRuntime for ADepPack {
        fn name(&self) -> &str {
            Self::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            Self::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            Self::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            Self::HANDLERS
        }
        fn requires(&self) -> &'static [&'static str] {
            Self::REQUIRES
        }
        async fn dispatch(
            &self,
            verb: &str,
            _: Value,
            _: &VerbRegistry,
            _: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Err(RuntimeError::InvalidInput(format!(
                "ADepPack has no verbs: {verb}"
            )))
        }
    }

    #[async_trait]
    impl PackRuntime for BDepPack {
        fn name(&self) -> &str {
            Self::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            Self::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            Self::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            Self::HANDLERS
        }
        fn requires(&self) -> &'static [&'static str] {
            Self::REQUIRES
        }
        async fn dispatch(
            &self,
            verb: &str,
            _: Value,
            _: &VerbRegistry,
            _: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Err(RuntimeError::InvalidInput(format!(
                "BDepPack has no verbs: {verb}"
            )))
        }
    }

    #[test]
    fn test_pack_deps_happy_path() {
        let mut builder = VerbRegistryBuilder::new();
        builder.register(MemoryDepPack);
        builder.register(KgDepPack);
        let reg = builder
            .build()
            .expect("kg_dep satisfies memory_dep dependency");
        assert_eq!(reg.pack_requires("memory_dep").unwrap(), &["kg_dep"]);
        let names = reg.pack_names();
        let kg_pos = names.iter().position(|&n| n == "kg_dep").unwrap();
        let mem_pos = names.iter().position(|&n| n == "memory_dep").unwrap();
        assert!(
            kg_pos < mem_pos,
            "kg_dep must be loaded before memory_dep; order: {names:?}"
        );
    }

    #[test]
    fn test_pack_deps_missing() {
        let mut builder = VerbRegistryBuilder::new();
        builder.register(MemoryDepPack);
        let err = match builder.build() {
            Ok(_) => panic!("expected Err, got Ok"),
            Err(e) => e,
        };
        assert!(
            matches!(err, RuntimeError::MissingPackDependency(_)),
            "expected MissingPackDependency, got {err:?}"
        );
        let msg = err.to_string();
        assert!(
            msg.contains("memory_dep"),
            "error must name the dependent pack: {msg}"
        );
        assert!(
            msg.contains("kg_dep"),
            "error must name the missing dep: {msg}"
        );
    }

    #[test]
    fn test_pack_deps_circular() {
        let mut builder = VerbRegistryBuilder::new();
        builder.register(ADepPack);
        builder.register(BDepPack);
        let err = match builder.build() {
            Ok(_) => panic!("expected Err, got Ok"),
            Err(e) => e,
        };
        assert!(
            matches!(err, RuntimeError::CircularPackDependency(_)),
            "expected CircularPackDependency, got {err:?}"
        );
        let msg = err.to_string();
        assert!(msg.contains("pack_a"), "error must name pack_a: {msg}");
        assert!(msg.contains("pack_b"), "error must name pack_b: {msg}");
    }

    #[test]
    fn test_pack_deps_no_deps() {
        struct NoDepsA;
        struct NoDepsB;

        impl Pack for NoDepsA {
            const NAME: &'static str = "no_deps_a";
            const NOTE_KINDS: &'static [&'static str] = &[];
            const ENTITY_KINDS: &'static [&'static str] = &[];
            const HANDLERS: &'static [HandlerDef] = &[];
        }

        impl Pack for NoDepsB {
            const NAME: &'static str = "no_deps_b";
            const NOTE_KINDS: &'static [&'static str] = &[];
            const ENTITY_KINDS: &'static [&'static str] = &[];
            const HANDLERS: &'static [HandlerDef] = &[];
        }

        #[async_trait]
        impl PackRuntime for NoDepsA {
            fn name(&self) -> &str {
                Self::NAME
            }
            fn note_kinds(&self) -> &'static [&'static str] {
                Self::NOTE_KINDS
            }
            fn entity_kinds(&self) -> &'static [&'static str] {
                Self::ENTITY_KINDS
            }
            fn handlers(&self) -> &'static [HandlerDef] {
                Self::HANDLERS
            }
            async fn dispatch(
                &self,
                verb: &str,
                _: Value,
                _: &VerbRegistry,
                _: &NamespaceToken,
            ) -> Result<Value, RuntimeError> {
                Err(RuntimeError::InvalidInput(format!("NoDepsA: {verb}")))
            }
        }

        #[async_trait]
        impl PackRuntime for NoDepsB {
            fn name(&self) -> &str {
                Self::NAME
            }
            fn note_kinds(&self) -> &'static [&'static str] {
                Self::NOTE_KINDS
            }
            fn entity_kinds(&self) -> &'static [&'static str] {
                Self::ENTITY_KINDS
            }
            fn handlers(&self) -> &'static [HandlerDef] {
                Self::HANDLERS
            }
            async fn dispatch(
                &self,
                verb: &str,
                _: Value,
                _: &VerbRegistry,
                _: &NamespaceToken,
            ) -> Result<Value, RuntimeError> {
                Err(RuntimeError::InvalidInput(format!("NoDepsB: {verb}")))
            }
        }

        let mut builder = VerbRegistryBuilder::new();
        builder.register(NoDepsA);
        builder.register(NoDepsB);
        let reg = builder.build().expect("packs with REQUIRES=&[] build");
        assert_eq!(reg.pack_requires("no_deps_a").unwrap(), &[] as &[&str]);
        assert_eq!(reg.pack_requires("no_deps_b").unwrap(), &[] as &[&str]);
    }
}

// ── Dispatch hook tests (Issue #158) ─────────────────────────────────────────

#[cfg(test)]
mod hook_tests {
    use super::*;
    use async_trait::async_trait;
    use khive_types::Pack;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Mutex as StdMutex;

    struct SimplePack;

    impl Pack for SimplePack {
        const NAME: &'static str = "simple";
        const NOTE_KINDS: &'static [&'static str] = &[];
        const ENTITY_KINDS: &'static [&'static str] = &[];
        const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
            name: "ping",
            description: "ping",
            visibility: Visibility::Verb,
            category: VerbCategory::Assertive,
            params: &[],
        }];
    }

    #[async_trait]
    impl PackRuntime for SimplePack {
        fn name(&self) -> &str {
            SimplePack::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            SimplePack::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            SimplePack::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            SimplePack::HANDLERS
        }
        async fn dispatch(
            &self,
            verb: &str,
            _params: Value,
            _registry: &VerbRegistry,
            _token: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Ok(serde_json::json!({ "verb": verb }))
        }
    }

    /// Hook that counts calls and records the last verb seen.
    #[derive(Default)]
    struct CountingHook {
        calls: AtomicUsize,
        last_verb: StdMutex<String>,
    }

    #[async_trait]
    impl DispatchHook for CountingHook {
        async fn on_dispatch(&self, view: &EventView) {
            self.calls.fetch_add(1, Ordering::SeqCst);
            *self.last_verb.lock().unwrap() = view.event.verb.clone();
        }
    }

    #[tokio::test]
    async fn dispatch_hook_fires_on_successful_dispatch() {
        let hook = Arc::new(CountingHook::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(SimplePack);
        builder.with_dispatch_hook(hook.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("ping", Value::Null).await.unwrap();

        assert_eq!(
            hook.calls.load(Ordering::SeqCst),
            1,
            "hook must fire once per successful dispatch"
        );
        assert_eq!(
            hook.last_verb.lock().unwrap().as_str(),
            "ping",
            "hook event must carry the dispatched verb"
        );
    }

    #[tokio::test]
    async fn dispatch_hook_fires_multiple_times() {
        let hook = Arc::new(CountingHook::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(SimplePack);
        builder.with_dispatch_hook(hook.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("ping", Value::Null).await.unwrap();
        reg.dispatch("ping", Value::Null).await.unwrap();
        reg.dispatch("ping", Value::Null).await.unwrap();

        assert_eq!(
            hook.calls.load(Ordering::SeqCst),
            3,
            "hook must fire once per successful dispatch"
        );
    }

    #[tokio::test]
    async fn dispatch_hook_does_not_fire_on_unknown_verb() {
        let hook = Arc::new(CountingHook::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(SimplePack);
        builder.with_dispatch_hook(hook.clone());
        let reg = builder.build().expect("registry builds");

        let _ = reg.dispatch("nonexistent", Value::Null).await;

        assert_eq!(
            hook.calls.load(Ordering::SeqCst),
            0,
            "hook must NOT fire for unknown verb (dispatch returns error)"
        );
    }

    #[tokio::test]
    async fn dispatch_hook_does_not_fire_on_gate_deny() {
        use khive_gate::{Gate, GateDecision, GateError};

        #[derive(Debug)]
        struct AlwaysDenyGate;
        impl Gate for AlwaysDenyGate {
            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
                Ok(GateDecision::deny("test deny"))
            }
        }

        let hook = Arc::new(CountingHook::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(SimplePack);
        builder.with_gate(Arc::new(AlwaysDenyGate));
        builder.with_dispatch_hook(hook.clone());
        let reg = builder.build().expect("registry builds");

        let err = reg.dispatch("ping", Value::Null).await.unwrap_err();
        assert!(matches!(err, RuntimeError::PermissionDenied { .. }));

        assert_eq!(
            hook.calls.load(Ordering::SeqCst),
            0,
            "hook must NOT fire when gate denies dispatch"
        );
    }

    #[tokio::test]
    async fn dispatch_hook_event_carries_namespace_from_params() {
        let hook = Arc::new(CountingHook::default());

        #[derive(Default)]
        struct NsCapturingHook {
            ns: StdMutex<String>,
        }

        #[async_trait]
        impl DispatchHook for NsCapturingHook {
            async fn on_dispatch(&self, view: &EventView) {
                *self.ns.lock().unwrap() = view.event.namespace.clone();
            }
        }

        let ns_hook = Arc::new(NsCapturingHook::default());
        let mut builder = VerbRegistryBuilder::new();
        builder.register(SimplePack);
        builder.with_dispatch_hook(ns_hook.clone());
        let reg = builder.build().expect("registry builds");

        reg.dispatch("ping", serde_json::json!({"namespace": "tenant-abc"}))
            .await
            .unwrap();

        assert_eq!(
            ns_hook.ns.lock().unwrap().as_str(),
            "tenant-abc",
            "dispatch hook event must carry the resolved namespace"
        );

        // Suppress unused-variable warning from the outer hook.
        drop(hook);
    }

    #[tokio::test]
    async fn no_dispatch_hook_configured_dispatch_succeeds() {
        // Regression: registries without a hook must still work.
        let mut builder = VerbRegistryBuilder::new();
        builder.register(SimplePack);
        // No with_dispatch_hook call.
        let reg = builder.build().expect("registry builds");

        let res = reg.dispatch("ping", Value::Null).await.unwrap();
        assert_eq!(res["verb"], "ping");
    }
}

// ── help=true tests (issue #287) ──────────────────────────────────────────────

#[cfg(test)]
mod help_tests {
    use super::*;
    use async_trait::async_trait;
    use khive_types::Pack;
    use std::sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    };

    // ── HelpPack: a minimal pack with one handler that records invocation count.
    //
    // Used to verify that help=true never reaches the pack's dispatch method.

    static CREATE_PARAMS: [ParamDef; 2] = [
        ParamDef {
            name: "kind",
            param_type: "string",
            required: true,
            description: "Granular kind (concept | document | ...).",
        },
        ParamDef {
            name: "name",
            param_type: "string",
            required: false,
            description: "Human-readable name.",
        },
    ];

    static RECALL_PARAMS: [ParamDef; 2] = [
        ParamDef {
            name: "query",
            param_type: "string",
            required: true,
            description: "Semantic recall query.",
        },
        ParamDef {
            name: "limit",
            param_type: "integer",
            required: false,
            description: "Maximum memories to return.",
        },
    ];

    // A subhandler with no params — mirrors recall.embed / brain.emit / etc.
    // Used to test that help=true on a Subhandler returns callable_via_mcp: false.
    static EMBED_PARAMS: [ParamDef; 0] = [];

    struct HelpPack {
        invocations: Arc<AtomicUsize>,
    }

    impl Pack for HelpPack {
        const NAME: &'static str = "helptest";
        const NOTE_KINDS: &'static [&'static str] = &[];
        const ENTITY_KINDS: &'static [&'static str] = &[];
        const HANDLERS: &'static [HandlerDef] = &[
            HandlerDef {
                name: "create",
                description: "Create an entity or note",
                visibility: Visibility::Verb,
                category: VerbCategory::Commissive,
                params: &CREATE_PARAMS,
            },
            HandlerDef {
                name: "recall",
                description: "Recall memory notes with decay-aware hybrid ranking",
                visibility: Visibility::Verb,
                category: VerbCategory::Assertive,
                params: &RECALL_PARAMS,
            },
            // ue-help-introspection C1: a Subhandler used to test that
            // help=true returns callable_via_mcp: false for internal verbs.
            HandlerDef {
                name: "recall.embed",
                description: "Return the embedding vector used by memory recall",
                visibility: Visibility::Subhandler,
                category: VerbCategory::Assertive,
                params: &EMBED_PARAMS,
            },
        ];
    }

    #[async_trait]
    impl PackRuntime for HelpPack {
        fn name(&self) -> &str {
            HelpPack::NAME
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            HelpPack::NOTE_KINDS
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            HelpPack::ENTITY_KINDS
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            HelpPack::HANDLERS
        }
        async fn dispatch(
            &self,
            verb: &str,
            _params: Value,
            _registry: &VerbRegistry,
            _token: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            self.invocations.fetch_add(1, Ordering::SeqCst);
            Ok(serde_json::json!({ "pack": "helptest", "verb": verb }))
        }
    }

    fn build_help_registry(invocations: Arc<AtomicUsize>) -> VerbRegistry {
        let mut builder = VerbRegistryBuilder::new();
        builder.register(HelpPack { invocations });
        builder.build().expect("help registry builds")
    }

    /// help=true on `create` returns a schema envelope with the correct verb name,
    /// pack name, description, and at least the required `kind` parameter.
    #[tokio::test]
    async fn test_help_true_returns_schema_for_kg_create() {
        let invocations = Arc::new(AtomicUsize::new(0));
        let reg = build_help_registry(invocations.clone());

        let result = reg
            .dispatch("create", serde_json::json!({ "help": true }))
            .await
            .expect("help=true must succeed for a known verb");

        // Shape checks.
        assert_eq!(result["verb"], "create", "envelope must name the verb");
        assert_eq!(
            result["pack"], "helptest",
            "envelope must name the owning pack"
        );
        assert!(
            result["description"].as_str().is_some(),
            "description must be a string"
        );

        // Params array must be present and non-empty.
        let params = result["params"]
            .as_array()
            .expect("params must be a JSON array");
        assert!(!params.is_empty(), "params array must not be empty");

        // The required `kind` param must appear.
        let kind_param = params.iter().find(|p| p["name"] == "kind");
        assert!(
            kind_param.is_some(),
            "params array must include the 'kind' parameter"
        );
        let kind_param = kind_param.unwrap();
        assert_eq!(
            kind_param["required"],
            serde_json::json!(true),
            "'kind' must be required"
        );
        assert_eq!(kind_param["type"], "string", "'kind' type must be 'string'");
    }

    /// help=true on `recall` returns a schema envelope including the `query` param.
    #[tokio::test]
    async fn test_help_true_returns_schema_for_recall() {
        let invocations = Arc::new(AtomicUsize::new(0));
        let reg = build_help_registry(invocations.clone());

        let result = reg
            .dispatch("recall", serde_json::json!({ "help": true }))
            .await
            .expect("help=true must succeed for recall");

        assert_eq!(result["verb"], "recall");
        assert_eq!(result["pack"], "helptest");

        let params = result["params"]
            .as_array()
            .expect("params must be a JSON array");

        // `query` must be present and required.
        let query_param = params.iter().find(|p| p["name"] == "query");
        assert!(query_param.is_some(), "params must include 'query'");
        let query_param = query_param.unwrap();
        assert_eq!(
            query_param["required"],
            serde_json::json!(true),
            "'query' must be required"
        );

        // `limit` must be present and optional.
        let limit_param = params.iter().find(|p| p["name"] == "limit");
        assert!(limit_param.is_some(), "params must include 'limit'");
        let limit_param = limit_param.unwrap();
        assert_eq!(
            limit_param["required"],
            serde_json::json!(false),
            "'limit' must be optional"
        );
    }

    /// help=true is intercepted before pack dispatch — the pack's dispatch method
    /// must never be invoked when help=true is in the params.
    #[tokio::test]
    async fn test_help_true_does_not_execute_the_verb() {
        let invocations = Arc::new(AtomicUsize::new(0));
        let reg = build_help_registry(invocations.clone());

        // Call both verbs with help=true.
        reg.dispatch("create", serde_json::json!({ "help": true }))
            .await
            .expect("help=true must succeed");
        reg.dispatch("recall", serde_json::json!({ "help": true }))
            .await
            .expect("help=true must succeed");

        assert_eq!(
            invocations.load(Ordering::SeqCst),
            0,
            "pack dispatch MUST NOT be invoked when help=true; \
             got {} invocation(s)",
            invocations.load(Ordering::SeqCst)
        );

        // Confirm that a normal call (without help=true) DOES invoke dispatch.
        reg.dispatch("create", serde_json::json!({}))
            .await
            .expect("normal dispatch must succeed");
        assert_eq!(
            invocations.load(Ordering::SeqCst),
            1,
            "pack dispatch must fire exactly once for a normal call"
        );
    }

    // ── ue-help-introspection C1 regressions ─────────────────────────────────
    //
    // Subhandler verbs must return `callable_via_mcp: false` in their help
    // schema so agents who read help=true before probing see accurate
    // availability — not a "looks callable" schema followed by permission denied.

    /// help=true on a `Visibility::Subhandler` verb returns `callable_via_mcp: false`
    /// and `visibility: "internal"` rather than a plain callable-looking envelope.
    #[tokio::test]
    async fn help_true_on_subhandler_returns_callable_via_mcp_false() {
        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));

        let result = reg
            .dispatch("recall.embed", serde_json::json!({ "help": true }))
            .await
            .expect("help=true on subhandler must succeed (no permission check on help path)");

        assert_eq!(
            result["callable_via_mcp"],
            serde_json::json!(false),
            "subhandler help must carry callable_via_mcp: false"
        );
        assert_eq!(
            result["visibility"], "internal",
            "subhandler help must carry visibility: internal"
        );
        // The verb and pack fields must still be present so the caller knows
        // what the schema belongs to.
        assert_eq!(result["verb"], "recall.embed");
        assert_eq!(result["pack"], "helptest");
    }

    /// Public Verb-visibility handlers must NOT have `callable_via_mcp: false`.
    #[tokio::test]
    async fn help_true_on_public_verb_does_not_have_callable_via_mcp_false() {
        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));

        let result = reg
            .dispatch("create", serde_json::json!({ "help": true }))
            .await
            .expect("help=true on public verb must succeed");

        // callable_via_mcp must be absent or true for public verbs.
        assert_ne!(
            result.get("callable_via_mcp"),
            Some(&serde_json::json!(false)),
            "public verb help must NOT carry callable_via_mcp: false"
        );
        // visibility must be absent or 'public' (never 'internal') for public verbs.
        assert_ne!(
            result.get("visibility"),
            Some(&serde_json::json!("internal")),
            "public verb help must NOT carry visibility: internal"
        );
    }

    /// help=true on an unknown verb returns an error (same behavior as normal dispatch).
    #[tokio::test]
    async fn help_true_on_unknown_verb_returns_error() {
        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));

        let err = reg
            .dispatch("nonexistent_verb", serde_json::json!({ "help": true }))
            .await
            .unwrap_err();

        assert!(
            matches!(err, RuntimeError::InvalidInput(_)),
            "help=true on unknown verb must return InvalidInput, got {err:?}"
        );
        let msg = err.to_string();
        assert!(
            msg.contains("nonexistent_verb"),
            "error must name the unknown verb: {msg}"
        );
    }

    /// Subhandler help must include params: [] even when the verb has no params.
    #[tokio::test]
    async fn help_true_on_subhandler_includes_params_field() {
        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));

        let result = reg
            .dispatch("recall.embed", serde_json::json!({ "help": true }))
            .await
            .expect("help=true on subhandler must succeed");

        // params must always be present (H1: consistent shape).
        let params = result
            .get("params")
            .expect("subhandler help must include 'params' field");
        assert!(
            params.is_array(),
            "subhandler help params must be a JSON array"
        );
    }

    // ── codex High: unknown-verb error must not leak subhandler names ─────────

    /// `describe_verb` on an unknown verb must list only Verb-visibility names
    /// in the "available" list — never subhandler names like `recall.embed`
    /// (codex High — ue-help-introspection C1 / unknown-verb path).
    #[tokio::test]
    async fn help_true_unknown_verb_available_list_excludes_subhandlers() {
        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));

        let err = reg
            .dispatch("not_a_verb", serde_json::json!({ "help": true }))
            .await
            .unwrap_err();

        let msg = err.to_string();
        // `recall.embed` is a Subhandler in HelpPack — must NOT appear in the
        // "available" list of an unknown-verb error.
        assert!(
            !msg.contains("recall.embed"),
            "unknown-verb help error must not advertise subhandler recall.embed: {msg}"
        );
        // Public verbs must still appear so the agent knows what to call.
        assert!(
            msg.contains("create"),
            "unknown-verb help error must still list public verb 'create': {msg}"
        );
        assert!(
            msg.contains("recall"),
            "unknown-verb help error must still list public verb 'recall': {msg}"
        );
    }

    /// Normal dispatch on an unknown verb must also not leak subhandler names.
    #[tokio::test]
    async fn dispatch_unknown_verb_available_list_excludes_subhandlers() {
        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));

        let err = reg
            .dispatch("not_a_verb", serde_json::json!({}))
            .await
            .unwrap_err();

        let msg = err.to_string();
        // `recall.embed` is a Subhandler in HelpPack — must NOT appear in the
        // "available" list of an unknown-verb dispatch error.
        assert!(
            !msg.contains("recall.embed"),
            "dispatch unknown-verb error must not advertise subhandler recall.embed: {msg}"
        );
        // Public verbs must still appear so the agent knows what to call.
        assert!(
            msg.contains("create"),
            "dispatch unknown-verb error must still list public verb 'create': {msg}"
        );
        assert!(
            msg.contains("recall"),
            "dispatch unknown-verb error must still list public verb 'recall': {msg}"
        );
    }

    // ── ADR-028 multi-backend schema routing tests ───────────────────────────

    /// A test pack that returns a real SchemaPlan so we can assert routing.
    struct SchemaPack {
        pack_name: &'static str,
        statements: &'static [&'static str],
    }

    impl Pack for SchemaPack {
        const NAME: &'static str = "schema-pack";
        const NOTE_KINDS: &'static [&'static str] = &[];
        const ENTITY_KINDS: &'static [&'static str] = &[];
        const HANDLERS: &'static [HandlerDef] = &[];
    }

    #[async_trait]
    impl PackRuntime for SchemaPack {
        fn name(&self) -> &str {
            self.pack_name
        }
        fn note_kinds(&self) -> &'static [&'static str] {
            &[]
        }
        fn entity_kinds(&self) -> &'static [&'static str] {
            &[]
        }
        fn handlers(&self) -> &'static [HandlerDef] {
            &[]
        }
        fn schema_plan(&self) -> SchemaPlan {
            SchemaPlan {
                pack: self.pack_name,
                statements: self.statements,
            }
        }
        async fn dispatch(
            &self,
            verb: &str,
            _params: Value,
            _registry: &VerbRegistry,
            _token: &NamespaceToken,
        ) -> Result<Value, RuntimeError> {
            Ok(serde_json::json!({ "pack": self.pack_name, "verb": verb }))
        }
    }

    // ADR-028: all_schema_plans_named returns (pack_name, SchemaPlan) pairs
    // where pack_name comes from SchemaPlan::pack (always &'static str).
    #[test]
    fn all_schema_plans_named_returns_correct_pairs() {
        let mut builder = VerbRegistryBuilder::new();
        builder.register_boxed(Box::new(SchemaPack {
            pack_name: "alpha",
            statements: &["CREATE TABLE IF NOT EXISTS t_alpha (id INTEGER PRIMARY KEY)"],
        }));
        builder.register_boxed(Box::new(SchemaPack {
            pack_name: "beta",
            statements: &[],
        }));
        let reg = builder.build().expect("registry builds");

        let named = reg.all_schema_plans_named();
        assert_eq!(named.len(), 2);

        let alpha_entry = named.iter().find(|(n, _)| *n == "alpha");
        let beta_entry = named.iter().find(|(n, _)| *n == "beta");

        assert!(alpha_entry.is_some(), "alpha must appear in named plans");
        assert!(beta_entry.is_some(), "beta must appear in named plans");

        let (_, alpha_plan) = alpha_entry.unwrap();
        assert_eq!(alpha_plan.statements.len(), 1);
        assert!(!alpha_plan.is_empty());

        let (_, beta_plan) = beta_entry.unwrap();
        assert!(beta_plan.is_empty());
    }

    // ADR-028: apply_schema_plans_with_map routes non-empty plans to the
    // correct per-pack backend instead of the default.
    //
    // Verification: apply DDL to routed backend, then confirm the table is
    // present on pack_backend and absent on default_backend by attempting to
    // apply the same DDL again — if the table already exists on pack_backend
    // the idempotent CREATE IF NOT EXISTS succeeds; applying to default_backend
    // would only matter if the table were routed there.  We verify isolation
    // by applying the plan and then running a targeted DDL on each backend
    // that would fail if the table did not already exist (CREATE without
    // IF NOT EXISTS on a duplicate raises an error), combined with a no-error
    // path on the correct backend.
    //
    // Simpler approach: confirm the plan applies without error (routing is
    // correct) and that the opposite backend returns an error when we try to
    // INSERT into the routed table (table-not-found = SQLITE_ERROR).
    #[tokio::test]
    async fn apply_schema_plans_with_map_routes_to_correct_backend() {
        use khive_storage::types::{SqlStatement, SqlValue};

        let default_backend = khive_db::StorageBackend::memory().expect("default memory backend");
        let pack_backend =
            khive_db::StorageBackend::memory().expect("pack-specific memory backend");

        let mut builder = VerbRegistryBuilder::new();
        builder.register_boxed(Box::new(SchemaPack {
            pack_name: "routed",
            statements: &["CREATE TABLE IF NOT EXISTS t_routed (id INTEGER PRIMARY KEY)"],
        }));
        let reg = builder.build().expect("registry builds");

        let mut backend_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
        backend_map.insert("routed", &pack_backend);

        reg.apply_schema_plans_with_map(&backend_map, &default_backend)
            .expect("schema application must not collide");

        // On pack_backend: INSERT must succeed (table exists).
        let mut writer = pack_backend.sql().writer().await.expect("writer");
        let result = writer
            .execute(SqlStatement {
                sql: "INSERT INTO t_routed (id) VALUES (?1)".into(),
                params: vec![SqlValue::Integer(1)],
                label: None,
            })
            .await;
        assert!(
            result.is_ok(),
            "t_routed must exist on pack_backend after routing: {result:?}"
        );

        // On default_backend: INSERT must fail (table not there).
        let mut default_writer = default_backend.sql().writer().await.expect("writer");
        let default_result = default_writer
            .execute(SqlStatement {
                sql: "INSERT INTO t_routed (id) VALUES (?1)".into(),
                params: vec![SqlValue::Integer(2)],
                label: None,
            })
            .await;
        assert!(
            default_result.is_err(),
            "t_routed must NOT exist on default_backend (table should not be there)"
        );
    }

    // ADR-028: apply_schema_plans_with_map uses default backend for packs
    // absent from the map.
    #[tokio::test]
    async fn apply_schema_plans_with_map_falls_back_to_default_for_unmapped_packs() {
        use khive_storage::types::{SqlStatement, SqlValue};

        let default_backend = khive_db::StorageBackend::memory().expect("default memory backend");

        let mut builder = VerbRegistryBuilder::new();
        builder.register_boxed(Box::new(SchemaPack {
            pack_name: "unmapped",
            statements: &["CREATE TABLE IF NOT EXISTS t_unmapped (id INTEGER PRIMARY KEY)"],
        }));
        let reg = builder.build().expect("registry builds");

        let backend_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
        reg.apply_schema_plans_with_map(&backend_map, &default_backend)
            .expect("schema application must not collide");

        // On default_backend: INSERT must succeed (table fell back here).
        let mut writer = default_backend.sql().writer().await.expect("writer");
        let result = writer
            .execute(SqlStatement {
                sql: "INSERT INTO t_unmapped (id) VALUES (?1)".into(),
                params: vec![SqlValue::Integer(1)],
                label: None,
            })
            .await;
        assert!(
            result.is_ok(),
            "t_unmapped must exist on default_backend for unmapped pack: {result:?}"
        );
    }

    // ADR-028 B-SHOULD-FIX-3: two packs declaring the same auxiliary table on
    // the same backend must cause apply_schema_plans_with_map to return an
    // error that names both packs and the table — it is a boot-time failure,
    // not a silent DDL race.
    #[test]
    fn apply_schema_plans_with_map_collision_is_an_error() {
        let backend = khive_db::StorageBackend::memory().expect("memory backend");
        let empty_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();

        let mut builder = VerbRegistryBuilder::new();
        builder.register_boxed(Box::new(SchemaPack {
            pack_name: "pack_alpha",
            statements: &["CREATE TABLE IF NOT EXISTS collision_table (id INTEGER PRIMARY KEY)"],
        }));
        builder.register_boxed(Box::new(SchemaPack {
            pack_name: "pack_beta",
            statements: &["CREATE TABLE IF NOT EXISTS collision_table (id INTEGER PRIMARY KEY)"],
        }));
        let registry = builder.build().expect("registry builds");

        let result = registry.apply_schema_plans_with_map(&empty_map, &backend);
        assert!(
            result.is_err(),
            "two packs declaring the same table on the same backend must produce a collision error"
        );
        let err = result.unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("pack_alpha"),
            "collision error must name first pack; got: {msg}"
        );
        assert!(
            msg.contains("pack_beta"),
            "collision error must name second pack; got: {msg}"
        );
        assert!(
            msg.contains("collision_table"),
            "collision error must name the table; got: {msg}"
        );
    }
}