Skip to main content

khive_runtime/
pack.rs

1// FILE SIZE JUSTIFICATION: pack.rs is the load-bearing dispatch core — VerbRegistry,
2// VerbRegistryBuilder, PackRuntime, DispatchHook, and their test scaffolding all
3// share internal state (packs Vec, gate, event_store) that cannot be cleanly split
4// without exposing private fields or duplicating the scaffolding. Inline tests cover
5// collision detection and dispatch path that require direct access to VerbRegistry
6// internals. Split plan: when the verb surface reaches a stable v1 API, extract
7// VerbRegistryBuilder into `pack/builder.rs` and gate/event logic into `pack/dispatch.rs`.
8//! Pack runtime trait and verb registry.
9//!
10//! `PackRuntime` mirrors `Pack`'s const associated items as methods for object safety.
11//! Build a [`VerbRegistry`] via `VerbRegistryBuilder::build()`; registration is builder-only.
12
13use std::collections::{HashMap, HashSet, VecDeque};
14use std::sync::Arc;
15use std::time::Instant;
16
17use crate::runtime::NamespaceToken;
18use async_trait::async_trait;
19use khive_gate::{AllowAllGate, AuditEvent, GateDecision, GateRef, GateRequest};
20use khive_storage::{Event, EventStore, EventView, SubstrateKind};
21use khive_types::{EventKind, EventOutcome, Namespace};
22use serde_json::Value;
23
24pub use khive_types::{
25    EdgeEndpointRule, EndpointKind, EntityTypeDef, HandlerDef, NoteKindSpec, NoteLifecycleSpec,
26    PackSchemaPlan, ParamDef, VerbCategory, VerbPresentationPolicy, Visibility,
27};
28// Backward-compat re-export.
29#[allow(deprecated)]
30pub use khive_types::VerbDef;
31
32use crate::validation::ValidationRule;
33
34/// Pack-auxiliary schema plan.
35///
36/// Declares `CREATE TABLE IF NOT EXISTS` statements for pack-owned tables that
37/// are NOT part of the core substrate schema (entities, notes, edges, events).
38/// Applied at boot via `StorageBackend::apply_schema` / `apply_pack_schema_plan`.
39///
40/// Core substrate tables evolve through versioned migrations. Pack schema is
41/// strictly for pack-auxiliary tables (e.g. GTD lifecycle audit, memory index).
42/// v1 pack schemas are non-versioned.
43#[derive(Debug, Default, Clone)]
44pub struct SchemaPlan {
45    /// Owning pack name.
46    pub pack: &'static str,
47    /// DDL statements applied idempotently at boot.
48    /// Each entry must be a self-contained `CREATE TABLE IF NOT EXISTS` or
49    /// similar idempotent statement.
50    pub statements: &'static [&'static str],
51}
52
53impl SchemaPlan {
54    /// Construct a `SchemaPlan` with no statements.
55    ///
56    /// Packs whose state lives entirely in the core substrate tables (entities,
57    /// notes, edges) use this as their `schema_plan()` return value.
58    pub const fn empty() -> Self {
59        Self {
60            pack: "",
61            statements: &[],
62        }
63    }
64
65    /// Returns `true` when the plan contains no DDL statements.
66    pub fn is_empty(&self) -> bool {
67        self.statements.is_empty()
68    }
69}
70
71/// Hook called after every successful verb dispatch.
72///
73/// Packs observe enriched event views so provenance-aware consumers can use
74/// `view.observations` while legacy folds can still consume `view.event`.
75#[async_trait]
76pub trait DispatchHook: Send + Sync {
77    /// Called with the dispatch-outcome event view after a successful pack dispatch.
78    ///
79    /// Errors are logged via `tracing::warn!` and never propagated to the
80    /// caller; the dispatch has already succeeded.
81    async fn on_dispatch(&self, view: &EventView);
82}
83
84use crate::error::{
85    CircularPackDependency, MissingPackDependencies, MissingPackDependency, RuntimeError,
86};
87use crate::KhiveRuntime;
88
89/// Async dispatch trait for packs.
90///
91/// This is the object-safe behavioral counterpart to `khive_types::Pack`.
92/// `Pack` uses const associated items (not object-safe in Rust); this trait
93/// mirrors that metadata as methods and adds async dispatch.
94///
95/// Registration requires `P: Pack + PackRuntime` — the compiler enforces
96/// that every runtime pack also declares its vocabulary via `Pack`.
97#[async_trait]
98pub trait PackRuntime: Send + Sync {
99    /// Pack name — must equal `<Self as Pack>::NAME`.
100    fn name(&self) -> &str;
101
102    /// Note kinds this pack owns — must equal `<Self as Pack>::NOTE_KINDS`.
103    fn note_kinds(&self) -> &'static [&'static str];
104
105    /// Entity kinds this pack owns — must equal `<Self as Pack>::ENTITY_KINDS`.
106    fn entity_kinds(&self) -> &'static [&'static str];
107
108    /// Handlers this pack registers — must equal `<Self as Pack>::HANDLERS`.
109    fn handlers(&self) -> &'static [HandlerDef];
110
111    /// Pack-extensible edge endpoint rules — must equal `<Self as Pack>::EDGE_RULES`.
112    /// Defaults to empty so existing packs that don't extend the edge contract
113    /// can ignore it.
114    fn edge_rules(&self) -> &'static [EdgeEndpointRule] {
115        &[]
116    }
117
118    /// Pack-extensible entity-type subtypes — must equal `<Self as Pack>::ENTITY_TYPES`.
119    /// Defaults to empty so existing packs that don't extend the entity_type
120    /// registry can ignore it.
121    fn entity_types(&self) -> &'static [EntityTypeDef] {
122        &[]
123    }
124
125    /// Pack names whose vocabulary this pack references.
126    /// Defaults to empty so existing packs compile without changes.
127    fn requires(&self) -> &'static [&'static str] {
128        &[]
129    }
130
131    /// NoteKindSpec declarations for note kinds this pack owns.
132    ///
133    /// Packs that introduce note kinds with explicit lifecycle semantics
134    /// declare the spec here.  The runtime collects these for introspection
135    /// and future enforcement.  Defaults to empty so existing packs compile
136    /// without changes.
137    fn note_kind_specs(&self) -> &'static [NoteKindSpec] {
138        &[]
139    }
140
141    /// Optional per-kind hook for shared CRUD specialization.
142    ///
143    /// When a kind is owned by this pack (declared in `note_kinds()` or
144    /// `entity_kinds()`), returning `Some(hook)` opts that kind into
145    /// pack-specific behavior — defaults, derived properties, side-effect
146    /// edges — through the shared `create` path. Returning `None` keeps
147    /// the kind as plain storage with no specialization.
148    fn kind_hook(&self, _kind: &str) -> Option<Arc<dyn KindHook>> {
149        None
150    }
151
152    /// Pack-auxiliary schema.
153    ///
154    /// Returns DDL statements for pack-owned tables that are NOT part of the
155    /// core substrate schema. Statements are idempotent (`CREATE TABLE IF NOT
156    /// EXISTS`) so callers can apply them safely on every registration. Core
157    /// substrate tables evolve through versioned migrations; pack schema is
158    /// strictly pack-auxiliary.
159    ///
160    /// Defaults to an empty plan — packs that store everything in the core
161    /// substrate tables (entities, notes, edges, events) return this default.
162    ///
163    /// Plans are aggregated via [`VerbRegistry::all_schema_plans`] and applied
164    /// at startup via `KhiveMcpServer::with_packs`. Packs that need their
165    /// schema present (e.g. GTD) also self-bootstrap lazily on first call for
166    /// robustness in test contexts that create fresh in-memory databases.
167    fn schema_plan(&self) -> SchemaPlan {
168        SchemaPlan::empty()
169    }
170
171    /// Domain-specific validation rules contributed by this pack.
172    ///
173    /// Rule IDs MUST follow the `<pack>/<rule-id>` namespace convention.
174    /// Built-in rules (no pack prefix) are reserved for the `khive-runtime`
175    /// validation infrastructure.
176    ///
177    /// Defaults to empty — packs with no domain-specific rules return `&[]`.
178    fn validation_rules(&self) -> &'static [ValidationRule] {
179        &[]
180    }
181
182    /// Register custom embedding providers with the runtime. Called during pack
183    /// initialisation, before the first verb dispatch, so `KhiveRuntime::embedder(name)`
184    /// resolves provider names declared here. Default no-op — packs that only use
185    /// built-in lattice models do not need to override this.
186    /// See `docs/api/pack.md#register_embedders` for a usage example.
187    fn register_embedders(&self, _runtime: &KhiveRuntime) {}
188
189    /// Install a pack-owned entity-type validator on the runtime, called during pack
190    /// initialisation (after the registry is built, before the first dispatch) so
191    /// `create_many`/`create_entity` reject unregistered `entity_type` values at the
192    /// runtime layer. Default no-op leaves the validator absent (skip-when-None).
193    /// See `docs/api/pack.md#register_entity_type_validator` for the two-hook compatibility contract.
194    fn register_entity_type_validator(&self, _runtime: &KhiveRuntime) {}
195
196    /// Install a pack-owned entity-type validator that also receives the boot-time
197    /// composed set of every loaded pack's `ENTITY_TYPES` ([`VerbRegistry::all_entity_types`]).
198    /// Defaults to calling [`register_entity_type_validator`](Self::register_entity_type_validator)
199    /// with just the runtime. `call_register_entity_type_validators` calls this hook, not
200    /// the simpler one — override this to receive the composed vocabulary.
201    /// See `docs/api/pack.md#register_entity_type_validator` for the two-hook compatibility contract.
202    fn register_entity_type_validator_with_types(
203        &self,
204        runtime: &KhiveRuntime,
205        _pack_entity_types: &[EntityTypeDef],
206    ) {
207        self.register_entity_type_validator(runtime);
208    }
209
210    /// Install a pack-owned note-mutation hook on the runtime, called during pack
211    /// initialisation with the same timing as `register_entity_type_validator`. Packs
212    /// that cache derived state keyed by note content (e.g. `khive-pack-memory`'s warm
213    /// ANN index) override this to install a hook via
214    /// `KhiveRuntime::install_note_mutation_hook`. Default no-op leaves the hook absent.
215    /// See `docs/api/pack.md#register_note_mutation_hook` for cross-pack notification rationale.
216    fn register_note_mutation_hook(&self, _runtime: &KhiveRuntime) {}
217
218    /// Warm up any in-memory state from persisted snapshots (optional). Called after
219    /// all packs are registered but before serving the first request. Must be
220    /// idempotent and infallible — errors are logged internally, never propagated.
221    async fn warm(&self) {}
222
223    /// Names of all embedding models registered on this pack's underlying runtime
224    /// handle. Defaults to empty — only packs that own embedding-bearing verbs
225    /// (kg, memory) need to override this.
226    /// See `docs/api/pack.md#registered_embedding_model_names` for the ADR-103 consumer.
227    fn registered_embedding_model_names(&self) -> Vec<String> {
228        Vec::new()
229    }
230
231    /// Dispatch a verb call. Returns serialized JSON response.
232    ///
233    /// The `registry` parameter gives the handler access to the merged
234    /// vocabulary and kind hooks across all loaded packs.
235    /// The `token` is an authorized namespace token minted by the dispatch
236    /// boundary after gate authorization — handlers must use it directly.
237    async fn dispatch(
238        &self,
239        verb: &str,
240        params: Value,
241        registry: &VerbRegistry,
242        token: &NamespaceToken,
243    ) -> Result<Value, RuntimeError>;
244}
245
246/// Per-kind specialization for shared CRUD.
247///
248/// Packs implement `KindHook` for kinds they own that need:
249/// - **Defaults** filled into create args (e.g. `status="inbox"` for tasks)
250/// - **Derived properties** computed from args (e.g. salience from priority)
251/// - **Side-effect writes** after the storage commit (e.g. `depends_on` edges)
252///
253/// Hooks are stateless from the framework's perspective — they receive the
254/// runtime as a method parameter and operate on the args `Value` directly.
255/// The pack registers them via [`PackRuntime::kind_hook`].
256///
257/// Lifecycle verbs (e.g. gtd's `complete`, `transition`) remain pack-owned
258/// verbs and do not flow through this trait — only the create path does.
259#[async_trait]
260pub trait KindHook: Send + Sync + std::fmt::Debug {
261    /// Mutate args before the storage write. Fill defaults, normalize values,
262    /// rearrange user-facing fields into the storage shape expected by the
263    /// shared CRUD handler.
264    ///
265    /// Returning an error aborts the create call (no storage write happens).
266    async fn prepare_create(
267        &self,
268        runtime: &KhiveRuntime,
269        args: &mut Value,
270    ) -> Result<(), RuntimeError>;
271
272    /// Fire side effects after a successful storage write — graph edges,
273    /// derived observations, etc. The newly created record's UUID is passed
274    /// so the hook can attach metadata referencing it.
275    ///
276    /// Errors here are **logged but not propagated** — the storage write has
277    /// already succeeded; failing the call would mislead the caller.
278    /// Implementations should `tracing::warn!` and return `Ok(())` for
279    /// best-effort side effects.
280    async fn after_create(
281        &self,
282        runtime: &KhiveRuntime,
283        id: uuid::Uuid,
284        args: &Value,
285    ) -> Result<(), RuntimeError>;
286}
287
288/// Optional sub-trait for packs that own private SQL tables and issue UUIDs
289/// that must be reachable through the generic `get(id)` and `delete(id)` verbs.
290///
291/// Implementing both methods is required — the sub-trait bundles them atomically
292/// so partial implementation is a compile-time error, not a runtime surprise.
293/// Packs whose records live in the shared entity/note substrate (gtd, memory)
294/// do not implement this sub-trait.
295#[async_trait]
296pub trait PackByIdResolver: Send + Sync {
297    /// Attempt to resolve a live (non-deleted) UUID owned by this pack's private tables.
298    ///
299    /// Returns `Some(Resolved::PackRecord { ... })` if this pack owns the UUID,
300    /// `None` if it does not (the caller continues to the next resolver),
301    /// or `Err(...)` on a storage error.
302    ///
303    /// Must query domain-authoritative tables before mirror tables.
304    /// Must NOT filter by namespace. UUID v4 is globally unique; by-ID
305    /// resolution is namespace-blind per ADR-007.
306    async fn resolve_by_id(
307        &self,
308        id: uuid::Uuid,
309    ) -> Result<Option<crate::Resolved>, crate::RuntimeError>;
310
311    /// Attempt to resolve a UUID including already-soft-deleted records.
312    ///
313    /// Used by the hard-delete path. Default delegates to `resolve_by_id`;
314    /// packs with `deleted_at` columns override this to query without the filter.
315    async fn resolve_by_id_including_deleted(
316        &self,
317        id: uuid::Uuid,
318    ) -> Result<Option<crate::Resolved>, crate::RuntimeError> {
319        self.resolve_by_id(id).await
320    }
321
322    /// Delete a record owned by this pack's private tables.
323    ///
324    /// `hard` mirrors the `delete` verb's `hard?` argument.
325    /// Default behavior for packs with a `deleted_at` column MUST be soft-delete;
326    /// `hard=true` performs permanent removal.
327    ///
328    /// Returns `Ok(Value)` with a `{ deleted: true, id, kind, hard }` body on success.
329    /// Returns `Err(RuntimeError::NotFound(...))` if the record does not exist.
330    async fn delete_by_id(
331        &self,
332        id: uuid::Uuid,
333        hard: bool,
334    ) -> Result<serde_json::Value, crate::RuntimeError>;
335}
336
337/// Builder for constructing a `VerbRegistry`.
338///
339/// Packs are registered here; once `.build()` is called the registry is
340/// immutable and cheaply cloneable.
341pub struct VerbRegistryBuilder {
342    packs: Vec<Box<dyn PackRuntime>>,
343    resolvers: Vec<(String, Box<dyn PackByIdResolver>)>,
344    gate: GateRef,
345    default_namespace: String,
346    /// Operator-configured read-visibility set (ADR-007 Rev 4 Rule 3b).
347    ///
348    /// Threads into `VerbRegistry::visible_namespaces` and is consumed by the
349    /// default dispatch path to widen read scope to `['local'] ∪ visible_namespaces`.
350    /// Writes remain pinned to `'local'`. An explicit `namespace=` request param
351    /// is a precise escape and is not widened by this set. A cloud gate may also
352    /// consult the list as policy input at its own layer.
353    visible_namespaces: Vec<Namespace>,
354    /// Configured actor identity label (ADR-057). When set, dispatch mints tokens
355    /// carrying this actor so that `comm.inbox` filters by `to_actor`.
356    actor_id: Option<String>,
357    /// Optional audit event sink.
358    ///
359    /// When set, every gate check writes a storage `Event` in addition to the
360    /// `tracing::info!` emission. The store is `Arc<dyn EventStore>` so the
361    /// registry does not depend on the full `KhiveRuntime` surface — only the
362    /// audit-persistence capability is needed here.
363    event_store: Option<Arc<dyn EventStore>>,
364    /// Optional post-dispatch hook.
365    ///
366    /// When set, every successful pack dispatch calls `hook.on_dispatch(event)`
367    /// with a synthesized Event describing the outcome. Opt-in: when None,
368    /// no overhead is incurred.
369    dispatch_hook: Option<Arc<dyn DispatchHook>>,
370}
371
372impl VerbRegistryBuilder {
373    /// Create a builder with no packs, `AllowAllGate`, and the local namespace as default.
374    pub fn new() -> Self {
375        Self {
376            packs: Vec::new(),
377            resolvers: Vec::new(),
378            gate: std::sync::Arc::new(AllowAllGate),
379            default_namespace: Namespace::local().as_str().to_string(),
380            visible_namespaces: vec![],
381            actor_id: None,
382            event_store: None,
383            dispatch_hook: None,
384        }
385    }
386
387    /// Set the operator-configured read-visibility set (ADR-007 Rev 4 Rule 3b).
388    ///
389    /// On the default (no explicit `namespace=` param) dispatch path, reads fan
390    /// out over `['local'] ∪ ns`. Writes remain pinned to `'local'`. An explicit
391    /// `namespace=` request parameter is a precise single-namespace escape and
392    /// is not widened by this set. A cloud gate may also consult the list as
393    /// policy input at its own layer.
394    pub fn with_visible_namespaces(&mut self, ns: Vec<Namespace>) -> &mut Self {
395        self.visible_namespaces = ns;
396        self
397    }
398
399    /// Set the configured actor identity label (ADR-057).
400    ///
401    /// When set, the dispatch path mints tokens carrying this actor so that
402    /// `comm.inbox` applies the `to_actor` filter for directed delivery.
403    /// When `None` (default), tokens carry `ActorRef::anonymous()` and inbox
404    /// falls back to party-line behavior.
405    pub fn with_actor_id(&mut self, actor_id: Option<String>) -> &mut Self {
406        self.actor_id = actor_id;
407        self
408    }
409
410    /// Register a pack. The bound `P: Pack + PackRuntime` ensures the pack
411    /// declares vocabulary via `Pack` consts alongside runtime dispatch.
412    pub fn register<P: khive_types::Pack + PackRuntime + 'static>(&mut self, pack: P) -> &mut Self {
413        self.packs.push(Box::new(pack));
414        self
415    }
416
417    /// Register a boxed pack directly.
418    ///
419    /// Crate-private: only [`PackRegistry::register_packs`] should call this.
420    /// External callers must use the typed [`Self::register`] which enforces the
421    /// `Pack + PackRuntime` dual-impl contract at the call site.  Here the
422    /// contract is satisfied upstream at the [`PackFactory::create`] site.
423    pub(crate) fn register_boxed(&mut self, pack: Box<dyn PackRuntime>) -> &mut Self {
424        self.packs.push(pack);
425        self
426    }
427
428    /// Register a by-ID resolver for a pack that owns private SQL tables.
429    ///
430    /// Packs that implement `PackByIdResolver` call this during their boot path
431    /// so that `get(id)` and `delete(id)` can reach their records.
432    pub fn register_resolver(
433        &mut self,
434        name: impl Into<String>,
435        resolver: Box<dyn PackByIdResolver>,
436    ) -> &mut Self {
437        self.resolvers.push((name.into(), resolver));
438        self
439    }
440
441    /// Set the authorization gate consulted on every dispatch.
442    ///
443    /// Defaults to `AllowAllGate` if not set. `Deny` is authoritative — a deny
444    /// decision aborts dispatch with `RuntimeError::PermissionDenied`. Gate
445    /// infrastructure errors fail open (logged via `tracing::warn!`, dispatch
446    /// proceeds).
447    pub fn with_gate(&mut self, gate: GateRef) -> &mut Self {
448        self.gate = gate;
449        self
450    }
451
452    /// Set the namespace surfaced to the gate when a verb does not carry an
453    /// explicit `namespace` argument. Transports should plumb the runtime's
454    /// `default_namespace` so the gate's `input.namespace` always reflects
455    /// the operation's true tenant.
456    pub fn with_default_namespace(&mut self, ns: impl Into<String>) -> &mut Self {
457        self.default_namespace = ns.into();
458        self
459    }
460
461    /// Set the `EventStore` used to persist audit events.
462    ///
463    /// When configured, every gate check appends one `Event` (substrate =
464    /// `Event`, outcome = `Success` on allow, `Denied` on deny) in addition to
465    /// the `tracing::info!` emission that was already present in v0.2.
466    ///
467    /// Callers that do not set this field continue to use tracing-only emission
468    /// (the v0.2 default). There is no behavior change for them.
469    pub fn with_event_store(&mut self, store: Arc<dyn EventStore>) -> &mut Self {
470        self.event_store = Some(store);
471        self
472    }
473
474    /// Register a post-dispatch hook.
475    ///
476    /// When set, every successful pack dispatch calls `hook.on_dispatch(event)`
477    /// with a synthesized [`Event`] describing the verb outcome. The hook is
478    /// opt-in: registries without a hook incur zero overhead on the dispatch
479    /// hot path.
480    ///
481    /// Brain pack uses this to update its posteriors in real time without
482    /// polling the EventStore. Errors from `on_dispatch` are logged via
483    /// `tracing::warn!` and never propagated.
484    pub fn with_dispatch_hook(&mut self, hook: Arc<dyn DispatchHook>) -> &mut Self {
485        self.dispatch_hook = Some(hook);
486        self
487    }
488
489    /// Consume the builder and produce an immutable, cloneable registry.
490    ///
491    /// Performs a topological sort of packs using Kahn's algorithm.
492    /// Returns an error if any declared dependency is missing from the loaded
493    /// pack set, or if a circular dependency is detected.
494    pub fn build(self) -> Result<VerbRegistry, RuntimeError> {
495        let packs = self.packs;
496        let mut name_to_idx: HashMap<&str, usize> = HashMap::with_capacity(packs.len());
497        for (idx, pack) in packs.iter().enumerate() {
498            if let Some(prev_idx) = name_to_idx.insert(pack.name(), idx) {
499                return Err(RuntimeError::PackRedeclared {
500                    name: pack.name().to_string(),
501                    first_idx: prev_idx,
502                    second_idx: idx,
503                });
504            }
505        }
506
507        let mut missing: Vec<MissingPackDependency> = Vec::new();
508        let mut indegree = vec![0usize; packs.len()];
509        let mut dependents: Vec<Vec<usize>> = vec![Vec::new(); packs.len()];
510
511        for (idx, pack) in packs.iter().enumerate() {
512            for &requires in pack.requires() {
513                match name_to_idx.get(requires).copied() {
514                    Some(dep_idx) => {
515                        dependents[dep_idx].push(idx);
516                        indegree[idx] += 1;
517                    }
518                    None => missing.push(MissingPackDependency {
519                        from: pack.name().to_string(),
520                        requires: requires.to_string(),
521                    }),
522                }
523            }
524        }
525
526        if !missing.is_empty() {
527            return if missing.len() == 1 {
528                Err(RuntimeError::MissingPackDependency(missing.remove(0)))
529            } else {
530                Err(RuntimeError::MissingPackDependencies(
531                    MissingPackDependencies { missing },
532                ))
533            };
534        }
535
536        let mut ready: VecDeque<usize> = indegree
537            .iter()
538            .enumerate()
539            .filter_map(|(idx, degree)| (*degree == 0).then_some(idx))
540            .collect();
541        let mut ordered_indices = Vec::with_capacity(packs.len());
542
543        while let Some(idx) = ready.pop_front() {
544            ordered_indices.push(idx);
545            for &dep_idx in &dependents[idx] {
546                indegree[dep_idx] -= 1;
547                if indegree[dep_idx] == 0 {
548                    ready.push_back(dep_idx);
549                }
550            }
551        }
552
553        if ordered_indices.len() != packs.len() {
554            let cycle_nodes: HashSet<usize> = indegree
555                .iter()
556                .enumerate()
557                .filter_map(|(idx, degree)| (*degree > 0).then_some(idx))
558                .collect();
559            let cycle = find_pack_dependency_cycle(&packs, &name_to_idx, &cycle_nodes);
560            return Err(RuntimeError::CircularPackDependency(
561                CircularPackDependency { cycle },
562            ));
563        }
564
565        let mut slots: Vec<Option<Box<dyn PackRuntime>>> = packs.into_iter().map(Some).collect();
566        let ordered_packs: Vec<Box<dyn PackRuntime>> = ordered_indices
567            .into_iter()
568            .map(|idx| slots[idx].take().expect("topological index must exist"))
569            .collect();
570
571        validate_unique_note_kinds(&ordered_packs)?;
572        validate_unique_verb_names(&ordered_packs)?;
573        validate_unique_entity_types(&ordered_packs)?;
574
575        let available_verbs: Vec<&'static str> = ordered_packs
576            .iter()
577            .flat_map(|p| p.handlers().iter())
578            .filter(|h| matches!(h.visibility, Visibility::Verb))
579            .map(|h| h.name)
580            .collect();
581
582        Ok(VerbRegistry {
583            packs: Arc::new(ordered_packs),
584            resolvers: Arc::new(self.resolvers),
585            gate: self.gate,
586            default_namespace: self.default_namespace,
587            visible_namespaces: self.visible_namespaces,
588            actor_id: self.actor_id,
589            event_store: self.event_store,
590            dispatch_hook: self.dispatch_hook,
591            available_verbs: Arc::new(available_verbs),
592            reference_ring: Arc::new(crate::reference_ring::ReferenceRing::new()),
593        })
594    }
595}
596
597/// Validate that no two packs declare the same note kind.
598///
599/// Boot-time duplicate detection prevents pack configuration errors from
600/// silently corrupting note kind routing. Returns an error naming the
601/// duplicate kind and the two packs that claim it.
602fn validate_unique_note_kinds(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
603    let mut seen: HashMap<&str, &str> = HashMap::new();
604    for pack in packs {
605        for &kind in pack.note_kinds() {
606            if let Some(first_pack) = seen.insert(kind, pack.name()) {
607                return Err(RuntimeError::InvalidInput(format!(
608                    "duplicate note kind {kind:?}: claimed by both {first_pack:?} and {:?}",
609                    pack.name()
610                )));
611            }
612        }
613    }
614    Ok(())
615}
616
617/// Validate that no two packs declare the same `Visibility::Verb` handler name.
618///
619/// `Visibility::Subhandler` entries are pack-prefixed by convention and excluded
620/// from cross-pack collision detection. Two packs declaring the same subhandler
621/// name prefix (e.g. `recall.embed`) would be a pack-authoring error but does not
622/// produce a cross-pack routing conflict since only the owning pack dispatches them.
623fn validate_unique_verb_names(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
624    let mut seen: HashMap<&str, &str> = HashMap::new();
625    for pack in packs {
626        for handler in pack.handlers() {
627            if !matches!(handler.visibility, Visibility::Verb) {
628                continue;
629            }
630            if let Some(first_pack) = seen.insert(handler.name, pack.name()) {
631                return Err(RuntimeError::VerbCollision {
632                    verb: handler.name.to_string(),
633                    first_pack: first_pack.to_string(),
634                    second_pack: pack.name().to_string(),
635                });
636            }
637        }
638    }
639    Ok(())
640}
641
642/// Validate that no two owners (the built-in table or a loaded pack) declare
643/// a colliding `entity_type` canonical name or alias.
644///
645/// Boot-time duplicate detection prevents pack configuration errors from
646/// silently applying insertion-order semantics to entity-type resolution
647/// (ADR-001's registry-ownership collision rule: same `(base_kind,
648/// canonical_name)` from two different packs, or an alias collision, is a
649/// boot error). Returns an error naming the colliding key and both
650/// contributing owners.
651fn validate_unique_entity_types(packs: &[Box<dyn PackRuntime>]) -> Result<(), RuntimeError> {
652    let owned_defs = packs
653        .iter()
654        .flat_map(|p| p.entity_types().iter().map(move |def| (p.name(), def)));
655    khive_types::EntityTypeRegistry::check_extra_collisions(owned_defs)
656        .map_err(RuntimeError::InvalidInput)
657}
658
659fn find_pack_dependency_cycle(
660    packs: &[Box<dyn PackRuntime>],
661    name_to_idx: &HashMap<&str, usize>,
662    cycle_nodes: &HashSet<usize>,
663) -> Vec<String> {
664    fn visit(
665        idx: usize,
666        packs: &[Box<dyn PackRuntime>],
667        name_to_idx: &HashMap<&str, usize>,
668        cycle_nodes: &HashSet<usize>,
669        visiting: &mut Vec<usize>,
670        visited: &mut HashSet<usize>,
671    ) -> Option<Vec<String>> {
672        if let Some(pos) = visiting.iter().position(|&seen| seen == idx) {
673            let mut cycle: Vec<String> = visiting[pos..]
674                .iter()
675                .map(|&i| packs[i].name().to_string())
676                .collect();
677            cycle.push(packs[idx].name().to_string());
678            return Some(cycle);
679        }
680        if !visited.insert(idx) {
681            return None;
682        }
683        visiting.push(idx);
684        for &req in packs[idx].requires() {
685            let Some(&dep_idx) = name_to_idx.get(req) else {
686                continue;
687            };
688            if cycle_nodes.contains(&dep_idx) {
689                if let Some(cycle) =
690                    visit(dep_idx, packs, name_to_idx, cycle_nodes, visiting, visited)
691                {
692                    return Some(cycle);
693                }
694            }
695        }
696        visiting.pop();
697        None
698    }
699
700    let mut visited = HashSet::new();
701    for &idx in cycle_nodes {
702        let mut visiting = Vec::new();
703        if let Some(cycle) = visit(
704            idx,
705            packs,
706            name_to_idx,
707            cycle_nodes,
708            &mut visiting,
709            &mut visited,
710        ) {
711            return cycle;
712        }
713    }
714    cycle_nodes
715        .iter()
716        .map(|&idx| packs[idx].name().to_string())
717        .collect()
718}
719
720impl Default for VerbRegistryBuilder {
721    fn default() -> Self {
722        Self::new()
723    }
724}
725
726/// Immutable registry that dispatches verb calls to registered packs.
727///
728/// Clone is cheap (Arc-wrapped). Constructed via `VerbRegistryBuilder`.
729#[derive(Clone)]
730pub struct VerbRegistry {
731    packs: std::sync::Arc<Vec<Box<dyn PackRuntime>>>,
732    /// Pack-level by-ID resolvers, in registration order.
733    resolvers: std::sync::Arc<Vec<(String, Box<dyn PackByIdResolver>)>>,
734    gate: GateRef,
735    default_namespace: String,
736    /// Operator-configured read-visibility set (ADR-007 Rev 4 Rule 3b).
737    ///
738    /// On the default (no explicit `namespace=` param) dispatch path, reads fan
739    /// out over `['local'] ∪ visible_namespaces`. Writes are unaffected — they
740    /// still pin to `'local'`. An explicit `namespace=` request param is a
741    /// precise single-namespace escape and is not widened by this set.
742    visible_namespaces: Vec<Namespace>,
743    /// Configured actor identity label (ADR-057). When `Some`, dispatch mints
744    /// tokens carrying this actor so that `comm.inbox` applies the `to_actor`
745    /// filter. When `None`, tokens carry `ActorRef::anonymous()` (party-line).
746    actor_id: Option<String>,
747    /// Audit event sink — `None` means tracing-only (v0.2 default).
748    event_store: Option<Arc<dyn EventStore>>,
749    /// Post-dispatch hook: `None` means no real-time observation.
750    dispatch_hook: Option<Arc<dyn DispatchHook>>,
751    /// Names of all `Visibility::Verb` handlers across all packs, precomputed
752    /// once at `build()` time. Used only to render the unknown-verb error
753    /// message — the pack set is fixed after construction, so there is no
754    /// need to re-scan every pack's handlers on every miss.
755    available_verbs: Arc<Vec<&'static str>>,
756    /// Recently-referenced ring (unified-verb draft ADR, Slice 1). Daemon-warm,
757    /// actor-scoped, never persisted — see `crate::reference_ring`. Shared
758    /// across every clone of this registry via the `Arc`, so admissions made
759    /// by one dispatch are visible to the next on the same warm daemon.
760    reference_ring: Arc<crate::reference_ring::ReferenceRing>,
761}
762
763/// Per-request identity context that overrides a [`VerbRegistry`]'s
764/// construction-baked `default_namespace` / `actor_id` / `visible_namespaces`
765/// for exactly one [`VerbRegistry::dispatch_with_identity`] call (ADR-096
766/// Fork 1 — warm-daemon per-request identity).
767///
768/// A single warm registry is built once with a baked identity, but must be
769/// able to serve requests whose caller resolved a *different* attribution
770/// identity (e.g. a different project-local `[actor]`) without a cold
771/// fallback and without mis-stamping writes under the registry's own baked
772/// actor. Supplying `Some(RequestIdentity { .. })` threads the caller's
773/// identity through token minting for that one call; the registry's fields
774/// (and every other in-flight call) are untouched. `None` is exactly
775/// [`VerbRegistry::dispatch`] — the baked scalars apply, unchanged from
776/// before this type existed.
777#[derive(Debug, Clone, Default)]
778pub struct RequestIdentity {
779    /// Storage/gate default namespace for this request (used when the verb's
780    /// own params carry no explicit `namespace` field). Overrides
781    /// `VerbRegistry::default_namespace`.
782    pub namespace: String,
783    /// Write-stamp / gate actor label for this request (ADR-057). Overrides
784    /// `VerbRegistry::actor_id`. `None` mints `ActorRef::anonymous()`, same
785    /// as an unconfigured baked `actor_id`.
786    pub actor_id: Option<String>,
787    /// Extra read-visibility namespaces for this request (ADR-007 Rev 4 Rule
788    /// 3b). Overrides `VerbRegistry::visible_namespaces`. Entries that fail
789    /// `Namespace::parse` are skipped with a `tracing::warn!` rather than
790    /// failing the whole request — a single malformed visibility entry from a
791    /// caller-supplied frame must not block dispatch.
792    pub visible_namespaces: Vec<String>,
793    /// Caller-supplied correlation id for this request (khive#948), carried
794    /// unchanged from the daemon frame's `request_id` field. Stamped into the
795    /// audit event's `resource.request_id` on every outcome (success, error,
796    /// and denied) so a client can join its own pre-send sample to the
797    /// server-side audit row for the same request. `None` means the caller
798    /// supplied no id (a pre-#948 client, or an internal/non-benchmark
799    /// caller) — the audit row then carries no `request_id` key at all.
800    pub request_id: Option<u64>,
801}
802
803/// A non-blank, out-of-band authenticated principal for [`VerbRegistry::dispatch_as`].
804///
805/// Embedding hosts authenticate a principal through their own channel (not the
806/// request DSL) and then need that principal to become the effective actor
807/// for one dispatch. The constructor rejects an empty or whitespace-only
808/// identifier so an authentication-integration failure (an empty subject)
809/// fails closed at construction time instead of silently resolving to the
810/// anonymous/local actor at dispatch time — see [`crate::actor_identity::resolve_actor`].
811#[derive(Debug, Clone, PartialEq, Eq)]
812pub struct VerifiedActor(String);
813
814impl VerifiedActor {
815    /// Validate and wrap a verified principal identifier.
816    ///
817    /// Returns `RuntimeError::InvalidInput` when `id` is empty or contains
818    /// only whitespace.
819    pub fn new(id: impl Into<String>) -> Result<Self, RuntimeError> {
820        let id = id.into();
821        if id.trim().is_empty() {
822            return Err(RuntimeError::InvalidInput(
823                "VerifiedActor: identifier must not be empty or whitespace-only".to_string(),
824            ));
825        }
826        Ok(Self(id))
827    }
828
829    /// Borrow the validated identifier.
830    pub fn as_str(&self) -> &str {
831        &self.0
832    }
833
834    fn into_inner(self) -> String {
835        self.0
836    }
837}
838
839/// Error returned by [`VerbRegistry::apply_schema_plans_with_map`] when two
840/// packs on the same backend declare the same auxiliary table (ADR-028 §7).
841#[derive(Debug)]
842pub struct PackSchemaCollisionError {
843    /// First pack to declare the table.
844    pub pack_a: &'static str,
845    /// Second pack that collides with `pack_a`.
846    pub pack_b: &'static str,
847    /// Table name or DDL error description.
848    pub table: String,
849}
850
851impl std::fmt::Display for PackSchemaCollisionError {
852    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
853        if self.pack_a == self.pack_b {
854            write!(
855                f,
856                "pack schema boot failure for pack {:?}: {}",
857                self.pack_a, self.table
858            )
859        } else {
860            write!(
861                f,
862                "pack schema collision: packs {:?} and {:?} both declare table {:?} \
863                 on the same backend — move one pack to a separate backend or rename the table",
864                self.pack_a, self.pack_b, self.table
865            )
866        }
867    }
868}
869
870impl std::error::Error for PackSchemaCollisionError {}
871
872/// Extract table names from a single DDL statement.
873///
874/// Handles `CREATE TABLE IF NOT EXISTS`, `CREATE TABLE`, and
875/// `CREATE VIRTUAL TABLE IF NOT EXISTS`, `CREATE VIRTUAL TABLE`.
876/// Returns an empty Vec when no table name is found (e.g. index DDL).
877fn extract_table_names(stmt: &str) -> Vec<String> {
878    let normalized = stmt.split_whitespace().collect::<Vec<_>>().join(" ");
879    let upper = normalized.to_ascii_uppercase();
880    let table_name = if let Some(rest) = upper.strip_prefix("CREATE VIRTUAL TABLE IF NOT EXISTS ") {
881        rest.split_whitespace().next()
882    } else if let Some(rest) = upper.strip_prefix("CREATE VIRTUAL TABLE ") {
883        rest.split_whitespace().next()
884    } else if let Some(rest) = upper.strip_prefix("CREATE TABLE IF NOT EXISTS ") {
885        rest.split_whitespace().next()
886    } else if let Some(rest) = upper.strip_prefix("CREATE TABLE ") {
887        rest.split_whitespace().next()
888    } else {
889        None
890    };
891    match table_name {
892        Some(name) => {
893            let clean = name.trim_matches(|c: char| c == '(' || c == ';');
894            if clean.is_empty() {
895                vec![]
896            } else {
897                vec![clean.to_ascii_lowercase()]
898            }
899        }
900        None => vec![],
901    }
902}
903
904/// Render an [`EndpointKind`] as the `"<substrate>:<kind>"` label used in
905/// `link(help=true)`'s `endpoint_rules` table.
906fn endpoint_kind_label(kind: &EndpointKind) -> String {
907    match kind {
908        EndpointKind::EntityOfKind(k) => format!("entity:{k}"),
909        EndpointKind::NoteOfKind(k) => format!("note:{k}"),
910        EndpointKind::EntityOfType { kind, entity_type } => {
911            format!("entity:{kind}({entity_type})")
912        }
913    }
914}
915
916/// Relations `validate_edge_relation_endpoints`
917/// (`crates/khive-runtime/src/operations.rs`) resolves in its own dedicated
918/// branch — before the generic pack-rule branch (`pack_rule_allows`) is ever
919/// reached. For these three relations the validator additionally accepts
920/// any `note -> note` pair unconditionally, regardless of note kind
921/// (ADR-002 §"Versioning" and §"Epistemic"), and never consults pack
922/// `EDGE_RULES` at all, on either substrate.
923const SPECIAL_RELATIONS: &[khive_types::EdgeRelation] = &[
924    khive_types::EdgeRelation::Supersedes,
925    khive_types::EdgeRelation::Supports,
926    khive_types::EdgeRelation::Refutes,
927];
928
929/// Compose the full per-relation endpoint allowlist surfaced by
930/// `link(help=true)` (issue #964).
931///
932/// Combines the base entity-to-entity endpoint contract
933/// (`operations::base_entity_endpoint_rules`) with every loaded pack's
934/// additive `EDGE_RULES`, the unconditional `note -> note` allowance for the
935/// three special relations (`supersedes` / `supports` / `refutes` —
936/// `operations.rs`'s dedicated special-relation branch), and the
937/// `annotates` note-to-any special case — the exact same sources
938/// `valid_relations_for_entity_pair` (`khive-pack-kg`) consults when
939/// enriching a rejected `link` call, so a caller reading this table cannot
940/// diverge from what the validator itself accepts.
941///
942/// Pack `EDGE_RULES` for a special relation are deliberately excluded: the
943/// validator's special-relation branch returns before `pack_rule_allows` is
944/// ever reached (`operations.rs`), so advertising such a rule here would
945/// claim enforcement that never actually happens.
946fn edge_endpoint_table(packs: &[Box<dyn PackRuntime>]) -> Vec<Value> {
947    let mut rows: Vec<Value> = crate::operations::base_entity_endpoint_rules()
948        .iter()
949        .map(|(src, rel, tgt)| {
950            serde_json::json!({
951                "relation": rel.as_str(),
952                "source": format!("entity:{src}"),
953                "target": format!("entity:{tgt}"),
954            })
955        })
956        .collect();
957
958    for rel in SPECIAL_RELATIONS {
959        rows.push(serde_json::json!({
960            "relation": rel.as_str(),
961            "source": "note:*",
962            "target": "note:*",
963        }));
964    }
965
966    for pack in packs.iter() {
967        for rule in pack.edge_rules().iter() {
968            if SPECIAL_RELATIONS.contains(&rule.relation) {
969                continue;
970            }
971            rows.push(serde_json::json!({
972                "relation": rule.relation.as_str(),
973                "source": endpoint_kind_label(&rule.source),
974                "target": endpoint_kind_label(&rule.target),
975            }));
976        }
977    }
978
979    rows.push(serde_json::json!({
980        "relation": "annotates",
981        "source": "note:*",
982        "target": "any (entity, note, edge, or event)",
983    }));
984
985    rows
986}
987
988impl VerbRegistry {
989    /// This registry's construction-baked default namespace.
990    ///
991    /// Used as the fallback when a request carries no [`RequestIdentity`]
992    /// override (ADR-096 Fork 1) and by transports that need to advertise
993    /// their own resolved identity when forwarding to a warm daemon.
994    pub fn default_namespace(&self) -> &str {
995        &self.default_namespace
996    }
997
998    /// This registry's construction-baked actor identity label, if configured
999    /// (ADR-057). `None` means dispatch mints `ActorRef::anonymous()` absent a
1000    /// per-request [`RequestIdentity`] override (ADR-096 Fork 1).
1001    pub fn actor_id(&self) -> Option<&str> {
1002        self.actor_id.as_deref()
1003    }
1004
1005    /// This registry's construction-baked extra read-visibility namespaces
1006    /// (ADR-007 Rev 4 Rule 3b), used absent a per-request [`RequestIdentity`]
1007    /// override (ADR-096 Fork 1).
1008    pub fn visible_namespaces(&self) -> &[Namespace] {
1009        &self.visible_namespaces
1010    }
1011
1012    /// This registry's configured audit `EventStore`, if any (ADR-094).
1013    ///
1014    /// Lets background tasks that hold a `VerbRegistry` but do not go through
1015    /// `dispatch` (e.g. the email channel poll loop) append best-effort
1016    /// lifecycle events to the same sink gate-check audit rows use, without
1017    /// threading a second `Option<Arc<dyn EventStore>>` field through every
1018    /// caller. `None` means tracing-only, matching the registry's own
1019    /// audit-persistence default.
1020    pub fn event_store(&self) -> Option<Arc<dyn EventStore>> {
1021        self.event_store.clone()
1022    }
1023
1024    /// Whether any loaded pack registers `verb`, at any visibility.
1025    ///
1026    /// Internal callers use this to skip optional cross-pack dispatches
1027    /// (e.g. serving-profile resolution) when the providing pack is not in
1028    /// the loaded set, instead of paying a failed dispatch plus its audit
1029    /// write on every call.
1030    pub fn has_verb(&self, verb: &str) -> bool {
1031        self.packs
1032            .iter()
1033            .flat_map(|p| p.handlers().iter())
1034            .any(|h| h.name == verb)
1035    }
1036
1037    /// Return the help schema envelope for a verb.
1038    ///
1039    /// Walks registered packs for the first matching `HandlerDef` and returns a
1040    /// structured JSON envelope. Subhandlers carry `callable_via_mcp: false`.
1041    /// `link`'s envelope additionally carries `endpoint_rules` — the composed
1042    /// per-relation source/target allowlist (issue #964) — so batch callers can
1043    /// defer to the kernel's own table instead of re-implementing it locally.
1044    /// Unknown verbs return `RuntimeError::InvalidInput`. Full shape documented
1045    /// in `docs/protocol.md` §Request Schema.
1046    pub fn describe_verb(&self, verb: &str) -> Result<Value, RuntimeError> {
1047        for pack in self.packs.iter() {
1048            for handler in pack.handlers().iter() {
1049                if handler.name == verb {
1050                    let category = format!("{:?}", handler.category);
1051                    let params_arr: Vec<Value> = handler
1052                        .params
1053                        .iter()
1054                        .map(|p| {
1055                            serde_json::json!({
1056                                "name": p.name,
1057                                "type": p.param_type,
1058                                "required": p.required,
1059                                "description": p.description,
1060                            })
1061                        })
1062                        .collect();
1063                    // Subhandlers are not callable via the MCP request surface;
1064                    // the help payload must match the behaviour the dispatch
1065                    // path enforces so callers reading `help=true` before
1066                    // probing see accurate availability.
1067                    if matches!(handler.visibility, Visibility::Subhandler) {
1068                        return Ok(serde_json::json!({
1069                            "verb": verb,
1070                            "pack": pack.name(),
1071                            "description": handler.description,
1072                            "category": category,
1073                            "params": params_arr,
1074                            "visibility": "internal",
1075                            "callable_via_mcp": false,
1076                            "note": "This is an internal subhandler. Calling it via the MCP \
1077                                     request surface returns permission denied. It can only be \
1078                                     invoked by internal runtime callers.",
1079                        }));
1080                    }
1081                    let mut envelope = serde_json::json!({
1082                        "verb": verb,
1083                        "pack": pack.name(),
1084                        "description": handler.description,
1085                        "category": category,
1086                        "params": params_arr,
1087                    });
1088                    if verb == "link" {
1089                        envelope["endpoint_rules"] = Value::Array(edge_endpoint_table(&self.packs));
1090                    }
1091                    return Ok(envelope);
1092                }
1093            }
1094        }
1095        // Verb-visibility handler names, precomputed at build() time (internal
1096        // subhandlers are excluded so they are not advertised in the
1097        // unknown-verb error).
1098        Err(RuntimeError::InvalidInput(format!(
1099            "unknown verb {verb:?}; available: {}",
1100            self.available_verbs.join(", ")
1101        )))
1102    }
1103
1104    /// Check whether the gate permits writes into `ns`.
1105    ///
1106    /// Performs a gate evaluation with verb `"authorize"` before any background
1107    /// loop is spawned (ADR-056 §6).  Returns `Ok(())` when the gate allows the
1108    /// namespace, or `Err(RuntimeError::PermissionDenied{..})` when denied.
1109    /// Gate errors (implementation failures) are surfaced as
1110    /// `RuntimeError::Internal`.
1111    pub fn authorize_namespace(&self, ns: Namespace) -> Result<(), RuntimeError> {
1112        let actor = crate::actor_identity::resolve_actor(self.actor_id.as_deref());
1113        let req = GateRequest::new(actor, ns, "authorize", serde_json::Value::Null);
1114        match self.gate.check(&req) {
1115            Ok(decision) if decision.is_allow() => Ok(()),
1116            Ok(GateDecision::Deny { reason }) => Err(RuntimeError::PermissionDenied {
1117                verb: "authorize".to_string(),
1118                reason,
1119            }),
1120            Ok(_) => Err(RuntimeError::PermissionDenied {
1121                verb: "authorize".to_string(),
1122                reason: "gate denied".to_string(),
1123            }),
1124            Err(e) => Err(RuntimeError::Internal(format!("gate error: {e}"))),
1125        }
1126    }
1127
1128    /// Dispatch a verb to the first pack that handles it.
1129    ///
1130    /// Routes through the gate, then invokes the matching pack handler. When
1131    /// `params["help"] == true`, short-circuits to `describe_verb` with no side effects.
1132    /// Gate errors are fail-open. Full dispatch flow documented in `docs/protocol.md`.
1133    ///
1134    /// Equivalent to `self.dispatch_with_identity(verb, params, None)` — uses
1135    /// this registry's construction-baked `default_namespace` / `actor_id` /
1136    /// `visible_namespaces`.
1137    pub async fn dispatch(&self, verb: &str, params: Value) -> Result<Value, RuntimeError> {
1138        self.dispatch_with_identity(verb, params, None).await
1139    }
1140
1141    /// Dispatch a verb, optionally overriding this registry's baked identity
1142    /// scalars for exactly this call (ADR-096 Fork 1).
1143    ///
1144    /// `identity = None` behaves exactly like [`Self::dispatch`]. `identity =
1145    /// Some(id)` uses `id.namespace` / `id.actor_id` / `id.visible_namespaces`
1146    /// in place of `self.default_namespace` / `self.actor_id` /
1147    /// `self.visible_namespaces` for this call's namespace resolution, gate
1148    /// request, and token minting. The registry's own fields are never mutated,
1149    /// so concurrent calls with different (or no) identity are independent.
1150    /// See `docs/api/pack.md#dispatch_with_identity` for why this enables one warm
1151    /// registry to serve many attribution identities over a shared backend.
1152    pub async fn dispatch_with_identity(
1153        &self,
1154        verb: &str,
1155        params: Value,
1156        identity: Option<RequestIdentity>,
1157    ) -> Result<Value, RuntimeError> {
1158        // help=true interception: short-circuit before gate/pack.
1159        if params.get("help").and_then(Value::as_bool) == Some(true) {
1160            return self.describe_verb(verb);
1161        }
1162        // Resolve namespace before `params` is moved into pack.dispatch, so the
1163        // post-dispatch hook can reference it.
1164        //
1165        // Absent `namespace` and a present-but-malformed `namespace` are
1166        // different cases. A present non-string value (null, number, bool,
1167        // array, object) is explicit caller input that failed to parse and
1168        // must fail closed, not silently coerce to the default namespace.
1169        // Only a genuinely absent key defaults. Shared with the multi-backend
1170        // coordinator intercept via `resolve_explicit_namespace` so every MCP
1171        // ingress path applies the same fail-closed rule.
1172        let explicit_namespace = params.get("namespace").is_some_and(Value::is_string);
1173        // The caller-supplied correlation id (khive#948), if any. Read once
1174        // here so it is in scope for every audit-append site below,
1175        // including the ones that run before pack dispatch is attempted.
1176        let request_id: Option<u64> = identity.as_ref().and_then(|id| id.request_id);
1177        // A supplied per-request identity overrides the baked
1178        // default_namespace/actor_id/visible_namespaces for this call only.
1179        let default_namespace_str: &str = identity
1180            .as_ref()
1181            .map(|id| id.namespace.as_str())
1182            .unwrap_or(self.default_namespace.as_str());
1183        let ns = resolve_explicit_namespace(&params, default_namespace_str)?;
1184        let actor_id_str: Option<&str> = match identity.as_ref() {
1185            Some(id) => id.actor_id.as_deref(),
1186            None => self.actor_id.as_deref(),
1187        };
1188        // Thread the configured actor identity into the gate request so the
1189        // gate can distinguish human vs agent callers at the dispatch seam.
1190        // Resolved once via the shared actor-identity policy and reused for
1191        // token minting below, so the gate's notion of "who is the caller"
1192        // and the storage token's notion can never drift apart.
1193        let resolved_actor = crate::actor_identity::resolve_actor(actor_id_str);
1194        let gate_req = GateRequest::new(resolved_actor.clone(), ns.clone(), verb, params.clone());
1195
1196        // Consult the gate.
1197        //
1198        // - Ok(Allow) → proceed to pack dispatch (tracing + optional EventStore).
1199        // - Ok(Deny) → emit audit, persist if store configured, return PermissionDenied.
1200        // - Err(_) → warn via tracing, fail-open (no audit persisted).
1201        let (gate_blocked, mut deferred_audit) = match self.gate.check(&gate_req) {
1202            Ok(decision) => {
1203                let is_deny = matches!(decision, GateDecision::Deny { .. });
1204
1205                // Emit audit event via tracing.
1206                let audit = AuditEvent::from_check(&gate_req, &decision, self.gate.impl_name());
1207                tracing::info!(
1208                    audit_event = %serde_json::to_string(&audit)
1209                        .unwrap_or_else(|_| "{\"error\":\"serialize\"}".into()),
1210                    "gate.check"
1211                );
1212
1213                // Drain any process-lifetime `OnceLock` config locks queued
1214                // since the last dispatch and persist them as `ConfigLocked`
1215                // events, riding this same audit-persistence gate. The
1216                // namespace/actor stamped on these rows are whichever
1217                // dispatch happens to observe the queue non-empty first:
1218                // an accepted provenance quirk, preferred over threading an
1219                // `EventStore` handle into every synchronous
1220                // `OnceLock::get_or_init` call site.
1221                if let Some(store) = &self.event_store {
1222                    if crate::config_ledger::PENDING
1223                        .swap(false, std::sync::atomic::Ordering::AcqRel)
1224                    {
1225                        for (key, value) in crate::config_ledger::drain_config_locked() {
1226                            let payload = serde_json::json!({ "key": key, "value": value });
1227                            let storage_event = Event::new(
1228                                gate_req.namespace.as_str(),
1229                                verb,
1230                                EventKind::ConfigLocked,
1231                                SubstrateKind::Event,
1232                                format!("{}:{}", gate_req.actor.kind, gate_req.actor.id),
1233                            )
1234                            .with_payload(payload);
1235                            append_audit_event_best_effort(store, storage_event, verb).await;
1236                        }
1237                    }
1238                }
1239
1240                // Every Allow-outcome audit row defers its append until pack
1241                // dispatch returns, so the row can carry the measured
1242                // dispatch time in `duration_us` (persisting before dispatch
1243                // ran always recorded the `Event::new` default of 0). A
1244                // singleton `link` call (no `links` bulk array) additionally
1245                // enriches the deferred row with the created/resolved edge
1246                // fields (schema v2) once dispatch resolves. Denied calls
1247                // have no dispatch to wait for and keep the immediate v1
1248                // append below.
1249                //
1250                // Accepted trade-off: a crash between this Allow decision and
1251                // the deferred append below (post-dispatch, further down this
1252                // function) loses that dispatch's audit row entirely: a
1253                // deliberate choice, not an oversight.
1254                let defer_audit = !is_deny;
1255
1256                // Persist to EventStore immediately only for denied calls.
1257                if !defer_audit {
1258                    if let Some(store) = &self.event_store {
1259                        // ADR-103 Decision (a): the closed `work_class` enum
1260                        // is stamped on every event, denial included -- only
1261                        // `resource.cost_unit` is scoped to a successful
1262                        // dispatch by Amendment 1. `base_resource_payload()`
1263                        // carries `work_class` alone, no `cost_unit` key.
1264                        let storage_event = build_audit_storage_event(
1265                            &gate_req,
1266                            &audit,
1267                            EventOutcome::Denied,
1268                            Some(crate::cost_unit::base_resource_payload(request_id)),
1269                        );
1270                        append_audit_event_best_effort(store, storage_event, verb).await;
1271                    }
1272                }
1273
1274                let reason = if is_deny {
1275                    let reason = match decision {
1276                        GateDecision::Deny { reason } => reason,
1277                        _ => String::new(),
1278                    };
1279                    Some(reason)
1280                } else {
1281                    None
1282                };
1283                let deferred = if defer_audit { Some(audit) } else { None };
1284                (reason, deferred)
1285            }
1286            Err(err) => {
1287                // Gate infrastructure failure — fail-open.
1288                // No decision was produced; no audit event is persisted.
1289                tracing::warn!(verb, error = %err, "gate check failed (fail-open)");
1290                (None, None)
1291            }
1292        };
1293
1294        // Hard enforcement: Deny is authoritative.
1295        if let Some(reason) = gate_blocked {
1296            return Err(RuntimeError::PermissionDenied {
1297                verb: verb.to_string(),
1298                reason,
1299            });
1300        }
1301
1302        // Mint the authorized storage token at the dispatch boundary.
1303        //
1304        // Writes pin to `local` by default. Actor identity and config
1305        // `[actor] id` are attribution and gate-context inputs only: they
1306        // never route storage. The explicit `namespace=` request param is a
1307        // precise single-namespace escape: the caller deliberately
1308        // reads/writes exactly that one set; it is NOT widened by `visible_namespaces`.
1309        //
1310        // When actor_id is configured, mint a token carrying that actor
1311        // label so that comm.inbox applies the to_actor filter for directed delivery.
1312        // Otherwise, use ActorRef::anonymous() and inbox falls back to party-line.
1313        // `actor_id_str` already reflects the per-request identity override
1314        // when supplied (resolved above into `resolved_actor`, mirrored into
1315        // the gate request). Reusing the same value here guarantees the
1316        // gate's actor and the storage token's actor can never diverge.
1317        //
1318        // On the default (no explicit `namespace=`) path, the read scope
1319        // widens to `['local'] ∪ visible_namespaces` (baked, or the
1320        // per-request override). `'local'` is always included
1321        // (mint_with_visibility deduplicates). Writes remain pinned to
1322        // `'local'`. Per-actor distinctions use view-layer tag filters
1323        // (assignee, actor_id, from/to), not namespace partitions. `ns`/
1324        // `explicit_namespace` were already validated above: reuse them
1325        // instead of re-reading `params["namespace"]` with `as_str()`, which
1326        // would silently drop malformed non-string values again.
1327        let token = if explicit_namespace {
1328            // Explicit escape: precise single-namespace scope, read+write. NOT widened.
1329            NamespaceToken::mint_with_visibility(ns.clone(), vec![], resolved_actor)
1330        } else {
1331            // Default path: write namespace = local; read scope = ['local'] ∪ visible_namespaces.
1332            let primary = Namespace::local();
1333            let mut extra_visible: Vec<Namespace> = match identity.as_ref() {
1334                Some(id) => id
1335                    .visible_namespaces
1336                    .iter()
1337                    .filter_map(|s| match Namespace::parse(s) {
1338                        Ok(parsed) => Some(parsed),
1339                        Err(e) => {
1340                            tracing::warn!(
1341                                namespace = %s,
1342                                error = %e,
1343                                "dispatch_with_identity: skipping invalid visible_namespace \
1344                                 entry from per-request identity"
1345                            );
1346                            None
1347                        }
1348                    })
1349                    .collect(),
1350                None => self.visible_namespaces.clone(),
1351            };
1352            extra_visible.push(Namespace::local()); // 'local' always readable; mint dedups
1353            NamespaceToken::mint_with_visibility(primary, extra_visible, resolved_actor)
1354        };
1355
1356        for pack in self.packs.iter() {
1357            if let Some(handler_def) = pack.handlers().iter().find(|v| v.name == verb) {
1358                // Strip `namespace` from params before forwarding to packs.
1359                // The registry has already consumed it to mint the NamespaceToken.
1360                //
1361                // Exception: if the handler's own `params` schema declares
1362                // `"namespace"` as a valid field (e.g. brain.bind, brain.unbind,
1363                // brain.bindings, brain.resolve), the field is a *business* argument
1364                // — not a transport routing key — and must be passed through
1365                // unchanged. Stripping it would silently default the binding to the
1366                // "*" wildcard, broadening profile scope across namespaces.
1367                let handler_accepts_namespace =
1368                    handler_def.params.iter().any(|p| p.name == "namespace");
1369                let params = if !handler_accepts_namespace {
1370                    if let Value::Object(mut map) = params {
1371                        map.remove("namespace");
1372                        Value::Object(map)
1373                    } else {
1374                        params
1375                    }
1376                } else {
1377                    params
1378                };
1379                let dispatch_start = Instant::now();
1380                let result = pack.dispatch(verb, params, self, &token).await;
1381                let dispatch_us = dispatch_start.elapsed().as_micros() as i64;
1382
1383                // Append the deferred Allow-outcome audit row now that
1384                // dispatch has resolved, so `duration_us` carries the
1385                // measured `dispatch_us` instead of the `Event::new` default
1386                // of 0. A successful singleton `link` call enriches the row
1387                // with the created/resolved edge (schema v2); anything that
1388                // cannot be enriched, or is not a singleton `link` call,
1389                // falls back to the generic v1 audit shape so no audit row
1390                // is ever dropped for the deferred path.
1391                if let Some(audit) = deferred_audit.take() {
1392                    if let Some(store) = &self.event_store {
1393                        let is_link_singleton =
1394                            verb == "link" && gate_req.args.get("links").is_none();
1395                        match &result {
1396                            Ok(ok_val) if is_link_singleton => {
1397                                // ADR-103 Amendment 1: `link` (singleton or
1398                                // bulk) has no embedding-bearing path — edges
1399                                // carry no embedded body — so cost_unit is
1400                                // always base_weight("link") alone. The
1401                                // registered-model closure is never invoked
1402                                // (per_item_weight("link", ..) short-circuits
1403                                // to 0 before `model_count` reads it).
1404                                let resource = crate::cost_unit::resource_payload(
1405                                    verb,
1406                                    &gate_req.args,
1407                                    ok_val,
1408                                    || pack.registered_embedding_model_names().len() as i64,
1409                                    request_id,
1410                                );
1411                                match link_audit_success_from_result(audit.clone(), ok_val) {
1412                                    Some((edge_id, mut payload)) => {
1413                                        if let Value::Object(ref mut map) = payload {
1414                                            map.insert("resource".to_string(), resource);
1415                                        }
1416                                        let storage_event = Event::new(
1417                                            gate_req.namespace.as_str(),
1418                                            gate_req.verb.as_str(),
1419                                            EventKind::Audit,
1420                                            SubstrateKind::Event,
1421                                            format!(
1422                                                "{}:{}",
1423                                                gate_req.actor.kind, gate_req.actor.id
1424                                            ),
1425                                        )
1426                                        .with_outcome(EventOutcome::Success)
1427                                        .with_target(edge_id)
1428                                        .with_payload(payload)
1429                                        .with_payload_schema_version(2)
1430                                        .with_duration_us(dispatch_us);
1431                                        append_audit_event_best_effort(store, storage_event, verb)
1432                                            .await;
1433                                    }
1434                                    None => {
1435                                        tracing::warn!(
1436                                            verb,
1437                                            "link audit v2 enrichment parse failed; \
1438                                             falling back to v1 audit shape"
1439                                        );
1440                                        let storage_event = build_audit_storage_event(
1441                                            &gate_req,
1442                                            &audit,
1443                                            EventOutcome::Success,
1444                                            Some(resource),
1445                                        )
1446                                        .with_duration_us(dispatch_us);
1447                                        append_audit_event_best_effort(store, storage_event, verb)
1448                                            .await;
1449                                    }
1450                                }
1451                            }
1452                            _ => {
1453                                // The persisted audit outcome must reflect
1454                                // the dispatch result, not be hardcoded to
1455                                // Success — otherwise a failed dispatch is
1456                                // recorded as successful work and disappears
1457                                // from `outcome=error` queries.
1458                                //
1459                                // ADR-103 Amendment 1: `resource.cost_unit` is
1460                                // computed ONLY on a successful dispatch —
1461                                // there is no handler `Value` to read
1462                                // `item_count` from on an error, and the
1463                                // amendment's "absence has exactly two
1464                                // meanings" rule requires the field be
1465                                // omitted, never defaulted to 0, on an
1466                                // errored dispatch. `work_class` itself is
1467                                // NOT one of those two omission cases
1468                                // (ADR-103 Decision (a) stamps it on every
1469                                // event), so an errored dispatch still gets
1470                                // `resource: {"work_class": "interactive"}`,
1471                                // just with no `cost_unit` key.
1472                                let (outcome, resource) = match &result {
1473                                    Ok(ok_val) => (
1474                                        EventOutcome::Success,
1475                                        Some(crate::cost_unit::resource_payload(
1476                                            verb,
1477                                            &gate_req.args,
1478                                            ok_val,
1479                                            || pack.registered_embedding_model_names().len() as i64,
1480                                            request_id,
1481                                        )),
1482                                    ),
1483                                    Err(_) => (
1484                                        EventOutcome::Error,
1485                                        Some(crate::cost_unit::base_resource_payload(request_id)),
1486                                    ),
1487                                };
1488                                let storage_event =
1489                                    build_audit_storage_event(&gate_req, &audit, outcome, resource)
1490                                        .with_duration_us(dispatch_us);
1491                                append_audit_event_best_effort(store, storage_event, verb).await;
1492                            }
1493                        }
1494                    }
1495                }
1496
1497                // Post-dispatch hook: fires on success, opt-in.
1498                if let (Ok(ref ok_val), Some(hook)) = (&result, &self.dispatch_hook) {
1499                    let mut dispatch_event = Event::new(
1500                        ns.as_str(),
1501                        verb,
1502                        EventKind::Audit,
1503                        SubstrateKind::Event,
1504                        pack.name(),
1505                    )
1506                    .with_outcome(EventOutcome::Success)
1507                    .with_duration_us(dispatch_us);
1508
1509                    // For recall verbs: extract the first result's id as
1510                    // target_id so the brain temporal posterior can observe
1511                    // real hit/miss and latency.
1512                    if verb == "memory.recall" {
1513                        let first_note_id = ok_val
1514                            .as_array()
1515                            .and_then(|arr| arr.first())
1516                            .and_then(|v| v.get("id"))
1517                            .and_then(|v| v.as_str())
1518                            .and_then(|s| s.parse::<uuid::Uuid>().ok());
1519                        if let Some(note_id) = first_note_id {
1520                            dispatch_event = dispatch_event.with_target(note_id);
1521                        }
1522                        // No first result → target_id stays None (RecallMiss
1523                        // in brain's event interpreter).
1524                    }
1525
1526                    let dispatch_view = EventView {
1527                        event: dispatch_event,
1528                        observations: Vec::new(),
1529                    };
1530                    let hook = Arc::clone(hook);
1531                    hook.on_dispatch(&dispatch_view).await;
1532                }
1533
1534                // Recently-referenced ring admission: only by-id touches admit
1535                // an id. Runs unconditionally (not gated on `dispatch_hook`,
1536                // which is opt-in) because the ring is a core
1537                // dispatch-boundary capability, not an observer.
1538                //
1539                // Keyed on `token.namespace()`, NOT `ns`: `ns` is the
1540                // gate-resolved namespace, which on the default
1541                // (non-explicit) dispatch path can be a non-local
1542                // `default_namespace` (e.g. "foreign") while the storage
1543                // token that actually created/touched the record is pinned
1544                // to `local`. The ring must be keyed on the namespace the
1545                // record actually lives in: the same namespace
1546                // `resolve_reference`'s ring lookup uses: or admission and
1547                // lookup silently diverge on any non-local `default_namespace`
1548                // config.
1549                if let Ok(ref ok_val) = result {
1550                    let admissions = crate::reference_ring::ring_admissions_for(verb, ok_val);
1551                    if !admissions.is_empty() {
1552                        let actor_key = format!("{}:{}", gate_req.actor.kind, gate_req.actor.id);
1553                        for (id, name) in admissions {
1554                            self.reference_ring.admit(
1555                                token.namespace().as_str(),
1556                                &actor_key,
1557                                id,
1558                                name,
1559                            );
1560                        }
1561                    }
1562                }
1563
1564                return result;
1565            }
1566        }
1567
1568        // No pack owns this verb: the gate allowed it, but no dispatch runs.
1569        // Persist the deferred audit row now (duration stays at the
1570        // `Event::new` default of 0 — no dispatch occurred to measure) so an
1571        // allowed-but-unknown verb is never silently dropped from the audit
1572        // trail (matches the "no audit row is ever dropped" contract above).
1573        if let Some(audit) = deferred_audit.take() {
1574            if let Some(store) = &self.event_store {
1575                // Dispatch is about to return `InvalidInput` below (no pack
1576                // owns this verb), so the persisted outcome must be `Error`,
1577                // not `Success`. `work_class` is still stamped (ADR-103
1578                // Decision (a)); `resource.cost_unit` is omitted, matching
1579                // every other errored-dispatch row.
1580                let storage_event = build_audit_storage_event(
1581                    &gate_req,
1582                    &audit,
1583                    EventOutcome::Error,
1584                    Some(crate::cost_unit::base_resource_payload(request_id)),
1585                );
1586                append_audit_event_best_effort(store, storage_event, verb).await;
1587            }
1588        }
1589
1590        // Verb-visibility handler names, precomputed at build() time (internal
1591        // subhandlers are excluded so they are not advertised in the
1592        // unknown-verb error).
1593        Err(RuntimeError::InvalidInput(format!(
1594            "unknown verb {verb:?}; available: {}",
1595            self.available_verbs.join(", ")
1596        )))
1597    }
1598
1599    /// Dispatch a verb under an out-of-band verified actor identity.
1600    ///
1601    /// `verified_actor` is a typed [`VerifiedActor`] (constructor rejects blank
1602    /// identifiers) — only code holding a `VerbRegistry` handle can supply it.
1603    /// `dispatch_as` never reads `params["actor"]` to derive the effective actor;
1604    /// individual verbs may still accept an `actor` field for their own documented
1605    /// business semantics, unrelated to the acting principal. Every pack handler
1606    /// that reads "who is calling" resolves it from the `NamespaceToken` the
1607    /// dispatch boundary mints, so `verified_actor` becomes exactly the principal
1608    /// those handlers observe.
1609    ///
1610    /// Equivalent to `dispatch_with_identity(verb, params, Some(identity))` with
1611    /// `identity.actor_id = Some(verified_actor)` and every other identity scalar
1612    /// (namespace, visible namespaces) left at this registry's construction-baked
1613    /// value. [`Self::dispatch`] and [`Self::dispatch_with_identity`] are unaffected.
1614    /// See `docs/api/pack.md#dispatch_as` for the embedding-host use case and the
1615    /// blank-identifier safety rationale.
1616    pub async fn dispatch_as(
1617        &self,
1618        verb: &str,
1619        params: Value,
1620        verified_actor: VerifiedActor,
1621    ) -> Result<Value, RuntimeError> {
1622        let identity = RequestIdentity {
1623            namespace: self.default_namespace.clone(),
1624            actor_id: Some(verified_actor.into_inner()),
1625            visible_namespaces: self
1626                .visible_namespaces
1627                .iter()
1628                .map(|ns| ns.as_str().to_string())
1629                .collect(),
1630            request_id: None,
1631        };
1632        self.dispatch_with_identity(verb, params, Some(identity))
1633            .await
1634    }
1635
1636    /// Registered pack-level by-ID resolvers, in registration order.
1637    ///
1638    /// Each element is `(pack_name, resolver)`. The kg `get` and `delete` handlers
1639    /// iterate this slice to probe pack-private tables when the standard KG
1640    /// substrates (entity/note/edge/event) return `None` for a given UUID.
1641    pub fn resolvers(&self) -> &[(String, Box<dyn PackByIdResolver>)] {
1642        &self.resolvers
1643    }
1644
1645    /// The daemon-warm recently-referenced ring (unified-verb draft ADR,
1646    /// Slice 1). Consumed by `resolve_reference` (Layer 0 stage 2) and by the
1647    /// `resolve` verb handler; admitted-to by every successful by-id
1648    /// dispatch (see the admission block in `dispatch_with_identity`).
1649    pub fn reference_ring(&self) -> &Arc<crate::reference_ring::ReferenceRing> {
1650        &self.reference_ring
1651    }
1652
1653    /// Find a kind hook among the registered packs.
1654    ///
1655    /// Walks packs in registration order; the first pack that both owns the
1656    /// kind (declares it in `note_kinds()` or `entity_kinds()`) and returns
1657    /// a hook from `kind_hook(kind)` wins. Returns `None` if the kind is
1658    /// unknown to all packs or no owning pack registered a hook.
1659    pub fn find_kind_hook(&self, kind: &str) -> Option<Arc<dyn KindHook>> {
1660        for pack in self.packs.iter() {
1661            let owns = pack.note_kinds().contains(&kind) || pack.entity_kinds().contains(&kind);
1662            if owns {
1663                if let Some(hook) = pack.kind_hook(kind) {
1664                    return Some(hook);
1665                }
1666            }
1667        }
1668        None
1669    }
1670
1671    /// All MCP-exposed handlers across all registered packs (`Visibility::Verb` only).
1672    ///
1673    /// Subhandlers (`Visibility::Subhandler`) are excluded — they are internal
1674    /// pipeline steps not surfaced on the MCP wire. Returned with `'static`
1675    /// lifetime since pack handlers are `&'static [HandlerDef]` constants.
1676    pub fn all_verbs(&self) -> Vec<&'static HandlerDef> {
1677        self.packs
1678            .iter()
1679            .flat_map(|p| p.handlers().iter())
1680            .filter(|h| matches!(h.visibility, Visibility::Verb))
1681            .collect()
1682    }
1683
1684    /// All MCP-exposed handlers paired with the name of the pack that owns them
1685    /// (`Visibility::Verb` only).
1686    ///
1687    /// Subhandlers (`Visibility::Subhandler`) are excluded from the MCP catalog
1688    /// Use `all_handlers_with_names` when internal handlers must
1689    /// also be enumerated (e.g. runtime introspection).
1690    pub fn all_verbs_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
1691        self.packs
1692            .iter()
1693            .flat_map(|p| p.handlers().iter().map(move |v| (p.name(), v)))
1694            .filter(|(_, h)| matches!(h.visibility, Visibility::Verb))
1695            .collect()
1696    }
1697
1698    /// All handler definitions across all registered packs, including subhandlers.
1699    ///
1700    /// Unlike `all_verbs`, this includes `Visibility::Subhandler` entries. Useful
1701    /// for runtime introspection (e.g. `list_handlers`) and tooling that needs
1702    /// the complete handler surface.
1703    pub fn all_handlers_with_names(&self) -> Vec<(&str, &'static HandlerDef)> {
1704        self.packs
1705            .iter()
1706            .flat_map(|p| p.handlers().iter().map(move |v| (p.name(), v)))
1707            .collect()
1708    }
1709
1710    /// Merged set of note kinds across all registered packs (deduplicated,
1711    /// first-seen order preserved).
1712    pub fn all_note_kinds(&self) -> Vec<&'static str> {
1713        let mut seen = std::collections::HashSet::new();
1714        self.packs
1715            .iter()
1716            .flat_map(|p| p.note_kinds().iter().copied())
1717            .filter(|k| seen.insert(*k))
1718            .collect()
1719    }
1720
1721    /// Merged set of entity kinds across all registered packs (deduplicated,
1722    /// first-seen order preserved).
1723    pub fn all_entity_kinds(&self) -> Vec<&'static str> {
1724        let mut seen = std::collections::HashSet::new();
1725        self.packs
1726            .iter()
1727            .flat_map(|p| p.entity_kinds().iter().copied())
1728            .filter(|k| seen.insert(*k))
1729            .collect()
1730    }
1731
1732    /// Names of packs in topological load order.
1733    pub fn pack_names(&self) -> Vec<&str> {
1734        self.packs.iter().map(|p| p.name()).collect()
1735    }
1736
1737    /// Declared dependencies for a registered pack.
1738    pub fn pack_requires(&self, name: &str) -> Option<&'static [&'static str]> {
1739        self.packs
1740            .iter()
1741            .find(|p| p.name() == name)
1742            .map(|p| p.requires())
1743    }
1744
1745    /// Note kinds owned by a specific registered pack.
1746    ///
1747    /// Returns `None` if no pack with `name` is registered. The slice is
1748    /// the pack's `NOTE_KINDS` constant — `'static` lifetime, no allocation.
1749    pub fn pack_note_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
1750        self.packs
1751            .iter()
1752            .find(|p| p.name() == name)
1753            .map(|p| p.note_kinds())
1754    }
1755
1756    /// Entity kinds owned by a specific registered pack.
1757    ///
1758    /// Returns `None` if no pack with `name` is registered. The slice is
1759    /// the pack's `ENTITY_KINDS` constant — `'static` lifetime, no allocation.
1760    pub fn pack_entity_kinds(&self, name: &str) -> Option<&'static [&'static str]> {
1761        self.packs
1762            .iter()
1763            .find(|p| p.name() == name)
1764            .map(|p| p.entity_kinds())
1765    }
1766
1767    /// Handlers declared by a specific registered pack.
1768    ///
1769    /// Returns `None` if no pack with `name` is registered. Each `HandlerDef`
1770    /// carries name + description + visibility — sufficient for introspection clients.
1771    pub fn pack_verbs(&self, name: &str) -> Option<&'static [HandlerDef]> {
1772        self.packs
1773            .iter()
1774            .find(|p| p.name() == name)
1775            .map(|p| p.handlers())
1776    }
1777
1778    /// All pack-declared edge endpoint rules across registered packs.
1779    ///
1780    /// Order follows topological pack registration; duplicates are *not* deduplicated —
1781    /// validation only checks membership, and an exact-duplicate rule is a
1782    /// harmless restatement.
1783    pub fn all_edge_rules(&self) -> Vec<EdgeEndpointRule> {
1784        self.packs
1785            .iter()
1786            .flat_map(|p| p.edge_rules().iter().copied())
1787            .collect()
1788    }
1789
1790    /// All pack-declared entity-type subtypes across registered packs.
1791    ///
1792    /// Order follows topological pack registration; duplicates are *not*
1793    /// deduplicated here — same posture as [`all_edge_rules`](Self::all_edge_rules).
1794    /// Consumers compose this with `EntityTypeRegistry::builtin()` via
1795    /// `EntityTypeRegistry::with_extra` to get the boot-time composed registry.
1796    pub fn all_entity_types(&self) -> Vec<EntityTypeDef> {
1797        self.packs
1798            .iter()
1799            .flat_map(|p| p.entity_types().iter().cloned())
1800            .collect()
1801    }
1802
1803    /// Collect all `NoteKindSpec` declarations from every loaded pack.
1804    ///
1805    /// Used by the runtime for lifecycle introspection and future enforcement.
1806    pub fn all_note_kind_specs(&self) -> Vec<&'static NoteKindSpec> {
1807        self.packs
1808            .iter()
1809            .flat_map(|p| p.note_kind_specs().iter())
1810            .collect()
1811    }
1812
1813    /// All pack-contributed validation rules across registered packs.
1814    ///
1815    /// Returns references into the pack-owned `'static` slices — no allocation
1816    /// beyond the outer `Vec`. Rule IDs are namespaced by pack; callers can
1817    /// group by `rule.id.split_once('/')` to attribute rules to their packs.
1818    pub fn all_validation_rules(&self) -> Vec<&'static ValidationRule> {
1819        self.packs
1820            .iter()
1821            .flat_map(|p| p.validation_rules().iter())
1822            .collect()
1823    }
1824
1825    /// Pack-auxiliary schema plans for all registered packs.
1826    ///
1827    /// Returns one `SchemaPlan` per pack. Callers (typically the runtime
1828    /// bootstrap) apply each plan to the pack's assigned backend. Empty plans
1829    /// are included so the caller can iterate uniformly; callers that want to
1830    /// skip empty plans should check `plan.is_empty()`.
1831    pub fn all_schema_plans(&self) -> Vec<SchemaPlan> {
1832        self.packs.iter().map(|p| p.schema_plan()).collect()
1833    }
1834
1835    /// Invoke `PackRuntime::register_embedders` on every registered pack.
1836    ///
1837    /// Called by the transport during startup, after the registry is built and
1838    /// before the first verb dispatch, so that custom embedding providers
1839    /// contributed by packs are reachable via `KhiveRuntime::embedder(name)`.
1840    ///
1841    /// Packs whose `register_embedders` is the default no-op pay no overhead.
1842    /// The method is idempotent when the underlying registry uses last-wins
1843    /// semantics for duplicate provider names.
1844    pub fn call_register_embedders(&self, runtime: &KhiveRuntime) {
1845        for pack in self.packs.iter() {
1846            pack.register_embedders(runtime);
1847        }
1848    }
1849
1850    /// Invoke `PackRuntime::register_entity_type_validator` on every registered pack.
1851    ///
1852    /// Called by the transport during startup, after the registry is built and
1853    /// before the first verb dispatch, so that entity-type validation at the
1854    /// runtime layer is active for all write paths including direct `create_many`
1855    /// callers that bypass the handler layer.
1856    ///
1857    /// Packs whose `register_entity_type_validator` is the default no-op pay
1858    /// no overhead.
1859    ///
1860    /// Composes [`all_entity_types`](Self::all_entity_types) once and passes
1861    /// the same aggregate to every pack, mirroring how `install_edge_rules`
1862    /// installs one `all_edge_rules()` aggregate for the whole registry.
1863    pub fn call_register_entity_type_validators(&self, runtime: &KhiveRuntime) {
1864        let entity_types = self.all_entity_types();
1865        for pack in self.packs.iter() {
1866            pack.register_entity_type_validator_with_types(runtime, &entity_types);
1867        }
1868    }
1869
1870    /// Invoke `PackRuntime::register_note_mutation_hook` on every registered pack.
1871    ///
1872    /// Called by the transport during startup, after the registry is built and
1873    /// before the first verb dispatch, so that note-mutation notifications at
1874    /// the runtime layer are active for all write paths — including KG's
1875    /// `update`/`delete` verbs reaching a `kind="memory"` note, which have no
1876    /// crate-level dependency on `khive-pack-memory`.
1877    ///
1878    /// Packs whose `register_note_mutation_hook` is the default no-op pay no
1879    /// overhead.
1880    pub fn call_register_note_mutation_hooks(&self, runtime: &KhiveRuntime) {
1881        for pack in self.packs.iter() {
1882            pack.register_note_mutation_hook(runtime);
1883        }
1884    }
1885
1886    /// Invoke `PackRuntime::warm` on every registered pack.
1887    /// Called by the daemon at boot (in a background task) so expensive in-memory
1888    /// state (ANN indexes) is pre-loaded without blocking request serving.
1889    pub async fn call_warm_all(&self) {
1890        for pack in self.packs.iter() {
1891            pack.warm().await;
1892        }
1893    }
1894
1895    /// Resolve the presentation policy for a verb name.
1896    ///
1897    /// Walks all registered handlers (including subhandlers) for the first
1898    /// matching name and returns its declared [`VerbPresentationPolicy`].
1899    /// Returns `Standard` for unknown verbs — unknown verbs will fail at
1900    /// dispatch anyway, so the fallback here is safe.
1901    pub fn presentation_policy_for(&self, verb: &str) -> khive_types::VerbPresentationPolicy {
1902        for pack in self.packs.iter() {
1903            if let Some(handler) = pack.handlers().iter().find(|h| h.name == verb) {
1904                return handler.presentation_policy();
1905            }
1906        }
1907        khive_types::VerbPresentationPolicy::Standard
1908    }
1909
1910    /// Returns `true` if the named verb exists and is tagged
1911    /// `Visibility::Subhandler` (internal / operator-only).
1912    ///
1913    /// Used by the MCP server to gate subhandler invocation at the wire
1914    /// boundary without blocking internal callers that invoke the same verbs
1915    /// through the runtime directly.
1916    pub fn is_subhandler_verb(&self, verb: &str) -> bool {
1917        for pack in self.packs.iter() {
1918            if let Some(handler) = pack.handlers().iter().find(|h| h.name == verb) {
1919                return matches!(handler.visibility, Visibility::Subhandler);
1920            }
1921        }
1922        false
1923    }
1924
1925    /// Apply all non-empty pack-auxiliary schema plans to the given backend.
1926    ///
1927    /// This is the centralized startup hook that replaced the previous lazy
1928    /// per-pack self-bootstrap pattern. Each pack's `SchemaPlan` carries
1929    /// idempotent `CREATE TABLE IF NOT EXISTS` DDL; calling this more than once
1930    /// is safe. Empty plans are skipped.
1931    ///
1932    /// Errors from individual plans are logged via `tracing::warn!` and not
1933    /// propagated so that a single pack's schema failure does not prevent the
1934    /// rest from loading. Callers that need hard-failure semantics should call
1935    /// `all_schema_plans()` and apply each plan individually.
1936    pub fn apply_schema_plans(&self, backend: &khive_db::StorageBackend) {
1937        for plan in self.all_schema_plans() {
1938            if plan.is_empty() {
1939                continue;
1940            }
1941            if let Err(e) = backend.apply_pack_ddl_statements(plan.statements) {
1942                tracing::warn!(
1943                    pack = plan.pack,
1944                    error = %e,
1945                    "failed to apply pack schema plan at startup (non-fatal)"
1946                );
1947            }
1948        }
1949    }
1950
1951    /// Pack-auxiliary schema plans with their owning pack names.
1952    ///
1953    /// Returns `(pack_name, SchemaPlan)` pairs for every registered pack.
1954    /// Used by the multi-backend boot path to apply each plan to the pack's
1955    /// assigned backend rather than a single shared backend.
1956    pub fn all_schema_plans_named(&self) -> Vec<(&'static str, SchemaPlan)> {
1957        self.packs
1958            .iter()
1959            .map(|p| {
1960                let plan = p.schema_plan();
1961                (plan.pack, plan)
1962            })
1963            .collect()
1964    }
1965
1966    /// Apply pack-auxiliary schema plans using a per-pack backend map.
1967    ///
1968    /// For each `(pack_name, plan)` returned by `all_schema_plans_named()`,
1969    /// applies the plan to `backend_for_pack[pack_name]` when present,
1970    /// falling back to `default_backend` for any pack not in the map.
1971    ///
1972    /// Returns an error when two packs on the same backend declare the same
1973    /// auxiliary table (ADR-028 §7 collision policy: boot failure naming both
1974    /// packs and the conflicting table).
1975    ///
1976    /// This is the multi-backend boot path (ADR-028). Single-backend callers
1977    /// should continue using [`Self::apply_schema_plans`].
1978    pub fn apply_schema_plans_with_map(
1979        &self,
1980        backend_for_pack: &HashMap<&str, &khive_db::StorageBackend>,
1981        default_backend: &khive_db::StorageBackend,
1982    ) -> Result<(), crate::PackSchemaCollisionError> {
1983        // Track which pack first claimed each table on each backend.
1984        // Backend identity is the raw pointer of the underlying connection pool Arc.
1985        let mut claimed: HashMap<(*const (), String), &'static str> = HashMap::new();
1986
1987        for (pack_name, plan) in self.all_schema_plans_named() {
1988            if plan.is_empty() {
1989                continue;
1990            }
1991            let backend = backend_for_pack
1992                .get(pack_name)
1993                .copied()
1994                .unwrap_or(default_backend);
1995            let backend_ptr = std::sync::Arc::as_ptr(&backend.pool_arc()) as *const ();
1996
1997            // Pre-scan DDL for table names and detect collisions before applying.
1998            for stmt in plan.statements {
1999                for table_name in extract_table_names(stmt) {
2000                    let key = (backend_ptr, table_name.clone());
2001                    match claimed.entry(key) {
2002                        std::collections::hash_map::Entry::Vacant(e) => {
2003                            e.insert(pack_name);
2004                        }
2005                        std::collections::hash_map::Entry::Occupied(e) => {
2006                            let prior_pack = *e.get();
2007                            return Err(crate::PackSchemaCollisionError {
2008                                pack_a: prior_pack,
2009                                pack_b: pack_name,
2010                                table: table_name,
2011                            });
2012                        }
2013                    }
2014                }
2015            }
2016
2017            backend
2018                .apply_pack_ddl_statements(plan.statements)
2019                .map_err(|e| crate::PackSchemaCollisionError {
2020                    pack_a: pack_name,
2021                    pack_b: pack_name,
2022                    table: format!("DDL error: {e}"),
2023                })?;
2024        }
2025        Ok(())
2026    }
2027}
2028
2029// ── Inventory-based dynamic pack loading ────────────────────────────────────
2030
2031/// Output of [`PackFactory::create_install`] — bundles the pack runtime with
2032/// its optional by-ID resolver and dispatch hook so a factory can hand back
2033/// all three built from one shared instance (a dispatch hook must observe
2034/// the same state the runtime mutates, not a second unrelated instance).
2035pub struct PackInstall {
2036    /// The pack runtime, registered into the builder's pack list.
2037    pub runtime: Box<dyn PackRuntime>,
2038    /// Optional by-ID resolver, registered when present.
2039    pub resolver: Option<Box<dyn PackByIdResolver>>,
2040    /// Optional post-dispatch observer, wired via `VerbRegistryBuilder::with_dispatch_hook`.
2041    pub dispatch_hook: Option<Arc<dyn DispatchHook>>,
2042}
2043
2044/// Factory for creating pack instances registered via `inventory` at link time.
2045/// Each pack crate submits a `&'static dyn PackFactory` wrapped in a
2046/// [`PackRegistration`]; the binary's linker collects them all into a single
2047/// slice iterable at runtime.
2048///
2049/// Implementors must be `Send + Sync + 'static` because the registry is built
2050/// once and shared across async tasks.
2051pub trait PackFactory: Send + Sync + 'static {
2052    /// Canonical lowercase name for this pack (e.g. `"kg"`, `"gtd"`).
2053    fn name(&self) -> &'static str;
2054
2055    /// Names of packs that must be loaded before this one.
2056    ///
2057    /// Defaults to empty so pack crates that have no dependencies compile
2058    /// without changes. [`PackRegistry::register_packs`] validates that every
2059    /// name listed here is present in the caller's explicit pack list — absent
2060    /// dependencies are a boot error, not silently auto-added.
2061    fn requires(&self) -> &'static [&'static str] {
2062        &[]
2063    }
2064
2065    /// Create a new pack instance for the given runtime.
2066    fn create(&self, runtime: KhiveRuntime) -> Box<dyn PackRuntime>;
2067
2068    /// Build the full installation bundle for this pack: runtime, optional
2069    /// resolver, optional dispatch hook.
2070    ///
2071    /// Defaults to composing `create` and `create_resolver` with no dispatch
2072    /// hook, so existing factories compile unchanged. Packs whose dispatch
2073    /// hook must observe the same instance as the runtime (e.g. `brain`)
2074    /// override this method instead of `create`, since the default would
2075    /// otherwise require two independent instances to share state.
2076    fn create_install(&self, runtime: KhiveRuntime) -> PackInstall {
2077        let resolver = self.create_resolver(runtime.clone());
2078        PackInstall {
2079            runtime: self.create(runtime),
2080            resolver,
2081            dispatch_hook: None,
2082        }
2083    }
2084
2085    /// Optionally create a `PackByIdResolver` for this pack.
2086    ///
2087    /// Packs that own private SQL tables implement this to hook into
2088    /// `get(id)` and `delete(id)`. Defaults to `None` so existing packs
2089    /// compile without changes.
2090    fn create_resolver(&self, _runtime: KhiveRuntime) -> Option<Box<dyn PackByIdResolver>> {
2091        None
2092    }
2093}
2094
2095/// Newtype wrapper collected by `inventory` so pack crates can submit
2096/// `&'static dyn PackFactory` references without the type-ascription syntax
2097/// that `inventory::submit!` does not support for bare trait-object references.
2098pub struct PackRegistration(pub &'static dyn PackFactory);
2099
2100inventory::collect!(PackRegistration);
2101
2102/// Error returned by [`PackRegistry::register_packs`] when boot validation fails.
2103#[derive(Debug)]
2104pub enum PackLoadError {
2105    /// The requested pack name was not found in the inventory.
2106    UnknownPack(String),
2107    /// A pack was requested but a declared dependency is absent from the list.
2108    MissingDependency {
2109        /// The pack that declared the dependency.
2110        pack: String,
2111        /// The dependency that is missing from the requested pack list.
2112        dep: String,
2113    },
2114}
2115
2116impl std::fmt::Display for PackLoadError {
2117    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2118        match self {
2119            PackLoadError::UnknownPack(name) => write!(f, "unknown pack {name:?}"),
2120            PackLoadError::MissingDependency { pack, dep } => write!(
2121                f,
2122                "pack {pack:?} requires {dep:?}, which is not in the requested pack list; \
2123                 add --pack {dep} before --pack {pack}"
2124            ),
2125        }
2126    }
2127}
2128
2129impl std::error::Error for PackLoadError {}
2130
2131/// Registry of pack factories discovered via `inventory` at link time.
2132///
2133/// No instance is needed — all methods are associated functions that walk the
2134/// globally-collected [`PackRegistration`] slice.
2135pub struct PackRegistry;
2136
2137impl PackRegistry {
2138    /// Names of all pack factories discovered via `inventory`.
2139    pub fn discovered_names() -> Vec<&'static str> {
2140        inventory::iter::<PackRegistration>
2141            .into_iter()
2142            .map(|r| r.0.name())
2143            .collect()
2144    }
2145
2146    /// Register the named packs into `builder` using the supplied `runtime`.
2147    ///
2148    /// Validates the explicit pack list against `PackFactory::requires()` —
2149    /// if any requested pack declares a dependency that is absent from `names`,
2150    /// registration fails (missing dependency is a boot error, not silently
2151    /// auto-added). Callers must include all required packs explicitly.
2152    ///
2153    /// The [`VerbRegistryBuilder::build`] topo-sort enforces correct load order.
2154    ///
2155    /// Returns `Ok(())` when all names are recognised and all declared
2156    /// dependencies are satisfied; returns `Err(PackLoadError)` with a
2157    /// distinct variant for unknown pack vs missing dependency.
2158    pub fn register_packs(
2159        names: &[String],
2160        runtime: KhiveRuntime,
2161        builder: &mut VerbRegistryBuilder,
2162    ) -> Result<(), PackLoadError> {
2163        // Build a name→factory index once.
2164        let all: Vec<&'static dyn PackFactory> = inventory::iter::<PackRegistration>
2165            .into_iter()
2166            .map(|r| r.0)
2167            .collect();
2168        let factory_for = |name: &str| -> Option<&'static dyn PackFactory> {
2169            all.iter().copied().find(|f| f.name() == name)
2170        };
2171
2172        // Validate that every requested name is a known factory.
2173        let requested: std::collections::HashSet<&str> = names.iter().map(String::as_str).collect();
2174        for name in names {
2175            factory_for(name.as_str()).ok_or_else(|| PackLoadError::UnknownPack(name.clone()))?;
2176        }
2177
2178        // Validate that all requires() dependencies are explicitly present in
2179        // the requested set. Missing dep → boot error, not auto-add.
2180        for name in names {
2181            let factory = factory_for(name.as_str()).unwrap(); // validated above
2182            for &dep in factory.requires() {
2183                if !requested.contains(dep) {
2184                    return Err(PackLoadError::MissingDependency {
2185                        pack: name.clone(),
2186                        dep: dep.to_string(),
2187                    });
2188                }
2189            }
2190        }
2191
2192        // Register every requested pack; VerbRegistryBuilder::build()
2193        // performs the topo-sort, so insertion order here does not matter.
2194        for name in names {
2195            let factory = factory_for(name.as_str()).unwrap(); // validated above
2196            let install = factory.create_install(runtime.clone());
2197            builder.register_boxed(install.runtime);
2198            if let Some(resolver) = install.resolver {
2199                builder.register_resolver(name.clone(), resolver);
2200            }
2201            if let Some(hook) = install.dispatch_hook {
2202                builder.with_dispatch_hook(hook);
2203            }
2204        }
2205
2206        Ok(())
2207    }
2208
2209    /// Register the named packs into `builder`, routing each pack to its own runtime.
2210    ///
2211    /// `runtimes` maps pack name → `KhiveRuntime` (one per backend assignment).
2212    /// `default_runtime` is used for any pack whose name is not in `runtimes`.
2213    /// The validation logic (unknown pack, missing dependency) is identical to
2214    /// [`PackRegistry::register_packs`].
2215    ///
2216    /// This is the multi-backend boot path (ADR-028). Single-backend callers
2217    /// should continue using [`PackRegistry::register_packs`].
2218    pub fn register_packs_with_runtimes(
2219        names: &[String],
2220        runtimes: &HashMap<String, KhiveRuntime>,
2221        default_runtime: &KhiveRuntime,
2222        builder: &mut VerbRegistryBuilder,
2223    ) -> Result<(), PackLoadError> {
2224        let all: Vec<&'static dyn PackFactory> = inventory::iter::<PackRegistration>
2225            .into_iter()
2226            .map(|r| r.0)
2227            .collect();
2228        let factory_for = |name: &str| -> Option<&'static dyn PackFactory> {
2229            all.iter().copied().find(|f| f.name() == name)
2230        };
2231
2232        let requested: std::collections::HashSet<&str> = names.iter().map(String::as_str).collect();
2233        for name in names {
2234            factory_for(name.as_str()).ok_or_else(|| PackLoadError::UnknownPack(name.clone()))?;
2235        }
2236
2237        for name in names {
2238            let factory = factory_for(name.as_str()).unwrap();
2239            for &dep in factory.requires() {
2240                if !requested.contains(dep) {
2241                    return Err(PackLoadError::MissingDependency {
2242                        pack: name.clone(),
2243                        dep: dep.to_string(),
2244                    });
2245                }
2246            }
2247        }
2248
2249        for name in names {
2250            let factory = factory_for(name.as_str()).unwrap();
2251            let runtime = runtimes
2252                .get(name.as_str())
2253                .cloned()
2254                .unwrap_or_else(|| default_runtime.clone());
2255            let install = factory.create_install(runtime);
2256            builder.register_boxed(install.runtime);
2257            if let Some(resolver) = install.resolver {
2258                builder.register_resolver(name.clone(), resolver);
2259            }
2260            if let Some(hook) = install.dispatch_hook {
2261                builder.with_dispatch_hook(hook);
2262            }
2263        }
2264
2265        Ok(())
2266    }
2267}
2268
2269fn target_id_from_args(args: &serde_json::Value) -> Option<uuid::Uuid> {
2270    args.get("target_id")
2271        .and_then(serde_json::Value::as_str)
2272        .and_then(|s| s.parse::<uuid::Uuid>().ok())
2273}
2274
2275/// Build a v1-shape audit storage event from a gate check outcome.
2276/// See `docs/api/pack.md#build_audit_storage_event` for the `resource` payload contract.
2277fn build_audit_storage_event(
2278    gate_req: &GateRequest,
2279    audit: &AuditEvent,
2280    outcome: EventOutcome,
2281    resource: Option<Value>,
2282) -> Event {
2283    let mut audit_data = serde_json::to_value(audit).unwrap_or_else(|e| {
2284        tracing::warn!(error = %e, "failed to serialize AuditEvent for EventStore");
2285        serde_json::Value::Null
2286    });
2287    if let Some(resource) = resource {
2288        if let Value::Object(ref mut map) = audit_data {
2289            map.insert("resource".to_string(), resource);
2290        }
2291    }
2292    let mut storage_event = Event::new(
2293        gate_req.namespace.as_str(),
2294        gate_req.verb.as_str(),
2295        EventKind::Audit,
2296        SubstrateKind::Event,
2297        format!("{}:{}", gate_req.actor.kind, gate_req.actor.id),
2298    )
2299    .with_outcome(outcome)
2300    .with_payload(audit_data);
2301    if let Some(target_id) = target_id_from_args(&gate_req.args) {
2302        storage_event = storage_event.with_target(target_id);
2303    }
2304    storage_event
2305}
2306
2307/// Append an audit event, logging and swallowing store failures.
2308///
2309/// Audit persistence is best-effort everywhere in the dispatch path: a store
2310/// write failure must never fail the verb call it is auditing.
2311async fn append_audit_event_best_effort(store: &Arc<dyn EventStore>, event: Event, verb: &str) {
2312    if let Err(store_err) = store.append_event(event).await {
2313        tracing::warn!(
2314            verb,
2315            error = %store_err,
2316            "audit event store write failed (non-fatal)"
2317        );
2318    }
2319}
2320
2321/// Schema v2 audit payload for a successful singleton `link` call — additive
2322/// over v1 via `#[serde(flatten)]`. See `docs/api/pack.md#linkauditsuccessv2`.
2323#[derive(Debug, Clone, serde::Serialize)]
2324struct LinkAuditSuccessV2 {
2325    #[serde(flatten)]
2326    audit: AuditEvent,
2327    edge_id: uuid::Uuid,
2328    source_id: uuid::Uuid,
2329    target_id: uuid::Uuid,
2330    relation: String,
2331    weight: f64,
2332}
2333
2334/// Extract edge fields to enrich a successful singleton `link` audit row.
2335/// Returns `None` on any missing/malformed field (falls back to v1 shape).
2336/// See `docs/api/pack.md#link_audit_success_from_result`.
2337fn link_audit_success_from_result(
2338    audit: AuditEvent,
2339    result: &serde_json::Value,
2340) -> Option<(uuid::Uuid, serde_json::Value)> {
2341    let edge_id = result.get("id")?.as_str()?.parse::<uuid::Uuid>().ok()?;
2342    let source_id = result
2343        .get("source_id")?
2344        .as_str()?
2345        .parse::<uuid::Uuid>()
2346        .ok()?;
2347    let target_id = result
2348        .get("target_id")?
2349        .as_str()?
2350        .parse::<uuid::Uuid>()
2351        .ok()?;
2352    let relation = result.get("relation")?.as_str()?.to_string();
2353    let weight = result.get("weight")?.as_f64()?;
2354    let enriched = LinkAuditSuccessV2 {
2355        audit,
2356        edge_id,
2357        source_id,
2358        target_id,
2359        relation,
2360        weight,
2361    };
2362    let payload = serde_json::to_value(&enriched).ok()?;
2363    Some((edge_id, payload))
2364}
2365
2366/// Resolve and validate a caller-supplied `namespace` argument the same way
2367/// on every MCP ingress path.
2368///
2369/// - Absent `namespace` key → parse `default_namespace`.
2370/// - Present `namespace: "<string>"` → parse the caller's value.
2371/// - Present non-string `namespace` (null, number, bool, array, object) →
2372///   fail closed with `RuntimeError::InvalidInput`. ADR-018 requires this:
2373///   a malformed explicit value must never be silently coerced to the
2374///   default namespace.
2375///
2376/// Single chokepoint for both `VerbRegistry::dispatch` and the multi-backend
2377/// coordinator intercept — see `docs/api/pack.md#resolve_explicit_namespace`.
2378pub fn resolve_explicit_namespace(
2379    params: &Value,
2380    default_namespace: &str,
2381) -> Result<Namespace, RuntimeError> {
2382    match params.get("namespace") {
2383        None => Namespace::parse(default_namespace)
2384            .map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace: {e}"))),
2385        Some(Value::String(ns_str)) => Namespace::parse(ns_str)
2386            .map_err(|e| RuntimeError::InvalidInput(format!("invalid namespace {ns_str:?}: {e}"))),
2387        Some(other) => Err(RuntimeError::InvalidInput(format!(
2388            "invalid namespace: expected string when present, got {}",
2389            json_type_name(other),
2390        ))),
2391    }
2392}
2393
2394/// JSON type name for error messages: describes a present-but-malformed
2395/// `namespace` value without echoing its contents.
2396pub fn json_type_name(v: &Value) -> &'static str {
2397    match v {
2398        Value::Null => "null",
2399        Value::Bool(_) => "boolean",
2400        Value::Number(_) => "number",
2401        Value::String(_) => "string",
2402        Value::Array(_) => "array",
2403        Value::Object(_) => "object",
2404    }
2405}
2406
2407// INLINE TEST JUSTIFICATION: tests here exercise VerbRegistry collision detection,
2408// gate enforcement, and dispatch ordering that depend on direct access to the
2409// registry's private `packs` Vec and gate field. Moving them to tests/ would
2410// require pub-exporting registry internals. Broad behavioral dispatch tests
2411// live in tests/integration.rs.
2412#[cfg(test)]
2413mod tests {
2414    use super::*;
2415    use crate::ActorRef;
2416    use khive_types::Pack;
2417
2418    struct AlphaPack;
2419
2420    impl Pack for AlphaPack {
2421        const NAME: &'static str = "alpha";
2422        const NOTE_KINDS: &'static [&'static str] = &["memo", "log"];
2423        const ENTITY_KINDS: &'static [&'static str] = &["widget"];
2424        const HANDLERS: &'static [HandlerDef] = &[
2425            HandlerDef {
2426                name: "create",
2427                description: "create a widget",
2428                visibility: Visibility::Verb,
2429                category: VerbCategory::Commissive,
2430                params: &[],
2431            },
2432            HandlerDef {
2433                name: "list",
2434                description: "list widgets",
2435                visibility: Visibility::Verb,
2436                category: VerbCategory::Assertive,
2437                params: &[],
2438            },
2439        ];
2440    }
2441
2442    #[async_trait]
2443    impl PackRuntime for AlphaPack {
2444        fn name(&self) -> &str {
2445            AlphaPack::NAME
2446        }
2447        fn note_kinds(&self) -> &'static [&'static str] {
2448            AlphaPack::NOTE_KINDS
2449        }
2450        fn entity_kinds(&self) -> &'static [&'static str] {
2451            AlphaPack::ENTITY_KINDS
2452        }
2453        fn handlers(&self) -> &'static [HandlerDef] {
2454            AlphaPack::HANDLERS
2455        }
2456        async fn dispatch(
2457            &self,
2458            verb: &str,
2459            _params: Value,
2460            _registry: &VerbRegistry,
2461            _token: &NamespaceToken,
2462        ) -> Result<Value, RuntimeError> {
2463            Ok(serde_json::json!({ "pack": "alpha", "verb": verb }))
2464        }
2465    }
2466
2467    /// A pack whose `dispatch` sleeps for a fixed, generous duration so
2468    /// `duration_us` regression tests (ADR-103 Stage 1) have a reliably
2469    /// nonzero, non-flaky measured dispatch time to assert against.
2470    struct SleepingPack;
2471
2472    impl Pack for SleepingPack {
2473        const NAME: &'static str = "sleeping";
2474        const NOTE_KINDS: &'static [&'static str] = &[];
2475        const ENTITY_KINDS: &'static [&'static str] = &[];
2476        const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
2477            name: "slow_op",
2478            description: "sleeps before returning",
2479            visibility: Visibility::Verb,
2480            category: VerbCategory::Assertive,
2481            params: &[],
2482        }];
2483    }
2484
2485    #[async_trait]
2486    impl PackRuntime for SleepingPack {
2487        fn name(&self) -> &str {
2488            SleepingPack::NAME
2489        }
2490        fn note_kinds(&self) -> &'static [&'static str] {
2491            SleepingPack::NOTE_KINDS
2492        }
2493        fn entity_kinds(&self) -> &'static [&'static str] {
2494            SleepingPack::ENTITY_KINDS
2495        }
2496        fn handlers(&self) -> &'static [HandlerDef] {
2497            SleepingPack::HANDLERS
2498        }
2499        async fn dispatch(
2500            &self,
2501            verb: &str,
2502            _params: Value,
2503            _registry: &VerbRegistry,
2504            _token: &NamespaceToken,
2505        ) -> Result<Value, RuntimeError> {
2506            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
2507            Ok(serde_json::json!({ "pack": "sleeping", "verb": verb }))
2508        }
2509    }
2510
2511    struct BetaPack;
2512
2513    impl Pack for BetaPack {
2514        const NAME: &'static str = "beta";
2515        const NOTE_KINDS: &'static [&'static str] = &["alert"];
2516        const ENTITY_KINDS: &'static [&'static str] = &["widget", "gadget"];
2517        const HANDLERS: &'static [HandlerDef] = &[
2518            HandlerDef {
2519                name: "notify",
2520                description: "send alert",
2521                visibility: Visibility::Verb,
2522                category: VerbCategory::Commissive,
2523                params: &[],
2524            },
2525            // "create" is Subhandler so it does NOT collide with AlphaPack's
2526            // Verb-visibility "create" — subhandlers are pack-internal and
2527            // excluded from cross-pack collision detection.
2528            HandlerDef {
2529                name: "create",
2530                description: "beta internal create (subhandler)",
2531                visibility: Visibility::Subhandler,
2532                category: VerbCategory::Commissive,
2533                params: &[],
2534            },
2535        ];
2536    }
2537
2538    /// Build a registry with AlphaPack + BetaPack.
2539    ///
2540    /// BetaPack's `create` is Subhandler so there is no Verb-visibility
2541    /// collision with AlphaPack's `create` Verb. Tests that need a collision
2542    /// use `build_colliding_registry()` instead.
2543    fn build_registry() -> VerbRegistry {
2544        let mut builder = VerbRegistryBuilder::new();
2545        builder.register(AlphaPack);
2546        builder.register(BetaPack);
2547        builder.build().expect("registry builds without collision")
2548    }
2549
2550    /// Build a registry with two packs that declare the same Verb-visibility
2551    /// handler — used to test that `VerbCollision` is raised at build time.
2552    struct CollidingPack;
2553
2554    impl Pack for CollidingPack {
2555        const NAME: &'static str = "colliding";
2556        const NOTE_KINDS: &'static [&'static str] = &[];
2557        const ENTITY_KINDS: &'static [&'static str] = &[];
2558        const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
2559            name: "create",
2560            description: "duplicate Verb-visibility create",
2561            visibility: Visibility::Verb,
2562            category: VerbCategory::Commissive,
2563            params: &[],
2564        }];
2565    }
2566
2567    #[async_trait]
2568    impl PackRuntime for CollidingPack {
2569        fn name(&self) -> &str {
2570            Self::NAME
2571        }
2572        fn note_kinds(&self) -> &'static [&'static str] {
2573            Self::NOTE_KINDS
2574        }
2575        fn entity_kinds(&self) -> &'static [&'static str] {
2576            Self::ENTITY_KINDS
2577        }
2578        fn handlers(&self) -> &'static [HandlerDef] {
2579            Self::HANDLERS
2580        }
2581        async fn dispatch(
2582            &self,
2583            verb: &str,
2584            _params: Value,
2585            _registry: &VerbRegistry,
2586            _token: &NamespaceToken,
2587        ) -> Result<Value, RuntimeError> {
2588            Ok(serde_json::json!({ "pack": "colliding", "verb": verb }))
2589        }
2590    }
2591
2592    #[async_trait]
2593    impl PackRuntime for BetaPack {
2594        fn name(&self) -> &str {
2595            BetaPack::NAME
2596        }
2597        fn note_kinds(&self) -> &'static [&'static str] {
2598            BetaPack::NOTE_KINDS
2599        }
2600        fn entity_kinds(&self) -> &'static [&'static str] {
2601            BetaPack::ENTITY_KINDS
2602        }
2603        fn handlers(&self) -> &'static [HandlerDef] {
2604            BetaPack::HANDLERS
2605        }
2606        async fn dispatch(
2607            &self,
2608            verb: &str,
2609            _params: Value,
2610            _registry: &VerbRegistry,
2611            _token: &NamespaceToken,
2612        ) -> Result<Value, RuntimeError> {
2613            Ok(serde_json::json!({ "pack": "beta", "verb": verb }))
2614        }
2615    }
2616
2617    #[tokio::test]
2618    async fn dispatch_routes_to_correct_pack() {
2619        let reg = build_registry();
2620
2621        let res = reg.dispatch("list", Value::Null).await.unwrap();
2622        assert_eq!(res["pack"], "alpha");
2623
2624        let res = reg.dispatch("notify", Value::Null).await.unwrap();
2625        assert_eq!(res["pack"], "beta");
2626    }
2627
2628    /// Two packs declaring the same `Visibility::Verb` handler must be
2629    /// rejected at build time — the old "first registered wins" behaviour is
2630    /// replaced by a boot error.
2631    #[test]
2632    fn verb_collision_is_boot_time_error() {
2633        let mut builder = VerbRegistryBuilder::new();
2634        builder.register(AlphaPack);
2635        builder.register(CollidingPack);
2636        let err = builder
2637            .build()
2638            .err()
2639            .expect("duplicate Verb-visibility handler must be rejected at build time");
2640        assert!(
2641            matches!(err, RuntimeError::VerbCollision { ref verb, .. } if verb == "create"),
2642            "expected VerbCollision for 'create', got {err:?}"
2643        );
2644        let msg = err.to_string();
2645        assert!(
2646            msg.contains("create"),
2647            "error must name the colliding verb: {msg}"
2648        );
2649        assert!(
2650            msg.contains("alpha") || msg.contains("colliding"),
2651            "error must name one of the conflicting packs: {msg}"
2652        );
2653    }
2654
2655    /// Subhandler-visibility handlers with the same name across packs are NOT
2656    /// a collision — they are pack-internal and excluded from cross-pack
2657    /// collision detection.
2658    #[test]
2659    fn subhandler_same_name_across_packs_is_not_a_collision() {
2660        struct SubhandlerPack;
2661        impl Pack for SubhandlerPack {
2662            const NAME: &'static str = "subhandler_pack";
2663            const NOTE_KINDS: &'static [&'static str] = &[];
2664            const ENTITY_KINDS: &'static [&'static str] = &[];
2665            const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
2666                name: "create",
2667                description: "internal create",
2668                visibility: Visibility::Subhandler,
2669                category: VerbCategory::Commissive,
2670                params: &[],
2671            }];
2672        }
2673        #[async_trait]
2674        impl PackRuntime for SubhandlerPack {
2675            fn name(&self) -> &str {
2676                Self::NAME
2677            }
2678            fn note_kinds(&self) -> &'static [&'static str] {
2679                Self::NOTE_KINDS
2680            }
2681            fn entity_kinds(&self) -> &'static [&'static str] {
2682                Self::ENTITY_KINDS
2683            }
2684            fn handlers(&self) -> &'static [HandlerDef] {
2685                Self::HANDLERS
2686            }
2687            async fn dispatch(
2688                &self,
2689                verb: &str,
2690                _: Value,
2691                _: &VerbRegistry,
2692                _: &NamespaceToken,
2693            ) -> Result<Value, RuntimeError> {
2694                Ok(serde_json::json!({"pack": "subhandler_pack", "verb": verb}))
2695            }
2696        }
2697        let mut builder = VerbRegistryBuilder::new();
2698        builder.register(AlphaPack); // AlphaPack has Verb "create"
2699        builder.register(SubhandlerPack); // SubhandlerPack has Subhandler "create" — no collision
2700        builder
2701            .build()
2702            .expect("subhandler same name must NOT be a collision");
2703    }
2704
2705    #[tokio::test]
2706    async fn dispatch_unknown_verb_returns_error() {
2707        let reg = build_registry();
2708
2709        let err = reg.dispatch("explode", Value::Null).await.unwrap_err();
2710        let msg = err.to_string();
2711        assert!(msg.contains("explode"));
2712        assert!(msg.contains("create"));
2713    }
2714
2715    /// `all_verbs` returns only `Visibility::Verb` entries.
2716    ///
2717    /// BetaPack's `create` is `Visibility::Subhandler` — it must NOT appear
2718    /// in `all_verbs()` even though it has the same name as a Verb in AlphaPack.
2719    #[test]
2720    fn all_verbs_aggregates_across_packs_excludes_subhandlers() {
2721        let reg = build_registry();
2722        let verbs: Vec<&str> = reg.all_verbs().iter().map(|v| v.name).collect();
2723        // BetaPack's "create" (Subhandler) is absent; only Verb-visibility entries appear.
2724        assert_eq!(verbs, vec!["create", "list", "notify"]);
2725    }
2726
2727    #[test]
2728    fn all_verbs_with_names_pairs_pack_name_excludes_subhandlers() {
2729        let reg = build_registry();
2730        let pairs: Vec<(&str, &str)> = reg
2731            .all_verbs_with_names()
2732            .iter()
2733            .map(|(pack, v)| (*pack, v.name))
2734            .collect();
2735        // BetaPack's "create" is Subhandler and must NOT appear here.
2736        assert_eq!(
2737            pairs,
2738            vec![("alpha", "create"), ("alpha", "list"), ("beta", "notify"),]
2739        );
2740    }
2741
2742    #[test]
2743    fn all_handlers_with_names_includes_subhandlers() {
2744        let reg = build_registry();
2745        let pairs: Vec<(&str, &str)> = reg
2746            .all_handlers_with_names()
2747            .iter()
2748            .map(|(pack, v)| (*pack, v.name))
2749            .collect();
2750        // BetaPack's Subhandler "create" IS present in the full handler list.
2751        assert_eq!(
2752            pairs,
2753            vec![
2754                ("alpha", "create"),
2755                ("alpha", "list"),
2756                ("beta", "notify"),
2757                ("beta", "create"),
2758            ]
2759        );
2760    }
2761
2762    #[test]
2763    fn note_kinds_are_ordered() {
2764        let reg = build_registry();
2765        let kinds = reg.all_note_kinds();
2766        assert_eq!(kinds, vec!["memo", "log", "alert"]);
2767    }
2768
2769    #[test]
2770    fn note_kind_duplicate_rejected_at_build_time() {
2771        struct DupPack;
2772
2773        impl khive_types::Pack for DupPack {
2774            const NAME: &'static str = "dup";
2775            // "memo" is already declared by AlphaPack — must be rejected at build.
2776            const NOTE_KINDS: &'static [&'static str] = &["memo"];
2777            const ENTITY_KINDS: &'static [&'static str] = &[];
2778            const HANDLERS: &'static [HandlerDef] = &[];
2779        }
2780
2781        #[async_trait]
2782        impl PackRuntime for DupPack {
2783            fn name(&self) -> &str {
2784                Self::NAME
2785            }
2786            fn note_kinds(&self) -> &'static [&'static str] {
2787                Self::NOTE_KINDS
2788            }
2789            fn entity_kinds(&self) -> &'static [&'static str] {
2790                Self::ENTITY_KINDS
2791            }
2792            fn handlers(&self) -> &'static [HandlerDef] {
2793                Self::HANDLERS
2794            }
2795            async fn dispatch(
2796                &self,
2797                _verb: &str,
2798                _params: Value,
2799                _registry: &VerbRegistry,
2800                _token: &NamespaceToken,
2801            ) -> Result<Value, RuntimeError> {
2802                Ok(Value::Null)
2803            }
2804        }
2805
2806        let mut builder = VerbRegistryBuilder::new();
2807        builder.register(AlphaPack);
2808        builder.register(DupPack);
2809        let err = builder
2810            .build()
2811            .err()
2812            .expect("duplicate note kind must be rejected");
2813        let msg = err.to_string();
2814        assert!(
2815            msg.contains("memo"),
2816            "error must name the duplicate kind: {msg}"
2817        );
2818        assert!(
2819            msg.contains("alpha") || msg.contains("dup"),
2820            "error must name one of the conflicting packs: {msg}"
2821        );
2822    }
2823
2824    #[test]
2825    fn entity_kinds_are_deduplicated() {
2826        let reg = build_registry();
2827        let kinds = reg.all_entity_kinds();
2828        assert_eq!(kinds, vec!["widget", "gadget"]);
2829    }
2830
2831    // ---- ENTITY_TYPES composition (pack-declared entity-type subtypes) ----
2832
2833    struct GammaPack;
2834
2835    impl Pack for GammaPack {
2836        const NAME: &'static str = "gamma";
2837        const NOTE_KINDS: &'static [&'static str] = &[];
2838        const ENTITY_KINDS: &'static [&'static str] = &[];
2839        const HANDLERS: &'static [HandlerDef] = &[];
2840        const ENTITY_TYPES: &'static [EntityTypeDef] = &[EntityTypeDef {
2841            kind: khive_types::EntityKind::Document,
2842            type_name: "gamma_report",
2843            aliases: &["gamma_rep"],
2844        }];
2845    }
2846
2847    #[async_trait]
2848    impl PackRuntime for GammaPack {
2849        fn name(&self) -> &str {
2850            Self::NAME
2851        }
2852        fn note_kinds(&self) -> &'static [&'static str] {
2853            Self::NOTE_KINDS
2854        }
2855        fn entity_kinds(&self) -> &'static [&'static str] {
2856            Self::ENTITY_KINDS
2857        }
2858        fn handlers(&self) -> &'static [HandlerDef] {
2859            Self::HANDLERS
2860        }
2861        fn entity_types(&self) -> &'static [EntityTypeDef] {
2862            Self::ENTITY_TYPES
2863        }
2864        async fn dispatch(
2865            &self,
2866            verb: &str,
2867            _params: Value,
2868            _registry: &VerbRegistry,
2869            _token: &NamespaceToken,
2870        ) -> Result<Value, RuntimeError> {
2871            Ok(serde_json::json!({ "pack": "gamma", "verb": verb }))
2872        }
2873    }
2874
2875    /// Builtin-only behavior is unchanged when no pack declares extras:
2876    /// `all_entity_types()` is empty, and composing it with the builtin
2877    /// registry resolves exactly like `EntityTypeRegistry::builtin()`.
2878    #[test]
2879    fn all_entity_types_empty_when_no_pack_declares_extras() {
2880        let reg = build_registry(); // AlphaPack + BetaPack — neither declares ENTITY_TYPES.
2881        assert!(reg.all_entity_types().is_empty());
2882        let composed = khive_types::EntityTypeRegistry::with_extra(reg.all_entity_types());
2883        let resolved = composed
2884            .resolve(khive_types::EntityKind::Document, Some("paper"))
2885            .expect("builtin paper subtype must still resolve");
2886        assert_eq!(resolved.entity_type.as_deref(), Some("paper"));
2887    }
2888
2889    /// A pack-declared entity type validates through the composed registry,
2890    /// and builtin subtypes remain resolvable alongside it.
2891    #[test]
2892    fn pack_declared_entity_type_validates_through_composed_registry() {
2893        let mut builder = VerbRegistryBuilder::new();
2894        builder.register(AlphaPack);
2895        builder.register(GammaPack);
2896        let reg = builder.build().expect("registry builds");
2897
2898        let extras = reg.all_entity_types();
2899        assert_eq!(extras.len(), 1);
2900
2901        let composed = khive_types::EntityTypeRegistry::with_extra(extras);
2902        let resolved = composed
2903            .resolve(khive_types::EntityKind::Document, Some("gamma_rep"))
2904            .expect("pack-declared alias must resolve through the composed registry");
2905        assert_eq!(resolved.entity_type.as_deref(), Some("gamma_report"));
2906
2907        let builtin_resolved = composed
2908            .resolve(khive_types::EntityKind::Document, Some("paper"))
2909            .expect("builtin subtype must remain resolvable when a pack adds extras");
2910        assert_eq!(builtin_resolved.entity_type.as_deref(), Some("paper"));
2911
2912        composed
2913            .resolve(khive_types::EntityKind::Document, Some("nonexistent_type"))
2914            .expect_err("undeclared entity_type must still be rejected");
2915    }
2916
2917    /// Two packs declaring the exact same `(kind, type_name)` subtype are
2918    /// rejected at `build()` — ADR-001's registry-ownership collision rule
2919    /// ("same `(base_kind, canonical_name)` from two different packs = boot
2920    /// error") — instead of silently resolving via registration order the
2921    /// way `EntityTypeRegistry::with_extra`'s hard-`insert` semantics would.
2922    #[test]
2923    fn overlapping_pack_declared_entity_types_reject_at_boot() {
2924        struct DeltaPack;
2925        impl Pack for DeltaPack {
2926            const NAME: &'static str = "delta";
2927            const NOTE_KINDS: &'static [&'static str] = &[];
2928            const ENTITY_KINDS: &'static [&'static str] = &[];
2929            const HANDLERS: &'static [HandlerDef] = &[];
2930            const ENTITY_TYPES: &'static [EntityTypeDef] = &[EntityTypeDef {
2931                kind: khive_types::EntityKind::Document,
2932                type_name: "gamma_report",
2933                aliases: &["gamma_rep"],
2934            }];
2935        }
2936        #[async_trait]
2937        impl PackRuntime for DeltaPack {
2938            fn name(&self) -> &str {
2939                Self::NAME
2940            }
2941            fn note_kinds(&self) -> &'static [&'static str] {
2942                Self::NOTE_KINDS
2943            }
2944            fn entity_kinds(&self) -> &'static [&'static str] {
2945                Self::ENTITY_KINDS
2946            }
2947            fn handlers(&self) -> &'static [HandlerDef] {
2948                Self::HANDLERS
2949            }
2950            fn entity_types(&self) -> &'static [EntityTypeDef] {
2951                Self::ENTITY_TYPES
2952            }
2953            async fn dispatch(
2954                &self,
2955                verb: &str,
2956                _params: Value,
2957                _registry: &VerbRegistry,
2958                _token: &NamespaceToken,
2959            ) -> Result<Value, RuntimeError> {
2960                Ok(serde_json::json!({ "pack": "delta", "verb": verb }))
2961            }
2962        }
2963
2964        let mut builder = VerbRegistryBuilder::new();
2965        builder.register(GammaPack);
2966        builder.register(DeltaPack);
2967        let err = builder.build().err().expect(
2968            "overlapping ENTITY_TYPES declarations must fail at build, not silently compose",
2969        );
2970
2971        let msg = err.to_string();
2972        assert!(
2973            msg.contains("gamma") && msg.contains("delta"),
2974            "collision error must name both contributing packs: {msg}"
2975        );
2976        assert!(
2977            msg.contains("gamma_report"),
2978            "collision error must name the colliding entity_type key: {msg}"
2979        );
2980    }
2981
2982    // ---- Gate wiring ----
2983
2984    use khive_gate::{Gate, GateError};
2985    use std::sync::atomic::{AtomicUsize, Ordering};
2986    use std::sync::Arc;
2987
2988    #[derive(Default, Debug)]
2989    struct CountingGate {
2990        calls: AtomicUsize,
2991        deny_verb: Option<&'static str>,
2992    }
2993
2994    impl Gate for CountingGate {
2995        fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
2996            self.calls.fetch_add(1, Ordering::SeqCst);
2997            if Some(req.verb.as_str()) == self.deny_verb {
2998                Ok(GateDecision::deny(format!("test deny for {}", req.verb)))
2999            } else {
3000                Ok(GateDecision::allow())
3001            }
3002        }
3003    }
3004
3005    #[tokio::test]
3006    async fn dispatch_consults_the_gate() {
3007        let gate = Arc::new(CountingGate::default());
3008        let mut builder = VerbRegistryBuilder::new();
3009        builder.register(AlphaPack);
3010        builder.with_gate(gate.clone());
3011        let reg = builder.build().expect("registry builds");
3012
3013        reg.dispatch("list", Value::Null).await.unwrap();
3014        reg.dispatch("create", Value::Null).await.unwrap();
3015        assert_eq!(
3016            gate.calls.load(Ordering::SeqCst),
3017            2,
3018            "gate should be consulted once per dispatch"
3019        );
3020    }
3021
3022    #[tokio::test]
3023    async fn dispatch_returns_permission_denied_on_deny_v03() {
3024        let gate = Arc::new(CountingGate {
3025            calls: AtomicUsize::new(0),
3026            deny_verb: Some("create"),
3027        });
3028        let mut builder = VerbRegistryBuilder::new();
3029        builder.register(AlphaPack);
3030        builder.with_gate(gate.clone());
3031        let reg = builder.build().expect("registry builds");
3032
3033        // Gate denies — dispatch now returns PermissionDenied (hard enforcement).
3034        let err = reg.dispatch("create", Value::Null).await.unwrap_err();
3035        assert!(
3036            matches!(err, RuntimeError::PermissionDenied { ref verb, .. } if verb == "create"),
3037            "expected PermissionDenied, got {err:?}"
3038        );
3039        let msg = err.to_string();
3040        assert!(
3041            msg.contains("create"),
3042            "error message must name the verb: {msg}"
3043        );
3044        assert!(
3045            msg.contains("test deny for create"),
3046            "error message must carry the deny reason: {msg}"
3047        );
3048        assert_eq!(gate.calls.load(Ordering::SeqCst), 1);
3049    }
3050
3051    #[tokio::test]
3052    async fn dispatch_allow_verb_succeeds_even_with_deny_gate_for_other_verb() {
3053        // Deny only "create" — "list" must still work.
3054        let gate = Arc::new(CountingGate {
3055            calls: AtomicUsize::new(0),
3056            deny_verb: Some("create"),
3057        });
3058        let mut builder = VerbRegistryBuilder::new();
3059        builder.register(AlphaPack);
3060        builder.with_gate(gate.clone());
3061        let reg = builder.build().expect("registry builds");
3062
3063        let res = reg.dispatch("list", Value::Null).await.unwrap();
3064        assert_eq!(res["pack"], "alpha");
3065    }
3066
3067    #[tokio::test]
3068    async fn dispatch_uses_allow_all_gate_by_default() {
3069        // No `with_gate` call — builder should use `AllowAllGate` so dispatch works.
3070        let reg = build_registry();
3071        let res = reg.dispatch("list", Value::Null).await.unwrap();
3072        assert_eq!(res["pack"], "alpha");
3073    }
3074
3075    // Captures the namespace each call sees so we can assert what the gate
3076    // actually receives, rather than assuming a hard-wired `default_ns()`.
3077    #[derive(Default, Debug)]
3078    struct NamespaceCapturingGate {
3079        seen: std::sync::Mutex<Vec<String>>,
3080    }
3081
3082    impl Gate for NamespaceCapturingGate {
3083        fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
3084            self.seen
3085                .lock()
3086                .unwrap()
3087                .push(req.namespace.as_str().to_string());
3088            Ok(GateDecision::allow())
3089        }
3090    }
3091
3092    #[tokio::test]
3093    async fn dispatch_propagates_params_namespace_to_gate() {
3094        let gate = Arc::new(NamespaceCapturingGate::default());
3095        let mut builder = VerbRegistryBuilder::new();
3096        builder.register(AlphaPack);
3097        builder.with_gate(gate.clone());
3098        builder.with_default_namespace("tenant-x");
3099        let reg = builder.build().expect("registry builds");
3100
3101        // Explicit namespace in params wins.
3102        reg.dispatch("list", serde_json::json!({"namespace": "tenant-y"}))
3103            .await
3104            .unwrap();
3105        // Missing namespace → registry default.
3106        reg.dispatch("list", Value::Null).await.unwrap();
3107        // Empty string is rejected: Namespace::parse("") fails → InvalidInput error.
3108        let err = reg
3109            .dispatch("list", serde_json::json!({"namespace": ""}))
3110            .await
3111            .unwrap_err();
3112        assert!(
3113            matches!(err, RuntimeError::InvalidInput(_)),
3114            "empty namespace must return InvalidInput, got {err:?}"
3115        );
3116
3117        let seen = gate.seen.lock().unwrap().clone();
3118        assert_eq!(seen, vec!["tenant-y", "tenant-x"]);
3119    }
3120
3121    #[tokio::test]
3122    async fn dispatch_falls_back_to_local_when_no_default_set() {
3123        // Builder default mirrors `Namespace::default_ns()`.
3124        let gate = Arc::new(NamespaceCapturingGate::default());
3125        let mut builder = VerbRegistryBuilder::new();
3126        builder.register(AlphaPack);
3127        builder.with_gate(gate.clone());
3128        let reg = builder.build().expect("registry builds");
3129
3130        reg.dispatch("list", Value::Null).await.unwrap();
3131        let seen = gate.seen.lock().unwrap().clone();
3132        assert_eq!(seen, vec!["local"]);
3133    }
3134
3135    /// A present-but-malformed `namespace` value must never reach the gate as
3136    /// the default namespace. Table-driven over every
3137    /// non-string JSON type; the gate-spy proves no call is ever recorded (the
3138    /// dispatch must short-circuit with `InvalidInput` before `GateRequest` is
3139    /// built), so the default namespace can never appear as a coerced stand-in.
3140    #[tokio::test]
3141    async fn namespace_null_rejected_not_coerced() {
3142        let cases: Vec<(&str, Value)> = vec![
3143            ("null", Value::Null),
3144            ("number", serde_json::json!(42)),
3145            ("boolean", serde_json::json!(true)),
3146            ("array", serde_json::json!(["local"])),
3147            ("object", serde_json::json!({"ns": "local"})),
3148        ];
3149
3150        for (label, ns_value) in cases {
3151            let gate = Arc::new(NamespaceCapturingGate::default());
3152            let mut builder = VerbRegistryBuilder::new();
3153            builder.register(AlphaPack);
3154            builder.with_gate(gate.clone());
3155            builder.with_default_namespace("tenant-x");
3156            let reg = builder.build().expect("registry builds");
3157
3158            let err = reg
3159                .dispatch("list", serde_json::json!({"namespace": ns_value}))
3160                .await
3161                .unwrap_err();
3162            assert!(
3163                matches!(err, RuntimeError::InvalidInput(_)),
3164                "case {label}: expected InvalidInput, got {err:?}"
3165            );
3166
3167            // The gate must never have been consulted for this malformed input —
3168            // proves no Allow decision (and therefore no default-namespace write)
3169            // can ever be reached for it.
3170            let seen = gate.seen.lock().unwrap().clone();
3171            assert!(
3172                seen.is_empty(),
3173                "case {label}: gate must not be consulted for malformed namespace, saw {seen:?}"
3174            );
3175        }
3176    }
3177
3178    // ---- Audit event emission ----
3179
3180    use khive_gate::{AuditDecision, AuditEvent, Obligation};
3181
3182    /// A gate that records every audit event emitted via from_check.
3183    #[derive(Default, Debug)]
3184    struct AuditCapturingGate {
3185        events: std::sync::Mutex<Vec<AuditEvent>>,
3186        deny_verb: Option<&'static str>,
3187    }
3188
3189    impl Gate for AuditCapturingGate {
3190        fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
3191            let decision = if Some(req.verb.as_str()) == self.deny_verb {
3192                GateDecision::deny("test deny")
3193            } else {
3194                GateDecision::allow_with(vec![Obligation::Audit {
3195                    tag: format!("{}.check", req.verb),
3196                }])
3197            };
3198            // Capture what dispatch will also emit.
3199            let ev = AuditEvent::from_check(req, &decision, self.impl_name());
3200            self.events.lock().unwrap().push(ev);
3201            Ok(decision)
3202        }
3203
3204        fn impl_name(&self) -> &'static str {
3205            "AuditCapturingGate"
3206        }
3207    }
3208
3209    #[tokio::test]
3210    async fn dispatch_emits_one_audit_event_per_call() {
3211        let gate = Arc::new(AuditCapturingGate::default());
3212        let mut builder = VerbRegistryBuilder::new();
3213        builder.register(AlphaPack);
3214        builder.with_gate(gate.clone());
3215        let reg = builder.build().expect("registry builds");
3216
3217        reg.dispatch("list", Value::Null).await.unwrap();
3218        reg.dispatch("create", Value::Null).await.unwrap();
3219
3220        let evs = gate.events.lock().unwrap();
3221        assert_eq!(evs.len(), 2, "exactly one audit event per dispatch call");
3222    }
3223
3224    #[tokio::test]
3225    async fn dispatch_audit_event_allow_carries_obligations() {
3226        let gate = Arc::new(AuditCapturingGate::default());
3227        let mut builder = VerbRegistryBuilder::new();
3228        builder.register(AlphaPack);
3229        builder.with_gate(gate.clone());
3230        let reg = builder.build().expect("registry builds");
3231
3232        reg.dispatch("list", Value::Null).await.unwrap();
3233
3234        let evs = gate.events.lock().unwrap();
3235        let ev = &evs[0];
3236        assert_eq!(ev.verb, "list");
3237        assert_eq!(ev.decision, AuditDecision::Allow);
3238        assert!(ev.deny_reason.is_none());
3239        assert_eq!(ev.obligations.len(), 1);
3240        assert_eq!(ev.gate_impl, "AuditCapturingGate");
3241    }
3242
3243    #[tokio::test]
3244    async fn dispatch_audit_event_deny_carries_reason() {
3245        let gate = Arc::new(AuditCapturingGate {
3246            events: Default::default(),
3247            deny_verb: Some("create"),
3248        });
3249        let mut builder = VerbRegistryBuilder::new();
3250        builder.register(AlphaPack);
3251        builder.with_gate(gate.clone());
3252        let reg = builder.build().expect("registry builds");
3253
3254        // Gate denies — dispatch returns PermissionDenied (hard enforcement).
3255        // The audit event is still recorded (captured inside the gate impl).
3256        let err = reg.dispatch("create", Value::Null).await.unwrap_err();
3257        assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
3258
3259        let evs = gate.events.lock().unwrap();
3260        let ev = &evs[0];
3261        assert_eq!(ev.verb, "create");
3262        assert_eq!(ev.decision, AuditDecision::Deny);
3263        assert_eq!(ev.deny_reason.as_deref(), Some("test deny"));
3264        assert!(ev.obligations.is_empty());
3265    }
3266
3267    #[tokio::test]
3268    async fn dispatch_audit_event_fields_match_gate_request() {
3269        let gate = Arc::new(AuditCapturingGate::default());
3270        let mut builder = VerbRegistryBuilder::new();
3271        builder.register(AlphaPack);
3272        builder.with_gate(gate.clone());
3273        builder.with_default_namespace("tenant-z");
3274        let reg = builder.build().expect("registry builds");
3275
3276        reg.dispatch("list", serde_json::json!({"namespace": "tenant-q"}))
3277            .await
3278            .unwrap();
3279
3280        let evs = gate.events.lock().unwrap();
3281        let ev = &evs[0];
3282        // Namespace from params wins.
3283        assert_eq!(ev.namespace, "tenant-q");
3284        assert_eq!(ev.verb, "list");
3285        assert_eq!(ev.actor.kind, "anonymous");
3286    }
3287
3288    // ---- Actor attribution threading into gate request (ADR-057) ----
3289
3290    /// A gate spy that captures the raw `GateRequest` it receives.
3291    #[derive(Default, Debug)]
3292    struct ActorCapturingGate {
3293        requests: std::sync::Mutex<Vec<GateRequest>>,
3294    }
3295
3296    impl Gate for ActorCapturingGate {
3297        fn check(&self, req: &GateRequest) -> Result<GateDecision, GateError> {
3298            self.requests.lock().unwrap().push(req.clone());
3299            Ok(GateDecision::allow())
3300        }
3301    }
3302
3303    /// When `actor_id` is configured, the gate request carries that actor, not
3304    /// anonymous. This exercises the ADR-057 attribution fix: the gate can
3305    /// distinguish an agent caller from an unauthenticated caller.
3306    #[tokio::test]
3307    async fn gate_request_carries_configured_actor_when_actor_id_is_set() {
3308        let gate = Arc::new(ActorCapturingGate::default());
3309        let mut builder = VerbRegistryBuilder::new();
3310        builder.register(AlphaPack);
3311        builder.with_gate(gate.clone());
3312        builder.with_actor_id(Some("team-abc:implementer".to_string()));
3313        let reg = builder.build().expect("registry builds");
3314
3315        reg.dispatch("list", Value::Null).await.unwrap();
3316
3317        let reqs = gate.requests.lock().unwrap();
3318        assert_eq!(reqs.len(), 1);
3319        let req = &reqs[0];
3320        assert_eq!(
3321            req.actor.kind, "actor",
3322            "gate request must carry kind='actor' when actor_id is configured"
3323        );
3324        assert_eq!(
3325            req.actor.id, "team-abc:implementer",
3326            "gate request must carry the configured actor id"
3327        );
3328    }
3329
3330    /// When no `actor_id` is configured, the gate request still receives the
3331    /// anonymous actor (no regression to the party-line default).
3332    #[tokio::test]
3333    async fn gate_request_carries_anonymous_when_no_actor_id_configured() {
3334        let gate = Arc::new(ActorCapturingGate::default());
3335        let mut builder = VerbRegistryBuilder::new();
3336        builder.register(AlphaPack);
3337        builder.with_gate(gate.clone());
3338        // actor_id left at default (None).
3339        let reg = builder.build().expect("registry builds");
3340
3341        reg.dispatch("list", Value::Null).await.unwrap();
3342
3343        let reqs = gate.requests.lock().unwrap();
3344        assert_eq!(reqs.len(), 1);
3345        let req = &reqs[0];
3346        assert_eq!(
3347            req.actor.kind, "anonymous",
3348            "gate request must carry anonymous actor when no actor_id is configured"
3349        );
3350        assert_eq!(req.actor.id, "local");
3351    }
3352
3353    /// A pack that records the `ActorRef` carried by the `NamespaceToken` it
3354    /// is dispatched with, so tests can compare it against the gate's actor.
3355    struct TokenCapturingPack {
3356        actors: Arc<std::sync::Mutex<Vec<khive_gate::ActorRef>>>,
3357    }
3358
3359    impl Pack for TokenCapturingPack {
3360        const NAME: &'static str = "alpha";
3361        const NOTE_KINDS: &'static [&'static str] = &[];
3362        const ENTITY_KINDS: &'static [&'static str] = &[];
3363        const HANDLERS: &'static [HandlerDef] = AlphaPack::HANDLERS;
3364    }
3365
3366    #[async_trait]
3367    impl PackRuntime for TokenCapturingPack {
3368        fn name(&self) -> &str {
3369            Self::NAME
3370        }
3371        fn note_kinds(&self) -> &'static [&'static str] {
3372            Self::NOTE_KINDS
3373        }
3374        fn entity_kinds(&self) -> &'static [&'static str] {
3375            Self::ENTITY_KINDS
3376        }
3377        fn handlers(&self) -> &'static [HandlerDef] {
3378            Self::HANDLERS
3379        }
3380        async fn dispatch(
3381            &self,
3382            verb: &str,
3383            _params: Value,
3384            _registry: &VerbRegistry,
3385            token: &NamespaceToken,
3386        ) -> Result<Value, RuntimeError> {
3387            self.actors.lock().unwrap().push(token.actor().clone());
3388            Ok(serde_json::json!({ "pack": "alpha", "verb": verb }))
3389        }
3390    }
3391
3392    /// The gate's actor and the storage token's actor must be the exact same
3393    /// resolved value: both come from one `resolve_actor` call
3394    /// (`resolved_actor`) instead of two independently hand-synchronized
3395    /// `match` expressions, so a future edit to one copy but not the other
3396    /// cannot silently desynchronize "who the gate thinks the caller is" from
3397    /// "who the storage layer thinks the caller is". Reintroducing a second
3398    /// independent actor-resolution copy for the token would regress this and
3399    /// this test would catch it.
3400    #[tokio::test]
3401    async fn gate_actor_and_token_actor_are_identical_when_actor_id_is_set() {
3402        let gate = Arc::new(ActorCapturingGate::default());
3403        let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3404        let pack = TokenCapturingPack {
3405            actors: actors.clone(),
3406        };
3407        let mut builder = VerbRegistryBuilder::new();
3408        builder.register(pack);
3409        builder.with_gate(gate.clone());
3410        builder.with_actor_id(Some("actor-alpha".to_string()));
3411        let reg = builder.build().expect("registry builds");
3412
3413        reg.dispatch("list", Value::Null).await.unwrap();
3414
3415        let reqs = gate.requests.lock().unwrap();
3416        let gate_actor = reqs[0].actor.clone();
3417        drop(reqs);
3418
3419        let captured = actors.lock().unwrap();
3420        let token_actor = captured[0].clone();
3421
3422        assert_eq!(
3423            gate_actor.kind, token_actor.kind,
3424            "gate request actor and storage token actor must carry the same kind"
3425        );
3426        assert_eq!(
3427            gate_actor.id, token_actor.id,
3428            "gate request actor and storage token actor must carry the same id"
3429        );
3430        assert_eq!(gate_actor.id, "actor-alpha");
3431    }
3432
3433    /// Same identity check with no configured `actor_id`: both the gate and
3434    /// the storage token must independently land on `ActorRef::anonymous()`.
3435    #[tokio::test]
3436    async fn gate_actor_and_token_actor_are_identical_when_anonymous() {
3437        let gate = Arc::new(ActorCapturingGate::default());
3438        let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3439        let pack = TokenCapturingPack {
3440            actors: actors.clone(),
3441        };
3442        let mut builder = VerbRegistryBuilder::new();
3443        builder.register(pack);
3444        builder.with_gate(gate.clone());
3445        let reg = builder.build().expect("registry builds");
3446
3447        reg.dispatch("list", Value::Null).await.unwrap();
3448
3449        let reqs = gate.requests.lock().unwrap();
3450        let gate_actor = reqs[0].actor.clone();
3451        drop(reqs);
3452
3453        let captured = actors.lock().unwrap();
3454        let token_actor = captured[0].clone();
3455
3456        assert_eq!(gate_actor.kind, token_actor.kind);
3457        assert_eq!(gate_actor.id, token_actor.id);
3458        assert_eq!(gate_actor.id, "local");
3459    }
3460
3461    // ---- dispatch_as: verified-actor dispatch for embedding hosts ----
3462
3463    /// `dispatch_as` must thread the caller-supplied verified actor through
3464    /// to the pack handler's `NamespaceToken`, exactly as `dispatch_with_identity`
3465    /// does with a `RequestIdentity.actor_id` — this is the observable
3466    /// contract embedding hosts rely on.
3467    #[tokio::test]
3468    async fn dispatch_as_threads_verified_actor_into_token() {
3469        let gate = Arc::new(ActorCapturingGate::default());
3470        let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3471        let pack = TokenCapturingPack {
3472            actors: actors.clone(),
3473        };
3474        let mut builder = VerbRegistryBuilder::new();
3475        builder.register(pack);
3476        builder.with_gate(gate.clone());
3477        let reg = builder.build().expect("registry builds");
3478
3479        reg.dispatch_as(
3480            "list",
3481            Value::Null,
3482            VerifiedActor::new("gateway:principal-42").unwrap(),
3483        )
3484        .await
3485        .unwrap();
3486
3487        let reqs = gate.requests.lock().unwrap();
3488        assert_eq!(reqs[0].actor.kind, "actor");
3489        assert_eq!(reqs[0].actor.id, "gateway:principal-42");
3490        drop(reqs);
3491
3492        let captured = actors.lock().unwrap();
3493        assert_eq!(captured[0].kind, "actor");
3494        assert_eq!(
3495            captured[0].id, "gateway:principal-42",
3496            "the storage token actor must be the verified_actor supplied to dispatch_as, \
3497             matching exactly what pack handlers read as the acting principal"
3498        );
3499    }
3500
3501    /// `dispatch_as` is purely additive: a registry with a baked `actor_id`
3502    /// must still serve plain `dispatch()` calls under its own baked actor,
3503    /// unaffected by any `dispatch_as` call made on the same (cheaply
3504    /// cloneable) registry. No shared mutable state links the two calls.
3505    #[tokio::test]
3506    async fn dispatch_as_does_not_change_plain_dispatch_behavior() {
3507        let gate = Arc::new(ActorCapturingGate::default());
3508        let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3509        let pack = TokenCapturingPack {
3510            actors: actors.clone(),
3511        };
3512        let mut builder = VerbRegistryBuilder::new();
3513        builder.register(pack);
3514        builder.with_gate(gate.clone());
3515        builder.with_actor_id(Some("baked-actor".to_string()));
3516        let reg = builder.build().expect("registry builds");
3517
3518        reg.dispatch_as(
3519            "list",
3520            Value::Null,
3521            VerifiedActor::new("verified-actor").unwrap(),
3522        )
3523        .await
3524        .unwrap();
3525        reg.dispatch("list", Value::Null).await.unwrap();
3526
3527        let captured = actors.lock().unwrap();
3528        assert_eq!(captured.len(), 2);
3529        assert_eq!(captured[0].id, "verified-actor", "dispatch_as call");
3530        assert_eq!(
3531            captured[1].id, "baked-actor",
3532            "a later plain dispatch() call must still use the registry's baked \
3533             actor_id, unaffected by the prior dispatch_as call"
3534        );
3535    }
3536
3537    /// A request `params` payload cannot inject or override the actor:
3538    /// `dispatch_as` resolves the acting principal solely from its Rust-side
3539    /// `verified_actor` argument, never from `params`. An `actor` key placed
3540    /// in `params` passes through untouched to the pack handler like any
3541    /// other unrecognized field — the dispatch boundary itself never reads
3542    /// `params["actor"]`.
3543    #[tokio::test]
3544    async fn dispatch_as_ignores_actor_key_in_params() {
3545        let gate = Arc::new(ActorCapturingGate::default());
3546        let actors = Arc::new(std::sync::Mutex::new(Vec::new()));
3547        let pack = TokenCapturingPack {
3548            actors: actors.clone(),
3549        };
3550        let mut builder = VerbRegistryBuilder::new();
3551        builder.register(pack);
3552        builder.with_gate(gate.clone());
3553        let reg = builder.build().expect("registry builds");
3554
3555        reg.dispatch_as(
3556            "list",
3557            serde_json::json!({"actor": "spoofed-actor"}),
3558            VerifiedActor::new("verified-actor").unwrap(),
3559        )
3560        .await
3561        .unwrap();
3562
3563        let captured = actors.lock().unwrap();
3564        assert_eq!(
3565            captured[0].id, "verified-actor",
3566            "an 'actor' key inside params must never override the verified_actor \
3567             argument threaded through dispatch_as"
3568        );
3569    }
3570
3571    /// `VerifiedActor::new` must reject an empty identifier rather than
3572    /// letting it reach dispatch and silently resolve to the anonymous actor.
3573    #[test]
3574    fn verified_actor_rejects_empty_identifier() {
3575        let err = VerifiedActor::new("").unwrap_err();
3576        assert!(
3577            matches!(err, RuntimeError::InvalidInput(_)),
3578            "expected InvalidInput, got {err:?}"
3579        );
3580    }
3581
3582    /// `VerifiedActor::new` must reject a whitespace-only identifier for the
3583    /// same reason: it must never launder into `ActorRef::anonymous()`.
3584    #[test]
3585    fn verified_actor_rejects_whitespace_only_identifier() {
3586        let err = VerifiedActor::new("   ").unwrap_err();
3587        assert!(
3588            matches!(err, RuntimeError::InvalidInput(_)),
3589            "expected InvalidInput, got {err:?}"
3590        );
3591    }
3592
3593    // ---- Rego gate: fail-closed end-to-end ----
3594
3595    /// A `RegoGate` whose policy lacks the named entrypoint rule must cause
3596    /// `VerbRegistry::dispatch` to return `RuntimeError::PermissionDenied` —
3597    /// never to proceed to the pack handler.
3598    ///
3599    /// This is the runtime-level assertion that a gate evaluation failure
3600    /// fails closed rather than opening a security hole.
3601    /// `RegoGate::check` converts all evaluation failures (missing rule,
3602    /// undefined result, serialization error, poisoned engine) to
3603    /// `Ok(GateDecision::Deny)`, so dispatch is blocked. The runtime's
3604    /// fail-open `Err(_)` branch remains for non-evaluation gate errors
3605    /// (e.g. infrastructure faults from other `Gate` implementations).
3606    #[tokio::test]
3607    async fn rego_gate_missing_entrypoint_returns_permission_denied() {
3608        use khive_gate_rego::RegoGate;
3609
3610        // Policy defines `verdict` but NOT `data.khive.gate.decision` (the
3611        // default entrypoint).  Construction succeeds — from_policy_str does
3612        // not validate the default entrypoint.  check() must convert the
3613        // missing-rule evaluation error to Ok(Deny) so the runtime denies
3614        // the request rather than treating the Err as a fail-open signal.
3615        let policy = r#"
3616            package khive.gate
3617            import rego.v1
3618            verdict := "allow"
3619        "#;
3620        let gate = Arc::new(RegoGate::from_policy_str(policy).expect("policy compiles"));
3621
3622        let mut builder = VerbRegistryBuilder::new();
3623        builder.register(AlphaPack);
3624        builder.with_gate(gate);
3625        let reg = builder.build().expect("registry builds");
3626
3627        let err = reg.dispatch("create", Value::Null).await.unwrap_err();
3628        assert!(
3629            matches!(err, RuntimeError::PermissionDenied { ref verb, .. } if verb == "create"),
3630            "expected PermissionDenied for missing rego entrypoint, got {err:?}"
3631        );
3632    }
3633
3634    // ---- Audit tracing emission ----
3635    //
3636    // The AuditCapturingGate tests above prove that AuditEvent::from_check is
3637    // called with the right inputs, but they observe the event *inside* the
3638    // gate impl — they would still pass if dispatch's
3639    // `tracing::info!(audit_event = ..., "gate.check")` were deleted or
3640    // renamed. The tests below install a capture Layer and assert on the
3641    // actual tracing event surfaced from dispatch. This locks the public
3642    // observability contract: one `gate.check` info event per dispatch,
3643    // carrying an `audit_event` field that round-trips back to an `AuditEvent`.
3644
3645    use std::sync::{Mutex as StdMutex, Once, OnceLock};
3646
3647    use serial_test::serial;
3648    use tracing::field::{Field, Visit};
3649
3650    #[derive(Clone, Debug, Default)]
3651    struct CapturedEvent {
3652        message: Option<String>,
3653        audit_event: Option<String>,
3654    }
3655
3656    #[derive(Default)]
3657    struct CapturedEventVisitor(CapturedEvent);
3658
3659    impl Visit for CapturedEventVisitor {
3660        fn record_str(&mut self, field: &Field, value: &str) {
3661            match field.name() {
3662                "message" => self.0.message = Some(value.to_string()),
3663                "audit_event" => self.0.audit_event = Some(value.to_string()),
3664                _ => {}
3665            }
3666        }
3667
3668        fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
3669            // `tracing::info!(audit_event = %expr, "msg")` records via the
3670            // Display-wrapped Debug path, so we receive the JSON string here.
3671            // `"msg"` literal records as a `message` field via `record_debug`
3672            // with a quoted Debug representation; strip the surrounding quotes
3673            // so the captured message matches the source.
3674            let formatted = format!("{value:?}");
3675            let cleaned = formatted
3676                .trim_start_matches('"')
3677                .trim_end_matches('"')
3678                .to_string();
3679            match field.name() {
3680                "message" => self.0.message = Some(cleaned),
3681                "audit_event" => self.0.audit_event = Some(cleaned),
3682                _ => {}
3683            }
3684        }
3685    }
3686
3687    /// Minimal `tracing::Subscriber` that captures events into a shared vec.
3688    ///
3689    /// Implemented directly (without `tracing_subscriber::registry()` layering)
3690    /// to avoid the layer machinery that can cause thread-local dispatch to be
3691    /// bypassed when the registry's internal global state is initialised by
3692    /// another subscriber in the same test binary.
3693    ///
3694    /// Isolation across concurrent tests is handled at the dispatcher level by
3695    /// `tracing::dispatcher::with_default`, which installs this subscriber
3696    /// as the thread-local default for the duration of the test closure.
3697    /// Other threads (e.g. `#[tokio::test]` pool workers) emit through their
3698    /// own (typically NoSubscriber) dispatchers and never reach this instance.
3699    struct CaptureSubscriber {
3700        events: Arc<StdMutex<Vec<CapturedEvent>>>,
3701    }
3702
3703    impl CaptureSubscriber {
3704        fn new(events: Arc<StdMutex<Vec<CapturedEvent>>>) -> Self {
3705            Self { events }
3706        }
3707    }
3708
3709    impl tracing::Subscriber for CaptureSubscriber {
3710        fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
3711            true
3712        }
3713        fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
3714            tracing::span::Id::from_u64(1)
3715        }
3716        fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
3717        fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
3718        fn event(&self, event: &tracing::Event<'_>) {
3719            let mut visitor = CapturedEventVisitor::default();
3720            event.record(&mut visitor);
3721            self.events.lock().unwrap().push(visitor.0);
3722        }
3723        fn enter(&self, _: &tracing::span::Id) {}
3724        fn exit(&self, _: &tracing::span::Id) {}
3725    }
3726
3727    /// Global capture buffer for the tracing tests.
3728    ///
3729    /// The subscriber is installed exactly once via `set_global_default`
3730    /// (thread-local dispatchers via `with_default` proved unreliable when
3731    /// other tests in the binary configure their own dispatchers in parallel —
3732    /// the global state interacted unpredictably and events were lost).
3733    ///
3734    /// Each test that uses this buffer is `#[serial]`, so only one
3735    /// runs at a time. The buffer is cleared at the start of each capture call.
3736    static GLOBAL_CAPTURE: OnceLock<Arc<StdMutex<Vec<CapturedEvent>>>> = OnceLock::new();
3737    static GLOBAL_INIT: Once = Once::new();
3738
3739    fn global_capture() -> Arc<StdMutex<Vec<CapturedEvent>>> {
3740        GLOBAL_INIT.call_once(|| {
3741            let buffer = Arc::new(StdMutex::new(Vec::new()));
3742            let subscriber = CaptureSubscriber::new(Arc::clone(&buffer));
3743            // Ignore error: if another subscriber is already set globally, our
3744            // subscriber installation fails, but the buffer will simply stay
3745            // empty and tests will fail with a clear "got 0 events" message
3746            // rather than a silent corruption.
3747            let _ = tracing::subscriber::set_global_default(subscriber);
3748            let _ = GLOBAL_CAPTURE.set(buffer);
3749        });
3750        Arc::clone(GLOBAL_CAPTURE.get().expect("global capture initialized"))
3751    }
3752
3753    /// Run an async block under the global capture subscriber and return
3754    /// the events emitted during the run. Clears the buffer at the start.
3755    ///
3756    /// Callers MUST be `#[serial]` to prevent concurrent buffer pollution.
3757    fn capture_dispatch_events<Fut>(future: Fut) -> Vec<CapturedEvent>
3758    where
3759        Fut: std::future::Future<Output = ()>,
3760    {
3761        let buffer = global_capture();
3762        buffer.lock().unwrap().clear();
3763
3764        let rt = tokio::runtime::Builder::new_current_thread()
3765            .enable_all()
3766            .build()
3767            .expect("build current-thread tokio runtime");
3768        rt.block_on(future);
3769
3770        let result = buffer.lock().unwrap().clone();
3771        result
3772    }
3773
3774    /// Pull every captured event whose `message` matches `"gate.check"` AND
3775    /// whose audit_event JSON declares the expected `gate_impl` name.
3776    ///
3777    /// Filtering by `gate_impl` lets concurrent tests in the same binary
3778    /// emit their own gate.check events into the global capture buffer
3779    /// without polluting each others' counts.
3780    fn gate_check_events_for(events: &[CapturedEvent], gate_impl: &str) -> Vec<CapturedEvent> {
3781        events
3782            .iter()
3783            .filter(|e| e.message.as_deref() == Some("gate.check"))
3784            .filter(|e| {
3785                e.audit_event
3786                    .as_deref()
3787                    .and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok())
3788                    .and_then(|v| {
3789                        v.get("gate_impl")
3790                            .and_then(|g| g.as_str().map(|s| s.to_string()))
3791                    })
3792                    .as_deref()
3793                    == Some(gate_impl)
3794            })
3795            .cloned()
3796            .collect()
3797    }
3798
3799    #[test]
3800    #[serial]
3801    fn dispatch_tracing_emits_one_gate_check_event_on_allow() {
3802        #[derive(Debug)]
3803        struct TracingAllowGate;
3804        impl Gate for TracingAllowGate {
3805            fn check(&self, _: &GateRequest) -> Result<GateDecision, GateError> {
3806                Ok(GateDecision::allow())
3807            }
3808            fn impl_name(&self) -> &'static str {
3809                "TracingAllowGate"
3810            }
3811        }
3812
3813        let events = capture_dispatch_events(async {
3814            let mut builder = VerbRegistryBuilder::new();
3815            builder.register(AlphaPack);
3816            builder.with_gate(Arc::new(TracingAllowGate));
3817            builder.with_default_namespace("tenant-default");
3818            let reg = builder.build().expect("registry builds");
3819            reg.dispatch("list", serde_json::json!({"namespace": "tenant-q"}))
3820                .await
3821                .unwrap();
3822        });
3823
3824        let gate_events = gate_check_events_for(&events, "TracingAllowGate");
3825        assert_eq!(
3826            gate_events.len(),
3827            1,
3828            "exactly one gate.check tracing event per dispatch (allow); got {gate_events:?}"
3829        );
3830        let payload = gate_events[0]
3831            .audit_event
3832            .as_ref()
3833            .expect("gate.check event must carry an audit_event field");
3834        let audit: khive_gate::AuditEvent =
3835            serde_json::from_str(payload).expect("audit_event payload must decode to AuditEvent");
3836        assert_eq!(audit.decision, AuditDecision::Allow);
3837        assert_eq!(audit.verb, "list");
3838        assert_eq!(audit.namespace, "tenant-q");
3839        assert_eq!(audit.gate_impl, "TracingAllowGate");
3840        assert!(
3841            audit.deny_reason.is_none(),
3842            "deny_reason must be None on Allow"
3843        );
3844    }
3845
3846    // ---- Hard enforcement + EventStore persistence ----
3847
3848    use crate::runtime::NamespaceToken;
3849    use async_trait::async_trait;
3850    use khive_storage::{
3851        BatchWriteSummary, Event, EventFilter, EventStore, Page, PageRequest, SubstrateKind,
3852    };
3853    use khive_types::EventOutcome;
3854
3855    /// In-memory EventStore for unit tests — avoids file-backed SQLite.
3856    #[derive(Default, Debug)]
3857    struct MemoryEventStore {
3858        events: std::sync::Mutex<Vec<Event>>,
3859    }
3860
3861    #[async_trait]
3862    impl EventStore for MemoryEventStore {
3863        async fn append_event(&self, event: Event) -> khive_storage::StorageResult<()> {
3864            self.events.lock().unwrap().push(event);
3865            Ok(())
3866        }
3867        async fn append_events(
3868            &self,
3869            events: Vec<Event>,
3870        ) -> khive_storage::StorageResult<BatchWriteSummary> {
3871            let attempted = events.len() as u64;
3872            let affected = attempted;
3873            self.events.lock().unwrap().extend(events);
3874            Ok(BatchWriteSummary {
3875                attempted,
3876                affected,
3877                failed: 0,
3878                first_error: String::new(),
3879            })
3880        }
3881        async fn get_event(&self, id: uuid::Uuid) -> khive_storage::StorageResult<Option<Event>> {
3882            Ok(self
3883                .events
3884                .lock()
3885                .unwrap()
3886                .iter()
3887                .find(|e| e.id == id)
3888                .cloned())
3889        }
3890        async fn query_events(
3891            &self,
3892            _filter: EventFilter,
3893            _page: PageRequest,
3894        ) -> khive_storage::StorageResult<Page<Event>> {
3895            let items = self.events.lock().unwrap().clone();
3896            let total = items.len() as u64;
3897            Ok(Page {
3898                items,
3899                total: Some(total),
3900            })
3901        }
3902        async fn count_events(&self, _filter: EventFilter) -> khive_storage::StorageResult<u64> {
3903            Ok(self.events.lock().unwrap().len() as u64)
3904        }
3905    }
3906
3907    #[tokio::test]
3908    async fn allow_all_gate_default_remains_backward_compatible() {
3909        // No gate set — AllowAllGate is the default. Dispatch must succeed.
3910        let mut builder = VerbRegistryBuilder::new();
3911        builder.register(AlphaPack);
3912        let reg = builder.build().expect("registry builds");
3913
3914        let res = reg.dispatch("list", Value::Null).await.unwrap();
3915        assert_eq!(
3916            res["pack"], "alpha",
3917            "AllowAllGate must allow every verb — backward compat guarantee"
3918        );
3919        let res = reg.dispatch("create", Value::Null).await.unwrap();
3920        assert_eq!(res["pack"], "alpha");
3921    }
3922
3923    #[tokio::test]
3924    async fn deny_gate_returns_permission_denied_pack_never_invoked() {
3925        #[derive(Debug)]
3926        struct AlwaysDenyGate;
3927        impl Gate for AlwaysDenyGate {
3928            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
3929                Ok(GateDecision::deny("test: always deny"))
3930            }
3931        }
3932
3933        // Track whether dispatch was ever invoked on the pack.
3934        #[derive(Debug)]
3935        struct TrackedPack {
3936            invoked: Arc<AtomicUsize>,
3937        }
3938
3939        impl khive_types::Pack for TrackedPack {
3940            const NAME: &'static str = "tracked";
3941            const NOTE_KINDS: &'static [&'static str] = &[];
3942            const ENTITY_KINDS: &'static [&'static str] = &[];
3943            const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
3944                name: "guarded",
3945                description: "a guarded verb",
3946                visibility: Visibility::Verb,
3947                category: VerbCategory::Assertive,
3948                params: &[],
3949            }];
3950        }
3951
3952        #[async_trait]
3953        impl PackRuntime for TrackedPack {
3954            fn name(&self) -> &str {
3955                Self::NAME
3956            }
3957            fn note_kinds(&self) -> &'static [&'static str] {
3958                Self::NOTE_KINDS
3959            }
3960            fn entity_kinds(&self) -> &'static [&'static str] {
3961                Self::ENTITY_KINDS
3962            }
3963            fn handlers(&self) -> &'static [HandlerDef] {
3964                Self::HANDLERS
3965            }
3966            async fn dispatch(
3967                &self,
3968                _verb: &str,
3969                _params: Value,
3970                _registry: &VerbRegistry,
3971                _token: &NamespaceToken,
3972            ) -> Result<Value, RuntimeError> {
3973                self.invoked.fetch_add(1, Ordering::SeqCst);
3974                Ok(serde_json::json!({"invoked": true}))
3975            }
3976        }
3977
3978        let invoked = Arc::new(AtomicUsize::new(0));
3979        let mut builder = VerbRegistryBuilder::new();
3980        builder.register(TrackedPack {
3981            invoked: invoked.clone(),
3982        });
3983        builder.with_gate(Arc::new(AlwaysDenyGate));
3984        let reg = builder.build().expect("registry builds");
3985
3986        let err = reg.dispatch("guarded", Value::Null).await.unwrap_err();
3987        assert!(
3988            matches!(err, RuntimeError::PermissionDenied { ref verb, ref reason } if verb == "guarded" && reason.contains("always deny")),
3989            "expected PermissionDenied with verb=guarded and reason, got: {err:?}"
3990        );
3991        assert_eq!(
3992            invoked.load(Ordering::SeqCst),
3993            0,
3994            "pack dispatch MUST NOT be invoked when gate denies"
3995        );
3996    }
3997
3998    #[tokio::test]
3999    async fn audit_event_persists_to_event_store_on_allow() {
4000        let store = Arc::new(MemoryEventStore::default());
4001        let mut builder = VerbRegistryBuilder::new();
4002        builder.register(AlphaPack);
4003        builder.with_event_store(store.clone());
4004        let reg = builder.build().expect("registry builds");
4005
4006        reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4007            .await
4008            .unwrap();
4009
4010        let count = store.count_events(EventFilter::default()).await.unwrap();
4011        assert_eq!(count, 1, "one audit event persisted to EventStore on allow");
4012
4013        let page = store
4014            .query_events(
4015                EventFilter::default(),
4016                PageRequest {
4017                    limit: 10,
4018                    offset: 0,
4019                },
4020            )
4021            .await
4022            .unwrap();
4023        let ev = &page.items[0];
4024        assert_eq!(ev.verb, "list");
4025        assert_eq!(ev.namespace, "test-ns");
4026        assert_eq!(ev.substrate, SubstrateKind::Event);
4027        assert_eq!(ev.outcome, EventOutcome::Success);
4028    }
4029
4030    #[tokio::test]
4031    async fn audit_event_duration_us_reflects_measured_dispatch_time() {
4032        // The persisted audit row's `duration_us` must carry the measured
4033        // pack-dispatch time, not the `Event::new` default of 0 (persisting
4034        // the row before dispatch ran always yielded 0). `SleepingPack`
4035        // sleeps 20ms so the assertion has a wide, non-flaky margin over
4036        // scheduling jitter.
4037        let store = Arc::new(MemoryEventStore::default());
4038        let mut builder = VerbRegistryBuilder::new();
4039        builder.register(SleepingPack);
4040        builder.with_event_store(store.clone());
4041        let reg = builder.build().expect("registry builds");
4042
4043        reg.dispatch("slow_op", serde_json::json!({}))
4044            .await
4045            .unwrap();
4046
4047        let page = store
4048            .query_events(
4049                EventFilter::default(),
4050                PageRequest {
4051                    limit: 10,
4052                    offset: 0,
4053                },
4054            )
4055            .await
4056            .unwrap();
4057        assert_eq!(page.items.len(), 1);
4058        let ev = &page.items[0];
4059        assert!(
4060            ev.duration_us >= 10_000,
4061            "duration_us must reflect the ~20ms measured dispatch time, got {}",
4062            ev.duration_us
4063        );
4064    }
4065
4066    #[tokio::test]
4067    async fn dispatch_unknown_verb_allowed_by_gate_still_persists_audit_row() {
4068        // Generalizing audit-row deferral to every Allow-outcome verb (not
4069        // just singleton `link`) must not silently drop the audit row for a
4070        // verb the gate allows but no pack owns. `duration_us` stays at the
4071        // `Event::new` default of 0 here since no dispatch ever ran to measure.
4072        let store = Arc::new(MemoryEventStore::default());
4073        let mut builder = VerbRegistryBuilder::new();
4074        builder.register(AlphaPack);
4075        builder.with_event_store(store.clone());
4076        let reg = builder.build().expect("registry builds");
4077
4078        let result = reg.dispatch("no_such_verb", serde_json::json!({})).await;
4079        assert!(result.is_err(), "unknown verb must still return an error");
4080
4081        let count = store.count_events(EventFilter::default()).await.unwrap();
4082        assert_eq!(
4083            count, 1,
4084            "an allowed-but-unknown verb must still persist one audit row"
4085        );
4086        let page = store
4087            .query_events(
4088                EventFilter::default(),
4089                PageRequest {
4090                    limit: 10,
4091                    offset: 0,
4092                },
4093            )
4094            .await
4095            .unwrap();
4096        assert_eq!(page.items[0].duration_us, 0);
4097        // Dispatch returns InvalidInput for an unknown verb, so the
4098        // persisted outcome must be Error, not the previously-hardcoded
4099        // Success.
4100        assert_eq!(page.items[0].outcome, EventOutcome::Error);
4101    }
4102
4103    #[tokio::test]
4104    async fn audit_event_persists_to_event_store_on_deny() {
4105        #[derive(Debug)]
4106        struct AlwaysDenyGate;
4107        impl Gate for AlwaysDenyGate {
4108            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4109                Ok(GateDecision::deny("denied by test"))
4110            }
4111        }
4112
4113        let store = Arc::new(MemoryEventStore::default());
4114        let mut builder = VerbRegistryBuilder::new();
4115        builder.register(AlphaPack);
4116        builder.with_gate(Arc::new(AlwaysDenyGate));
4117        builder.with_event_store(store.clone());
4118        let reg = builder.build().expect("registry builds");
4119
4120        // Hard enforce → PermissionDenied returned.
4121        let err = reg
4122            .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4123            .await
4124            .unwrap_err();
4125        assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
4126
4127        let count = store.count_events(EventFilter::default()).await.unwrap();
4128        assert_eq!(count, 1, "one audit event persisted to EventStore on deny");
4129
4130        let page = store
4131            .query_events(
4132                EventFilter::default(),
4133                PageRequest {
4134                    limit: 10,
4135                    offset: 0,
4136                },
4137            )
4138            .await
4139            .unwrap();
4140        let ev = &page.items[0];
4141        assert_eq!(ev.verb, "list");
4142        assert_eq!(ev.outcome, EventOutcome::Denied);
4143    }
4144
4145    #[tokio::test]
4146    async fn gate_error_does_not_persist_to_event_store() {
4147        #[derive(Debug)]
4148        struct FailingGate;
4149        impl Gate for FailingGate {
4150            fn check(&self, _req: &GateRequest) -> Result<GateDecision, khive_gate::GateError> {
4151                Err(khive_gate::GateError::Internal("gate broken".into()))
4152            }
4153        }
4154
4155        let store = Arc::new(MemoryEventStore::default());
4156        let mut builder = VerbRegistryBuilder::new();
4157        builder.register(AlphaPack);
4158        builder.with_gate(Arc::new(FailingGate));
4159        builder.with_event_store(store.clone());
4160        let reg = builder.build().expect("registry builds");
4161
4162        // Gate Err → fail-open, dispatch proceeds.
4163        let res = reg.dispatch("list", Value::Null).await.unwrap();
4164        assert_eq!(
4165            res["pack"], "alpha",
4166            "gate error must fail-open, not block dispatch"
4167        );
4168
4169        let count = store.count_events(EventFilter::default()).await.unwrap();
4170        assert_eq!(
4171            count, 0,
4172            "gate infrastructure error must NOT produce an audit event in EventStore"
4173        );
4174    }
4175
4176    #[tokio::test]
4177    async fn no_event_store_configured_tracing_only() {
4178        // When no event_store is configured, dispatch must succeed without error.
4179        // (The tracing path is exercised in the tracing tests above; here we just
4180        // verify the absence of event_store does not break dispatch.)
4181        let mut builder = VerbRegistryBuilder::new();
4182        builder.register(AlphaPack);
4183        let reg = builder.build().expect("registry builds");
4184
4185        let res = reg.dispatch("list", Value::Null).await.unwrap();
4186        assert_eq!(res["pack"], "alpha");
4187    }
4188
4189    #[test]
4190    #[serial]
4191    fn dispatch_tracing_emits_gate_check_event_with_deny_payload() {
4192        #[derive(Debug)]
4193        struct TracingDenyGate;
4194        impl Gate for TracingDenyGate {
4195            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4196                Ok(GateDecision::deny("denied by test gate"))
4197            }
4198            fn impl_name(&self) -> &'static str {
4199                "TracingDenyGate"
4200            }
4201        }
4202
4203        let events = capture_dispatch_events(async {
4204            let mut builder = VerbRegistryBuilder::new();
4205            builder.register(AlphaPack);
4206            builder.with_gate(Arc::new(TracingDenyGate));
4207            let reg = builder.build().expect("registry builds");
4208            // Hard enforcement — dispatch returns PermissionDenied on Deny.
4209            // The tracing audit event is still emitted before the error is returned.
4210            let _ = reg.dispatch("create", serde_json::Value::Null).await;
4211        });
4212
4213        let gate_events = gate_check_events_for(&events, "TracingDenyGate");
4214        assert_eq!(
4215            gate_events.len(),
4216            1,
4217            "exactly one gate.check tracing event per dispatch (deny); got {gate_events:?}"
4218        );
4219        let payload = gate_events[0]
4220            .audit_event
4221            .as_ref()
4222            .expect("gate.check event must carry an audit_event field on Deny");
4223        let audit: khive_gate::AuditEvent =
4224            serde_json::from_str(payload).expect("audit_event payload must decode to AuditEvent");
4225        assert_eq!(audit.decision, AuditDecision::Deny);
4226        assert_eq!(audit.deny_reason.as_deref(), Some("denied by test gate"));
4227        assert_eq!(audit.gate_impl, "TracingDenyGate");
4228        // Wire-shape rule: obligations is always serialized as an array, empty
4229        // on Deny. Round-trip back through serde_json::Value to confirm the
4230        // field exists on the wire and is `[]`, not missing.
4231        let payload_json: serde_json::Value =
4232            serde_json::from_str(payload).expect("payload must be valid JSON");
4233        assert_eq!(
4234            payload_json["obligations"],
4235            serde_json::Value::Array(Vec::new()),
4236            "obligations must be `[]` on Deny on the tracing payload, not omitted"
4237        );
4238    }
4239
4240    // ---- EventStore audit envelope round-trip ----
4241    //
4242    // EventStore must not persist a summary Event without the full
4243    // AuditEvent fields (deny_reason, gate_impl, obligations). This test
4244    // verifies the complete envelope survives append_event → query_events.
4245
4246    #[tokio::test]
4247    async fn audit_envelope_round_trips_deny_reason_and_gate_impl_through_event_store() {
4248        #[derive(Debug)]
4249        struct DenyGateWithName;
4250        impl Gate for DenyGateWithName {
4251            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4252                Ok(GateDecision::deny("policy: write forbidden for anon"))
4253            }
4254            fn impl_name(&self) -> &'static str {
4255                "DenyGateWithName"
4256            }
4257        }
4258
4259        let store = Arc::new(MemoryEventStore::default());
4260        let mut builder = VerbRegistryBuilder::new();
4261        builder.register(AlphaPack);
4262        builder.with_gate(Arc::new(DenyGateWithName));
4263        builder.with_event_store(store.clone());
4264        let reg = builder.build().expect("registry builds");
4265
4266        // Dispatch is denied — PermissionDenied returned.
4267        let err = reg
4268            .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4269            .await
4270            .unwrap_err();
4271        assert!(
4272            matches!(err, RuntimeError::PermissionDenied { .. }),
4273            "expected PermissionDenied, got {err:?}"
4274        );
4275
4276        // Exactly one event in the store.
4277        let page = store
4278            .query_events(
4279                EventFilter::default(),
4280                PageRequest {
4281                    limit: 10,
4282                    offset: 0,
4283                },
4284            )
4285            .await
4286            .unwrap();
4287        assert_eq!(
4288            page.items.len(),
4289            1,
4290            "one audit event must be persisted on deny"
4291        );
4292
4293        let ev = &page.items[0];
4294        assert_eq!(ev.outcome, EventOutcome::Denied);
4295
4296        // The payload field must hold the full AuditEvent envelope.
4297        let data = &ev.payload;
4298
4299        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4300            .expect("Event.payload must deserialize to AuditEvent");
4301
4302        assert_eq!(
4303            audit.deny_reason.as_deref(),
4304            Some("policy: write forbidden for anon"),
4305            "deny_reason must be preserved through EventStore"
4306        );
4307        assert_eq!(
4308            audit.gate_impl, "DenyGateWithName",
4309            "gate_impl must be preserved through EventStore"
4310        );
4311        assert_eq!(
4312            audit.decision,
4313            khive_gate::AuditDecision::Deny,
4314            "decision field must be preserved through EventStore"
4315        );
4316    }
4317
4318    #[tokio::test]
4319    async fn audit_envelope_round_trips_obligations_through_event_store() {
4320        use khive_gate::Obligation;
4321
4322        #[derive(Debug)]
4323        struct ObligationGate;
4324        impl Gate for ObligationGate {
4325            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4326                Ok(GateDecision::allow_with(vec![Obligation::Audit {
4327                    tag: "billing.meter".into(),
4328                }]))
4329            }
4330            fn impl_name(&self) -> &'static str {
4331                "ObligationGate"
4332            }
4333        }
4334
4335        let store = Arc::new(MemoryEventStore::default());
4336        let mut builder = VerbRegistryBuilder::new();
4337        builder.register(AlphaPack);
4338        builder.with_gate(Arc::new(ObligationGate));
4339        builder.with_event_store(store.clone());
4340        let reg = builder.build().expect("registry builds");
4341
4342        reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4343            .await
4344            .unwrap();
4345
4346        let page = store
4347            .query_events(
4348                EventFilter::default(),
4349                PageRequest {
4350                    limit: 10,
4351                    offset: 0,
4352                },
4353            )
4354            .await
4355            .unwrap();
4356        assert_eq!(page.items.len(), 1);
4357
4358        let ev = &page.items[0];
4359        assert_eq!(ev.outcome, EventOutcome::Success);
4360
4361        let data = &ev.payload;
4362
4363        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4364            .expect("Event.payload must deserialize to AuditEvent");
4365
4366        assert_eq!(audit.gate_impl, "ObligationGate");
4367        assert_eq!(
4368            audit.obligations.len(),
4369            1,
4370            "obligations must be preserved through EventStore"
4371        );
4372        match &audit.obligations[0] {
4373            Obligation::Audit { tag } => assert_eq!(tag, "billing.meter"),
4374            other => panic!("expected Audit obligation, got {other:?}"),
4375        }
4376    }
4377
4378    // ---- SQL-backed audit envelope round-trip ----
4379    //
4380    // The two tests above use MemoryEventStore (no serialization). This test
4381    // wires the production SqlEventStore via KhiveRuntime::memory() to verify
4382    // that the full AuditEvent envelope survives the SQL text→parse round-trip
4383    // (Event.data is stored as TEXT and parsed back on read).
4384
4385    #[tokio::test]
4386    async fn sql_backed_audit_envelope_round_trips_deny_reason_gate_impl_and_obligations() {
4387        #[derive(Debug)]
4388        struct SqlTestDenyGate;
4389        impl Gate for SqlTestDenyGate {
4390            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4391                Ok(GateDecision::deny("sql-path: write denied"))
4392            }
4393            fn impl_name(&self) -> &'static str {
4394                "SqlTestDenyGate"
4395            }
4396        }
4397
4398        // KhiveRuntime::memory() creates an in-memory SQLite pool (is_file_backed=false).
4399        // events_for_namespace ensures the events schema and returns a SqlEventStore
4400        // scoped to "test-ns". The pool is shared so reads and writes see the same data.
4401        let rt = KhiveRuntime::memory().expect("in-memory runtime");
4402        let test_tok = NamespaceToken::for_namespace(Namespace::parse("test-ns").unwrap());
4403        let sql_store = rt
4404            .events(&test_tok)
4405            .expect("events_for_namespace must succeed");
4406
4407        let mut builder = VerbRegistryBuilder::new();
4408        builder.register(AlphaPack);
4409        builder.with_gate(Arc::new(SqlTestDenyGate));
4410        builder.with_event_store(sql_store.clone());
4411        let reg = builder.build().expect("registry builds");
4412
4413        // Dispatch is denied — PermissionDenied returned.
4414        let err = reg
4415            .dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4416            .await
4417            .unwrap_err();
4418        assert!(
4419            matches!(err, RuntimeError::PermissionDenied { .. }),
4420            "expected PermissionDenied, got {err:?}"
4421        );
4422
4423        // Query via the same SqlEventStore — this is the SQL read path.
4424        let page = sql_store
4425            .query_events(
4426                EventFilter::default(),
4427                PageRequest {
4428                    limit: 10,
4429                    offset: 0,
4430                },
4431            )
4432            .await
4433            .unwrap();
4434        assert_eq!(
4435            page.items.len(),
4436            1,
4437            "one audit event must be persisted on deny through SqlEventStore"
4438        );
4439
4440        let ev = &page.items[0];
4441        assert_eq!(ev.outcome, EventOutcome::Denied);
4442
4443        // Event.payload must hold the full AuditEvent serialized as JSON text and
4444        // parsed back. If the SQL path was lossy, this deserialization would fail
4445        // or the field assertions below would fail.
4446        let data = &ev.payload;
4447
4448        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4449            .expect("Event.payload must deserialize to AuditEvent after SQL round-trip");
4450
4451        assert_eq!(
4452            audit.deny_reason.as_deref(),
4453            Some("sql-path: write denied"),
4454            "deny_reason must survive the SQL text round-trip"
4455        );
4456        assert_eq!(
4457            audit.gate_impl, "SqlTestDenyGate",
4458            "gate_impl must survive the SQL text round-trip"
4459        );
4460        assert_eq!(
4461            audit.decision,
4462            khive_gate::AuditDecision::Deny,
4463            "decision field must survive the SQL text round-trip"
4464        );
4465        // obligations is [] on a Deny gate (no obligations returned).
4466        // Verify the field is present and empty after SQL round-trip.
4467        assert!(
4468            audit.obligations.is_empty(),
4469            "obligations must be preserved as empty [] through SQL round-trip"
4470        );
4471    }
4472
4473    // ---- SQL-backed audit envelope: non-empty obligations survive round-trip ----
4474    //
4475    // Blind spot: the deny-path SQL test above only
4476    // asserts obligations == [], which passes even if the SQL path drops the
4477    // field entirely (AuditEvent.obligations has #[serde(default)]).
4478    //
4479    // This test installs an allow-path gate that returns a non-empty obligations
4480    // vec. After dispatch, the same SqlEventStore is queried and both layers are
4481    // checked:
4482    //   1. Raw Event.data["obligations"] is a non-empty JSON array.
4483    //   2. Deserialized AuditEvent.obligations[0] matches the expected variant.
4484    #[tokio::test]
4485    async fn sql_backed_audit_envelope_round_trips_non_empty_obligations() {
4486        use khive_gate::Obligation;
4487
4488        #[derive(Debug)]
4489        struct SqlTestAllowWithObligationGate;
4490        impl Gate for SqlTestAllowWithObligationGate {
4491            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4492                Ok(GateDecision::allow_with(vec![Obligation::Audit {
4493                    tag: "sql-path-billing.meter".into(),
4494                }]))
4495            }
4496            fn impl_name(&self) -> &'static str {
4497                "SqlTestAllowWithObligationGate"
4498            }
4499        }
4500
4501        let rt = KhiveRuntime::memory().expect("in-memory runtime");
4502        let test_tok = NamespaceToken::for_namespace(Namespace::parse("test-ns").unwrap());
4503        let sql_store = rt
4504            .events(&test_tok)
4505            .expect("events_for_namespace must succeed");
4506
4507        let mut builder = VerbRegistryBuilder::new();
4508        builder.register(AlphaPack);
4509        builder.with_gate(Arc::new(SqlTestAllowWithObligationGate));
4510        builder.with_event_store(sql_store.clone());
4511        let reg = builder.build().expect("registry builds");
4512
4513        // Dispatch succeeds — the gate allows with obligations.
4514        reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
4515            .await
4516            .expect("dispatch must succeed when gate allows");
4517
4518        // Query via the same SqlEventStore — this is the SQL read path.
4519        let page = sql_store
4520            .query_events(
4521                EventFilter::default(),
4522                PageRequest {
4523                    limit: 10,
4524                    offset: 0,
4525                },
4526            )
4527            .await
4528            .unwrap();
4529        assert_eq!(
4530            page.items.len(),
4531            1,
4532            "one audit event must be persisted on allow through SqlEventStore"
4533        );
4534
4535        let ev = &page.items[0];
4536        assert_eq!(ev.outcome, EventOutcome::Success);
4537
4538        let data = &ev.payload;
4539
4540        // Layer 1: raw JSON check — obligations must be a non-empty array in
4541        // the persisted TEXT. If the SQL path dropped the field, the default
4542        // #[serde(default)] would silently deserialize it to [], so we verify
4543        // the raw JSON before deserializing.
4544        let obligations_raw = data
4545            .get("obligations")
4546            .expect("Event.data JSON must contain 'obligations' key");
4547        let obligations_arr = obligations_raw
4548            .as_array()
4549            .expect("'obligations' must be a JSON array");
4550        assert!(
4551            !obligations_arr.is_empty(),
4552            "raw Event.data['obligations'] must be non-empty after SQL round-trip"
4553        );
4554
4555        // Layer 2: deserialized AuditEvent check — the obligation variant and
4556        // payload must survive the text round-trip faithfully.
4557        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4558            .expect("Event.data must deserialize to AuditEvent after SQL round-trip");
4559
4560        assert_eq!(
4561            audit.gate_impl, "SqlTestAllowWithObligationGate",
4562            "gate_impl must survive the SQL text round-trip"
4563        );
4564        assert_eq!(
4565            audit.decision,
4566            khive_gate::AuditDecision::Allow,
4567            "decision field must survive the SQL text round-trip"
4568        );
4569        assert_eq!(
4570            audit.obligations.len(),
4571            1,
4572            "obligations must be non-empty after SQL round-trip (not silently defaulted to [])"
4573        );
4574        match &audit.obligations[0] {
4575            Obligation::Audit { tag } => assert_eq!(
4576                tag, "sql-path-billing.meter",
4577                "Audit obligation tag must survive the SQL text round-trip"
4578            ),
4579            other => panic!("expected Audit obligation, got {other:?}"),
4580        }
4581    }
4582
4583    // ---- Audit payload shape for 'create' verb dispatch ----
4584    //
4585    // The previous audit tests verify the envelope shape for the 'list' verb.
4586    // This test dispatches 'create' (matching the create_note + annotates path)
4587    // and verifies that ev.verb, ev.outcome, and ev.data all round-trip correctly
4588    // through the EventStore. Ensures the wire shape is independent of which verb
4589    // triggers the gate check.
4590    #[tokio::test]
4591    async fn audit_event_payload_shape_for_create_verb() {
4592        let store = Arc::new(MemoryEventStore::default());
4593        let mut builder = VerbRegistryBuilder::new();
4594        builder.register(AlphaPack);
4595        builder.with_event_store(store.clone());
4596        builder.with_default_namespace("test-ns");
4597        let reg = builder.build().expect("registry builds");
4598
4599        // Dispatch 'create' — AlphaPack returns a stub value; what matters is
4600        // the EventStore entry emitted by the registry's gate-check path.
4601        reg.dispatch("create", serde_json::json!({"namespace": "test-ns"}))
4602            .await
4603            .unwrap();
4604
4605        let count = store.count_events(EventFilter::default()).await.unwrap();
4606        assert_eq!(count, 1, "exactly one audit event for one dispatch");
4607
4608        let page = store
4609            .query_events(
4610                EventFilter::default(),
4611                PageRequest {
4612                    limit: 10,
4613                    offset: 0,
4614                },
4615            )
4616            .await
4617            .unwrap();
4618        let ev = &page.items[0];
4619
4620        // Top-level Event fields.
4621        assert_eq!(ev.verb, "create", "ev.verb must be the dispatched verb");
4622        assert_eq!(
4623            ev.outcome,
4624            EventOutcome::Success,
4625            "ev.outcome must be Success on allow"
4626        );
4627        assert_eq!(
4628            ev.namespace, "test-ns",
4629            "ev.namespace must match the dispatch namespace"
4630        );
4631
4632        // ev.payload must hold the full AuditEvent envelope.
4633        let data = &ev.payload;
4634
4635        let audit: khive_gate::AuditEvent = serde_json::from_value(data.clone())
4636            .expect("ev.payload must deserialize to AuditEvent");
4637
4638        assert_eq!(
4639            audit.decision,
4640            khive_gate::AuditDecision::Allow,
4641            "AuditEvent.decision must be Allow"
4642        );
4643        assert_eq!(audit.verb, "create", "AuditEvent.verb must be 'create'");
4644        assert_eq!(
4645            audit.namespace, "test-ns",
4646            "AuditEvent.namespace must be preserved"
4647        );
4648        assert_eq!(
4649            audit.gate_impl, "AllowAllGate",
4650            "AuditEvent.gate_impl must name the gate implementation"
4651        );
4652        assert!(
4653            audit.deny_reason.is_none(),
4654            "AuditEvent.deny_reason must be None on Allow"
4655        );
4656        // Wire-shape check: obligations serializes as [] on AllowAllGate.
4657        let payload_json: serde_json::Value =
4658            serde_json::from_value(data.clone()).expect("data must be valid JSON");
4659        assert_eq!(
4660            payload_json["obligations"],
4661            serde_json::Value::Array(Vec::new()),
4662            "obligations must be [] on AllowAllGate"
4663        );
4664    }
4665
4666    // ---- ADR-103 Amendment 1: resource.cost_unit emission ----
4667
4668    /// Test pack whose `create` handler is a stub (mirrors `AlphaPack`) but
4669    /// overrides `registered_embedding_model_names` to a configurable set,
4670    /// exercising ADR-103 Amendment 1's `model_count` computation for
4671    /// singleton `create` at the dispatch audit-row emission seam.
4672    struct EmbeddingAwarePack {
4673        models: Vec<String>,
4674    }
4675
4676    impl khive_types::Pack for EmbeddingAwarePack {
4677        const NAME: &'static str = "embedding_aware";
4678        const NOTE_KINDS: &'static [&'static str] = &[];
4679        const ENTITY_KINDS: &'static [&'static str] = &["widget"];
4680        const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
4681            name: "create",
4682            description: "create a widget (embedding-aware stub)",
4683            visibility: Visibility::Verb,
4684            category: VerbCategory::Commissive,
4685            params: &[],
4686        }];
4687    }
4688
4689    #[async_trait]
4690    impl PackRuntime for EmbeddingAwarePack {
4691        fn name(&self) -> &str {
4692            Self::NAME
4693        }
4694        fn note_kinds(&self) -> &'static [&'static str] {
4695            Self::NOTE_KINDS
4696        }
4697        fn entity_kinds(&self) -> &'static [&'static str] {
4698            Self::ENTITY_KINDS
4699        }
4700        fn handlers(&self) -> &'static [HandlerDef] {
4701            Self::HANDLERS
4702        }
4703        fn registered_embedding_model_names(&self) -> Vec<String> {
4704            self.models.clone()
4705        }
4706        async fn dispatch(
4707            &self,
4708            verb: &str,
4709            _params: Value,
4710            _registry: &VerbRegistry,
4711            _token: &NamespaceToken,
4712        ) -> Result<Value, RuntimeError> {
4713            Ok(serde_json::json!({ "pack": "embedding_aware", "verb": verb }))
4714        }
4715    }
4716
4717    /// Test pack whose one verb, `probe`, always fails — used to drive the
4718    /// general (non-link) deferred-audit Err arm without a real backend.
4719    struct FailingProbePack;
4720
4721    impl khive_types::Pack for FailingProbePack {
4722        const NAME: &'static str = "failing_probe";
4723        const NOTE_KINDS: &'static [&'static str] = &[];
4724        const ENTITY_KINDS: &'static [&'static str] = &[];
4725        const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
4726            name: "probe",
4727            description: "always fails",
4728            visibility: Visibility::Verb,
4729            category: VerbCategory::Assertive,
4730            params: &[],
4731        }];
4732    }
4733
4734    #[async_trait]
4735    impl PackRuntime for FailingProbePack {
4736        fn name(&self) -> &str {
4737            Self::NAME
4738        }
4739        fn note_kinds(&self) -> &'static [&'static str] {
4740            Self::NOTE_KINDS
4741        }
4742        fn entity_kinds(&self) -> &'static [&'static str] {
4743            Self::ENTITY_KINDS
4744        }
4745        fn handlers(&self) -> &'static [HandlerDef] {
4746            Self::HANDLERS
4747        }
4748        async fn dispatch(
4749            &self,
4750            _verb: &str,
4751            _params: Value,
4752            _registry: &VerbRegistry,
4753            _token: &NamespaceToken,
4754        ) -> Result<Value, RuntimeError> {
4755            Err(RuntimeError::InvalidInput("boom".into()))
4756        }
4757    }
4758
4759    #[tokio::test]
4760    async fn resource_cost_unit_present_on_non_embedding_successful_dispatch() {
4761        let store = Arc::new(MemoryEventStore::default());
4762        let mut builder = VerbRegistryBuilder::new();
4763        builder.register(AlphaPack);
4764        builder.with_event_store(store.clone());
4765        let reg = builder.build().expect("registry builds");
4766
4767        reg.dispatch("list", serde_json::json!({})).await.unwrap();
4768
4769        let page = store
4770            .query_events(
4771                EventFilter::default(),
4772                PageRequest {
4773                    limit: 10,
4774                    offset: 0,
4775                },
4776            )
4777            .await
4778            .unwrap();
4779        assert_eq!(page.items.len(), 1);
4780        assert_eq!(
4781            page.items[0].payload["resource"],
4782            serde_json::json!({"work_class": "interactive", "cost_unit": 1}),
4783            "non-embedding-bearing verb's resource.cost_unit must be base_weight(verb) alone"
4784        );
4785    }
4786
4787    #[tokio::test]
4788    async fn resource_cost_unit_scales_with_registered_model_count_for_create() {
4789        let store = Arc::new(MemoryEventStore::default());
4790        let mut builder = VerbRegistryBuilder::new();
4791        builder.register(EmbeddingAwarePack {
4792            models: vec!["all-minilm-l6-v2".into(), "paraphrase".into()],
4793        });
4794        builder.with_event_store(store.clone());
4795        let reg = builder.build().expect("registry builds");
4796
4797        reg.dispatch("create", serde_json::json!({"kind": "widget"}))
4798            .await
4799            .unwrap();
4800
4801        let page = store
4802            .query_events(
4803                EventFilter::default(),
4804                PageRequest {
4805                    limit: 10,
4806                    offset: 0,
4807                },
4808            )
4809            .await
4810            .unwrap();
4811        // base_weight(1) + per_item_weight(1) * item_count(1) * model_count(2)
4812        assert_eq!(
4813            page.items[0].payload["resource"],
4814            serde_json::json!({"work_class": "interactive", "cost_unit": 3}),
4815        );
4816    }
4817
4818    #[tokio::test]
4819    async fn resource_cost_unit_zero_registered_models_is_base_weight_only() {
4820        let store = Arc::new(MemoryEventStore::default());
4821        let mut builder = VerbRegistryBuilder::new();
4822        builder.register(EmbeddingAwarePack { models: vec![] });
4823        builder.with_event_store(store.clone());
4824        let reg = builder.build().expect("registry builds");
4825
4826        reg.dispatch("create", serde_json::json!({"kind": "widget"}))
4827            .await
4828            .unwrap();
4829
4830        let page = store
4831            .query_events(
4832                EventFilter::default(),
4833                PageRequest {
4834                    limit: 10,
4835                    offset: 0,
4836                },
4837            )
4838            .await
4839            .unwrap();
4840        assert_eq!(
4841            page.items[0].payload["resource"]["cost_unit"], 1,
4842            "zero registered embedding models must vanish the term, not error or omit"
4843        );
4844    }
4845
4846    #[tokio::test]
4847    async fn resource_work_class_present_cost_unit_absent_when_dispatch_returns_error() {
4848        let store = Arc::new(MemoryEventStore::default());
4849        let mut builder = VerbRegistryBuilder::new();
4850        builder.register(FailingProbePack);
4851        builder.with_event_store(store.clone());
4852        let reg = builder.build().expect("registry builds");
4853
4854        let err = reg
4855            .dispatch("probe", serde_json::json!({}))
4856            .await
4857            .unwrap_err();
4858        assert!(matches!(err, RuntimeError::InvalidInput(_)));
4859
4860        let page = store
4861            .query_events(
4862                EventFilter::default(),
4863                PageRequest {
4864                    limit: 10,
4865                    offset: 0,
4866                },
4867            )
4868            .await
4869            .unwrap();
4870        assert_eq!(page.items.len(), 1);
4871        assert_eq!(page.items[0].outcome, EventOutcome::Error);
4872        // ADR-103 Decision (a): work_class is stamped on EVERY event, denial
4873        // and error included -- only Amendment 1's cost_unit field is scoped
4874        // to a successful dispatch. An errored dispatch keeps
4875        // resource.work_class and omits only resource.cost_unit, never 0.
4876        assert_eq!(
4877            page.items[0].payload["resource"],
4878            serde_json::json!({"work_class": "interactive"}),
4879            "resource must carry work_class with cost_unit OMITTED (never 0) on an \
4880             errored dispatch: {:?}",
4881            page.items[0].payload
4882        );
4883    }
4884
4885    #[tokio::test]
4886    async fn resource_work_class_present_cost_unit_absent_when_no_pack_owns_the_verb() {
4887        let store = Arc::new(MemoryEventStore::default());
4888        let mut builder = VerbRegistryBuilder::new();
4889        builder.register(AlphaPack);
4890        builder.with_event_store(store.clone());
4891        let reg = builder.build().expect("registry builds");
4892
4893        let _ = reg
4894            .dispatch("no_such_verb_resource_test", serde_json::json!({}))
4895            .await;
4896
4897        let page = store
4898            .query_events(
4899                EventFilter::default(),
4900                PageRequest {
4901                    limit: 10,
4902                    offset: 0,
4903                },
4904            )
4905            .await
4906            .unwrap();
4907        assert_eq!(page.items.len(), 1);
4908        assert_eq!(
4909            page.items[0].payload["resource"],
4910            serde_json::json!({"work_class": "interactive"})
4911        );
4912    }
4913
4914    #[tokio::test]
4915    async fn resource_work_class_present_cost_unit_absent_on_denied_dispatch() {
4916        #[derive(Debug)]
4917        struct AlwaysDenyGate;
4918        impl Gate for AlwaysDenyGate {
4919            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
4920                Ok(GateDecision::deny("test: always deny"))
4921            }
4922        }
4923        let store = Arc::new(MemoryEventStore::default());
4924        let mut builder = VerbRegistryBuilder::new();
4925        builder.register(AlphaPack);
4926        builder.with_gate(Arc::new(AlwaysDenyGate));
4927        builder.with_event_store(store.clone());
4928        let reg = builder.build().expect("registry builds");
4929
4930        let _ = reg.dispatch("list", serde_json::json!({})).await;
4931
4932        let page = store
4933            .query_events(
4934                EventFilter::default(),
4935                PageRequest {
4936                    limit: 10,
4937                    offset: 0,
4938                },
4939            )
4940            .await
4941            .unwrap();
4942        assert_eq!(page.items.len(), 1);
4943        assert_eq!(page.items[0].outcome, EventOutcome::Denied);
4944        assert_eq!(
4945            page.items[0].payload["resource"],
4946            serde_json::json!({"work_class": "interactive"})
4947        );
4948    }
4949
4950    #[tokio::test]
4951    async fn resource_cost_unit_present_on_link_singleton_success() {
4952        let store = Arc::new(MemoryEventStore::default());
4953        let edge_id = uuid::Uuid::new_v4();
4954        let source_id = uuid::Uuid::new_v4();
4955        let target_id = uuid::Uuid::new_v4();
4956        let edge_json = serde_json::json!({
4957            "id": edge_id,
4958            "namespace": "local",
4959            "source_id": source_id,
4960            "target_id": target_id,
4961            "relation": "depends_on",
4962            "weight": 1.0,
4963        });
4964        let mut builder = VerbRegistryBuilder::new();
4965        builder.register(LinkResultPack::ok(edge_json));
4966        builder.with_event_store(store.clone());
4967        let reg = builder.build().expect("registry builds");
4968
4969        reg.dispatch(
4970            "link",
4971            serde_json::json!({
4972                "source_id": source_id,
4973                "target_id": target_id,
4974                "relation": "depends_on",
4975            }),
4976        )
4977        .await
4978        .unwrap();
4979
4980        let page = store
4981            .query_events(
4982                EventFilter::default(),
4983                PageRequest {
4984                    limit: 10,
4985                    offset: 0,
4986                },
4987            )
4988            .await
4989            .unwrap();
4990        assert_eq!(
4991            page.items[0].payload["resource"],
4992            serde_json::json!({"work_class": "interactive", "cost_unit": 1}),
4993            "link has no embedding-bearing path -> base_weight(link) alone, even on the v2-enriched singleton path"
4994        );
4995    }
4996
4997    #[tokio::test]
4998    async fn resource_work_class_present_cost_unit_absent_on_link_dispatch_failure() {
4999        let store = Arc::new(MemoryEventStore::default());
5000        let mut builder = VerbRegistryBuilder::new();
5001        builder.register(LinkResultPack::err("target endpoint not found"));
5002        builder.with_event_store(store.clone());
5003        let reg = builder.build().expect("registry builds");
5004
5005        let _ = reg
5006            .dispatch(
5007                "link",
5008                serde_json::json!({
5009                    "source_id": "note:alpha",
5010                    "target_id": "note:missing",
5011                    "relation": "depends_on",
5012                }),
5013            )
5014            .await;
5015
5016        let page = store
5017            .query_events(
5018                EventFilter::default(),
5019                PageRequest {
5020                    limit: 10,
5021                    offset: 0,
5022                },
5023            )
5024            .await
5025            .unwrap();
5026        assert_eq!(page.items.len(), 1);
5027        assert_eq!(
5028            page.items[0].payload["resource"],
5029            serde_json::json!({"work_class": "interactive"})
5030        );
5031    }
5032
5033    // Registry audit event must carry target_id when dispatch params include it.
5034    #[tokio::test]
5035    async fn audit_event_threads_target_id_from_dispatch_args() {
5036        let store = Arc::new(MemoryEventStore::default());
5037        let target = uuid::Uuid::new_v4();
5038        let mut builder = VerbRegistryBuilder::new();
5039        builder.register(AlphaPack);
5040        builder.with_event_store(store.clone());
5041        builder.with_default_namespace("test-ns");
5042        let reg = builder.build().expect("registry builds");
5043
5044        reg.dispatch(
5045            "create",
5046            serde_json::json!({"namespace": "test-ns", "target_id": target}),
5047        )
5048        .await
5049        .unwrap();
5050
5051        let page = store
5052            .query_events(
5053                EventFilter::default(),
5054                PageRequest {
5055                    offset: 0,
5056                    limit: 10,
5057                },
5058            )
5059            .await
5060            .unwrap();
5061        assert_eq!(
5062            page.items[0].target_id,
5063            Some(target),
5064            "#282: audit event must carry target_id from dispatch params"
5065        );
5066    }
5067
5068    // ---- Link-verb audit enrichment ----
5069
5070    /// Test pack exposing a single `link` verb whose one-shot result is
5071    /// configured up front — lets tests drive both the success and failure
5072    /// legs of the deferred link-audit path without a real KG backend.
5073    struct LinkResultPack {
5074        result: std::sync::Mutex<Option<Result<Value, RuntimeError>>>,
5075    }
5076
5077    impl LinkResultPack {
5078        fn ok(value: Value) -> Self {
5079            Self {
5080                result: std::sync::Mutex::new(Some(Ok(value))),
5081            }
5082        }
5083        fn err(message: &str) -> Self {
5084            Self {
5085                result: std::sync::Mutex::new(Some(Err(RuntimeError::InvalidInput(
5086                    message.to_string(),
5087                )))),
5088            }
5089        }
5090    }
5091
5092    impl khive_types::Pack for LinkResultPack {
5093        const NAME: &'static str = "kg";
5094        const NOTE_KINDS: &'static [&'static str] = &[];
5095        const ENTITY_KINDS: &'static [&'static str] = &[];
5096        const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
5097            name: "link",
5098            description: "test link handler",
5099            visibility: Visibility::Verb,
5100            category: VerbCategory::Commissive,
5101            params: &[],
5102        }];
5103    }
5104
5105    #[async_trait]
5106    impl PackRuntime for LinkResultPack {
5107        fn name(&self) -> &str {
5108            Self::NAME
5109        }
5110        fn note_kinds(&self) -> &'static [&'static str] {
5111            Self::NOTE_KINDS
5112        }
5113        fn entity_kinds(&self) -> &'static [&'static str] {
5114            Self::ENTITY_KINDS
5115        }
5116        fn handlers(&self) -> &'static [HandlerDef] {
5117            Self::HANDLERS
5118        }
5119        async fn dispatch(
5120            &self,
5121            _verb: &str,
5122            _params: Value,
5123            _registry: &VerbRegistry,
5124            _token: &NamespaceToken,
5125        ) -> Result<Value, RuntimeError> {
5126            self.result
5127                .lock()
5128                .unwrap()
5129                .take()
5130                .expect("LinkResultPack dispatch called more than once in a test")
5131        }
5132    }
5133
5134    #[tokio::test]
5135    async fn link_audit_enriches_successful_singleton_with_edge_v2() {
5136        let store = Arc::new(MemoryEventStore::default());
5137        let edge_id = uuid::Uuid::new_v4();
5138        let source_id = uuid::Uuid::new_v4();
5139        let target_id = uuid::Uuid::new_v4();
5140        let edge_json = serde_json::json!({
5141            "id": edge_id,
5142            "namespace": "local",
5143            "source_id": source_id,
5144            "target_id": target_id,
5145            "relation": "depends_on",
5146            "weight": 1.0,
5147        });
5148        let mut builder = VerbRegistryBuilder::new();
5149        builder.register(LinkResultPack::ok(edge_json));
5150        builder.with_event_store(store.clone());
5151        builder.with_default_namespace("test-ns");
5152        let reg = builder.build().expect("registry builds");
5153
5154        reg.dispatch(
5155            "link",
5156            serde_json::json!({
5157                "source_id": source_id,
5158                "target_id": target_id,
5159                "relation": "depends_on",
5160            }),
5161        )
5162        .await
5163        .unwrap();
5164
5165        let count = store.count_events(EventFilter::default()).await.unwrap();
5166        assert_eq!(
5167            count, 1,
5168            "exactly one deferred audit row must be persisted for a successful singleton link"
5169        );
5170        let page = store
5171            .query_events(
5172                EventFilter::default(),
5173                PageRequest {
5174                    limit: 10,
5175                    offset: 0,
5176                },
5177            )
5178            .await
5179            .unwrap();
5180        let ev = &page.items[0];
5181        assert_eq!(ev.verb, "link");
5182        assert_eq!(ev.outcome, EventOutcome::Success);
5183        assert_eq!(
5184            ev.payload_schema_version, 2,
5185            "successful singleton link uses audit schema v2"
5186        );
5187        assert_eq!(
5188            ev.target_id,
5189            Some(edge_id),
5190            "target_id must be the created/resolved edge id, not a raw caller arg"
5191        );
5192        assert_eq!(ev.payload["edge_id"], serde_json::json!(edge_id));
5193        assert_eq!(ev.payload["source_id"], serde_json::json!(source_id));
5194        assert_eq!(ev.payload["target_id"], serde_json::json!(target_id));
5195        assert_eq!(ev.payload["relation"], "depends_on");
5196        assert_eq!(ev.payload["weight"], 1.0);
5197        // v1 AuditEvent fields remain present via #[serde(flatten)].
5198        assert_eq!(ev.payload["verb"], "link");
5199        assert_eq!(ev.payload["decision"], "allow");
5200        assert!(ev.payload.get("gate_impl").is_some());
5201    }
5202
5203    #[tokio::test]
5204    async fn link_audit_falls_back_to_v1_when_dispatch_fails() {
5205        let store = Arc::new(MemoryEventStore::default());
5206        let mut builder = VerbRegistryBuilder::new();
5207        builder.register(LinkResultPack::err("target endpoint not found"));
5208        builder.with_event_store(store.clone());
5209        builder.with_default_namespace("test-ns");
5210        let reg = builder.build().expect("registry builds");
5211
5212        let err = reg
5213            .dispatch(
5214                "link",
5215                serde_json::json!({
5216                    "source_id": "note:alpha",
5217                    "target_id": "note:missing",
5218                    "relation": "depends_on",
5219                }),
5220            )
5221            .await
5222            .unwrap_err();
5223        assert!(
5224            matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("not found")),
5225            "the original dispatch error must be returned unchanged"
5226        );
5227
5228        let page = store
5229            .query_events(
5230                EventFilter::default(),
5231                PageRequest {
5232                    limit: 10,
5233                    offset: 0,
5234                },
5235            )
5236            .await
5237            .unwrap();
5238        assert_eq!(
5239            page.items.len(),
5240            1,
5241            "a v1 fallback audit row must still be persisted on dispatch failure"
5242        );
5243        let ev = &page.items[0];
5244        assert_eq!(
5245            ev.payload_schema_version, 1,
5246            "failed link keeps the v1 audit shape"
5247        );
5248        // The persisted outcome must reflect the dispatch result (Err →
5249        // Error), not be hardcoded to Success from the gate's Allow decision.
5250        assert_eq!(
5251            ev.outcome,
5252            EventOutcome::Error,
5253            "outcome reflects the dispatch result (Err), not the gate decision (Allow)"
5254        );
5255        assert!(
5256            ev.duration_us >= 0,
5257            "duration_us must still be populated (measured, not the Event::new \
5258             default sentinel) on a failed dispatch"
5259        );
5260        assert!(
5261            ev.target_id.is_none(),
5262            "non-UUID caller-supplied ids do not spuriously populate target_id"
5263        );
5264        assert!(
5265            ev.payload.get("edge_id").is_none(),
5266            "v1 fallback must not carry edge enrichment fields"
5267        );
5268        let _: khive_gate::AuditEvent = serde_json::from_value(ev.payload.clone())
5269            .expect("v1 fallback payload must deserialize as AuditEvent");
5270    }
5271
5272    #[tokio::test]
5273    async fn link_audit_falls_back_to_v1_when_result_missing_edge_fields() {
5274        let store = Arc::new(MemoryEventStore::default());
5275        let target_arg = uuid::Uuid::new_v4();
5276        let mut builder = VerbRegistryBuilder::new();
5277        builder.register(LinkResultPack::ok(serde_json::json!({"ok": true})));
5278        builder.with_event_store(store.clone());
5279        builder.with_default_namespace("test-ns");
5280        let reg = builder.build().expect("registry builds");
5281
5282        reg.dispatch(
5283            "link",
5284            serde_json::json!({
5285                "source_id": uuid::Uuid::new_v4(),
5286                "target_id": target_arg,
5287                "relation": "depends_on",
5288            }),
5289        )
5290        .await
5291        .unwrap();
5292
5293        let page = store
5294            .query_events(
5295                EventFilter::default(),
5296                PageRequest {
5297                    limit: 10,
5298                    offset: 0,
5299                },
5300            )
5301            .await
5302            .unwrap();
5303        assert_eq!(page.items.len(), 1);
5304        let ev = &page.items[0];
5305        assert_eq!(
5306            ev.payload_schema_version, 1,
5307            "an unparsable success result falls back to v1 rather than dropping the audit row"
5308        );
5309        assert_eq!(ev.outcome, EventOutcome::Success);
5310        assert_eq!(
5311            ev.target_id,
5312            Some(target_arg),
5313            "v1 fallback still extracts target_id from the raw dispatch args"
5314        );
5315        assert!(ev.payload.get("edge_id").is_none());
5316    }
5317
5318    #[tokio::test]
5319    async fn link_audit_bulk_links_get_no_enrichment() {
5320        let store = Arc::new(MemoryEventStore::default());
5321        let mut builder = VerbRegistryBuilder::new();
5322        builder.register(LinkResultPack::ok(serde_json::json!({
5323            "attempted": 2, "created": 2, "skipped": 0, "failed": 0
5324        })));
5325        builder.with_event_store(store.clone());
5326        builder.with_default_namespace("test-ns");
5327        let reg = builder.build().expect("registry builds");
5328
5329        reg.dispatch(
5330            "link",
5331            serde_json::json!({
5332                "links": [
5333                    {"source_id": "a", "target_id": "b", "relation": "depends_on"},
5334                    {"source_id": "c", "target_id": "d", "relation": "depends_on"},
5335                ],
5336            }),
5337        )
5338        .await
5339        .unwrap();
5340
5341        let count = store.count_events(EventFilter::default()).await.unwrap();
5342        assert_eq!(
5343            count, 1,
5344            "bulk `links` gets exactly one v1 audit row (deferred until dispatch \
5345             resolves like every other Allow-outcome row since ADR-103 Stage 1, \
5346             but never v2-enriched — enrichment is singleton-`link`-only)"
5347        );
5348        let page = store
5349            .query_events(
5350                EventFilter::default(),
5351                PageRequest {
5352                    limit: 10,
5353                    offset: 0,
5354                },
5355            )
5356            .await
5357            .unwrap();
5358        let ev = &page.items[0];
5359        assert_eq!(
5360            ev.payload_schema_version, 1,
5361            "bulk link mode is out of scope for #676's events.target_id enrichment"
5362        );
5363        assert!(ev.target_id.is_none());
5364    }
5365
5366    #[test]
5367    fn link_audit_success_from_result_extracts_edge_fields() {
5368        let gate_req = GateRequest::new(
5369            ActorRef::anonymous(),
5370            Namespace::local(),
5371            "link",
5372            serde_json::json!({}),
5373        );
5374        let decision = GateDecision::Allow {
5375            obligations: vec![],
5376        };
5377        let audit = AuditEvent::from_check(&gate_req, &decision, "AllowAllGate");
5378
5379        let edge_id = uuid::Uuid::new_v4();
5380        let source_id = uuid::Uuid::new_v4();
5381        let target_id = uuid::Uuid::new_v4();
5382        let result = serde_json::json!({
5383            "id": edge_id,
5384            "source_id": source_id,
5385            "target_id": target_id,
5386            "relation": "depends_on",
5387            "weight": 0.5,
5388        });
5389
5390        let (returned_id, payload) = link_audit_success_from_result(audit, &result)
5391            .expect("well-formed edge JSON must produce an enriched payload");
5392        assert_eq!(returned_id, edge_id);
5393        assert_eq!(payload["edge_id"], serde_json::json!(edge_id));
5394        assert_eq!(payload["relation"], "depends_on");
5395        assert_eq!(payload["weight"], 0.5);
5396        assert_eq!(
5397            payload["verb"], "link",
5398            "v1 AuditEvent fields must flatten into the v2 payload"
5399        );
5400    }
5401
5402    #[test]
5403    fn link_audit_success_from_result_rejects_incomplete_or_malformed_result() {
5404        let gate_req = GateRequest::new(
5405            ActorRef::anonymous(),
5406            Namespace::local(),
5407            "link",
5408            serde_json::json!({}),
5409        );
5410        let decision = GateDecision::Allow {
5411            obligations: vec![],
5412        };
5413        let audit = AuditEvent::from_check(&gate_req, &decision, "AllowAllGate");
5414
5415        assert!(
5416            link_audit_success_from_result(
5417                audit.clone(),
5418                &serde_json::json!({"id": uuid::Uuid::new_v4()}),
5419            )
5420            .is_none(),
5421            "missing source_id/target_id/relation/weight must not enrich"
5422        );
5423        assert!(
5424            link_audit_success_from_result(audit, &serde_json::json!({"id": "not-a-uuid"}))
5425                .is_none(),
5426            "a non-UUID id must not enrich"
5427        );
5428    }
5429
5430    // ---- khive#948: request_id survives to the persisted audit event ----
5431    //
5432    // The pure `resource_payload`/`base_resource_payload` helpers are unit
5433    // tested in `cost_unit.rs`; these tests prove the id actually reaches
5434    // `resource.request_id` on a persisted `Event` through every one of
5435    // `dispatch_with_identity`'s four audit-append sites (denied, ordinary
5436    // success/error, singleton-link v2 success and its v1 fallback, and the
5437    // unknown-verb error path), plus the "no id supplied" omission case.
5438
5439    async fn first_event(store: &Arc<MemoryEventStore>) -> Event {
5440        let page = store
5441            .query_events(
5442                EventFilter::default(),
5443                PageRequest {
5444                    limit: 10,
5445                    offset: 0,
5446                },
5447            )
5448            .await
5449            .unwrap();
5450        assert_eq!(
5451            page.items.len(),
5452            1,
5453            "expected exactly one persisted audit event"
5454        );
5455        page.items[0].clone()
5456    }
5457
5458    #[tokio::test]
5459    async fn dispatch_with_identity_stamps_request_id_on_success() {
5460        let store = Arc::new(MemoryEventStore::default());
5461        let mut builder = VerbRegistryBuilder::new();
5462        builder.register(AlphaPack);
5463        builder.with_event_store(store.clone());
5464        let reg = builder.build().expect("registry builds");
5465
5466        reg.dispatch_with_identity(
5467            "list",
5468            serde_json::json!({"namespace": "test-ns"}),
5469            Some(RequestIdentity {
5470                request_id: Some(101),
5471                ..Default::default()
5472            }),
5473        )
5474        .await
5475        .unwrap();
5476
5477        let ev = first_event(&store).await;
5478        assert_eq!(ev.outcome, EventOutcome::Success);
5479        assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(101));
5480    }
5481
5482    #[tokio::test]
5483    async fn dispatch_with_identity_stamps_request_id_on_dispatch_error() {
5484        let store = Arc::new(MemoryEventStore::default());
5485        let mut builder = VerbRegistryBuilder::new();
5486        builder.register(FailingProbePack);
5487        builder.with_event_store(store.clone());
5488        let reg = builder.build().expect("registry builds");
5489
5490        let err = reg
5491            .dispatch_with_identity(
5492                "probe",
5493                serde_json::json!({"namespace": "test-ns"}),
5494                Some(RequestIdentity {
5495                    request_id: Some(102),
5496                    ..Default::default()
5497                }),
5498            )
5499            .await
5500            .unwrap_err();
5501        assert!(matches!(err, RuntimeError::InvalidInput(_)));
5502
5503        let ev = first_event(&store).await;
5504        assert_eq!(ev.outcome, EventOutcome::Error);
5505        assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(102));
5506    }
5507
5508    #[tokio::test]
5509    async fn dispatch_with_identity_stamps_request_id_on_denied() {
5510        #[derive(Debug)]
5511        struct AlwaysDenyGate;
5512        impl Gate for AlwaysDenyGate {
5513            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
5514                Ok(GateDecision::deny("denied by test"))
5515            }
5516        }
5517
5518        let store = Arc::new(MemoryEventStore::default());
5519        let mut builder = VerbRegistryBuilder::new();
5520        builder.register(AlphaPack);
5521        builder.with_gate(Arc::new(AlwaysDenyGate));
5522        builder.with_event_store(store.clone());
5523        let reg = builder.build().expect("registry builds");
5524
5525        let err = reg
5526            .dispatch_with_identity(
5527                "list",
5528                serde_json::json!({"namespace": "test-ns"}),
5529                Some(RequestIdentity {
5530                    request_id: Some(103),
5531                    ..Default::default()
5532                }),
5533            )
5534            .await
5535            .unwrap_err();
5536        assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
5537
5538        let ev = first_event(&store).await;
5539        assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(103));
5540    }
5541
5542    #[tokio::test]
5543    async fn dispatch_with_identity_stamps_request_id_on_link_v2_success() {
5544        let store = Arc::new(MemoryEventStore::default());
5545        let edge_id = uuid::Uuid::new_v4();
5546        let source_id = uuid::Uuid::new_v4();
5547        let target_id = uuid::Uuid::new_v4();
5548        let edge_json = serde_json::json!({
5549            "id": edge_id,
5550            "namespace": "local",
5551            "source_id": source_id,
5552            "target_id": target_id,
5553            "relation": "depends_on",
5554            "weight": 1.0,
5555        });
5556        let mut builder = VerbRegistryBuilder::new();
5557        builder.register(LinkResultPack::ok(edge_json));
5558        builder.with_event_store(store.clone());
5559        builder.with_default_namespace("test-ns");
5560        let reg = builder.build().expect("registry builds");
5561
5562        reg.dispatch_with_identity(
5563            "link",
5564            serde_json::json!({
5565                "source_id": source_id,
5566                "target_id": target_id,
5567                "relation": "depends_on",
5568            }),
5569            Some(RequestIdentity {
5570                namespace: "test-ns".to_string(),
5571                request_id: Some(104),
5572                ..Default::default()
5573            }),
5574        )
5575        .await
5576        .unwrap();
5577
5578        let ev = first_event(&store).await;
5579        assert_eq!(
5580            ev.payload_schema_version, 2,
5581            "successful singleton link uses audit schema v2"
5582        );
5583        assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(104));
5584    }
5585
5586    #[tokio::test]
5587    async fn dispatch_with_identity_stamps_request_id_on_link_v1_fallback() {
5588        let store = Arc::new(MemoryEventStore::default());
5589        let mut builder = VerbRegistryBuilder::new();
5590        builder.register(LinkResultPack::err("target endpoint not found"));
5591        builder.with_event_store(store.clone());
5592        builder.with_default_namespace("test-ns");
5593        let reg = builder.build().expect("registry builds");
5594
5595        let err = reg
5596            .dispatch_with_identity(
5597                "link",
5598                serde_json::json!({
5599                    "source_id": "note:alpha",
5600                    "target_id": "note:missing",
5601                    "relation": "depends_on",
5602                }),
5603                Some(RequestIdentity {
5604                    namespace: "test-ns".to_string(),
5605                    request_id: Some(105),
5606                    ..Default::default()
5607                }),
5608            )
5609            .await
5610            .unwrap_err();
5611        assert!(matches!(err, RuntimeError::InvalidInput(_)));
5612
5613        let ev = first_event(&store).await;
5614        assert_eq!(
5615            ev.payload_schema_version, 1,
5616            "failed link keeps the v1 audit shape"
5617        );
5618        assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(105));
5619    }
5620
5621    #[tokio::test]
5622    async fn dispatch_with_identity_stamps_request_id_on_unknown_verb() {
5623        let store = Arc::new(MemoryEventStore::default());
5624        let mut builder = VerbRegistryBuilder::new();
5625        builder.register(AlphaPack);
5626        builder.with_event_store(store.clone());
5627        let reg = builder.build().expect("registry builds");
5628
5629        let err = reg
5630            .dispatch_with_identity(
5631                "no_such_verb",
5632                serde_json::json!({}),
5633                Some(RequestIdentity {
5634                    namespace: Namespace::local().as_str().to_string(),
5635                    request_id: Some(106),
5636                    ..Default::default()
5637                }),
5638            )
5639            .await
5640            .unwrap_err();
5641        assert!(matches!(err, RuntimeError::InvalidInput(_)));
5642
5643        let ev = first_event(&store).await;
5644        assert_eq!(ev.outcome, EventOutcome::Error);
5645        assert_eq!(ev.payload["resource"]["request_id"], serde_json::json!(106));
5646    }
5647
5648    #[tokio::test]
5649    async fn dispatch_with_identity_omits_request_id_key_when_absent() {
5650        let store = Arc::new(MemoryEventStore::default());
5651        let mut builder = VerbRegistryBuilder::new();
5652        builder.register(AlphaPack);
5653        builder.with_event_store(store.clone());
5654        let reg = builder.build().expect("registry builds");
5655
5656        // No identity at all — the pre-#948 call shape.
5657        reg.dispatch("list", serde_json::json!({"namespace": "test-ns"}))
5658            .await
5659            .unwrap();
5660
5661        let ev = first_event(&store).await;
5662        let resource = ev.payload["resource"]
5663            .as_object()
5664            .expect("resource must be an object");
5665        assert!(
5666            !resource.contains_key("request_id"),
5667            "request_id key must be entirely absent when no id is supplied, \
5668             not present as null or 0: got {resource:?}"
5669        );
5670    }
5671}
5672
5673// ---- Inter-pack dependency checking ----
5674
5675#[cfg(test)]
5676mod dep_tests {
5677    use super::*;
5678    use async_trait::async_trait;
5679    use khive_types::Pack;
5680    use serde_json::Value;
5681
5682    struct KgDepPack;
5683    struct MemoryDepPack;
5684    struct ADepPack;
5685    struct BDepPack;
5686
5687    impl Pack for KgDepPack {
5688        const NAME: &'static str = "kg_dep";
5689        const NOTE_KINDS: &'static [&'static str] = &["observation"];
5690        const ENTITY_KINDS: &'static [&'static str] = &["concept"];
5691        const HANDLERS: &'static [HandlerDef] = &[];
5692    }
5693
5694    impl Pack for MemoryDepPack {
5695        const NAME: &'static str = "memory_dep";
5696        const NOTE_KINDS: &'static [&'static str] = &["memory"];
5697        const ENTITY_KINDS: &'static [&'static str] = &[];
5698        const HANDLERS: &'static [HandlerDef] = &[];
5699        const REQUIRES: &'static [&'static str] = &["kg_dep"];
5700    }
5701
5702    impl Pack for ADepPack {
5703        const NAME: &'static str = "pack_a";
5704        const NOTE_KINDS: &'static [&'static str] = &[];
5705        const ENTITY_KINDS: &'static [&'static str] = &[];
5706        const HANDLERS: &'static [HandlerDef] = &[];
5707        const REQUIRES: &'static [&'static str] = &["pack_b"];
5708    }
5709
5710    impl Pack for BDepPack {
5711        const NAME: &'static str = "pack_b";
5712        const NOTE_KINDS: &'static [&'static str] = &[];
5713        const ENTITY_KINDS: &'static [&'static str] = &[];
5714        const HANDLERS: &'static [HandlerDef] = &[];
5715        const REQUIRES: &'static [&'static str] = &["pack_a"];
5716    }
5717
5718    #[async_trait]
5719    impl PackRuntime for KgDepPack {
5720        fn name(&self) -> &str {
5721            Self::NAME
5722        }
5723        fn note_kinds(&self) -> &'static [&'static str] {
5724            Self::NOTE_KINDS
5725        }
5726        fn entity_kinds(&self) -> &'static [&'static str] {
5727            Self::ENTITY_KINDS
5728        }
5729        fn handlers(&self) -> &'static [HandlerDef] {
5730            Self::HANDLERS
5731        }
5732        async fn dispatch(
5733            &self,
5734            verb: &str,
5735            _: Value,
5736            _: &VerbRegistry,
5737            _: &NamespaceToken,
5738        ) -> Result<Value, RuntimeError> {
5739            Err(RuntimeError::InvalidInput(format!(
5740                "KgDepPack has no verbs: {verb}"
5741            )))
5742        }
5743    }
5744
5745    #[async_trait]
5746    impl PackRuntime for MemoryDepPack {
5747        fn name(&self) -> &str {
5748            Self::NAME
5749        }
5750        fn note_kinds(&self) -> &'static [&'static str] {
5751            Self::NOTE_KINDS
5752        }
5753        fn entity_kinds(&self) -> &'static [&'static str] {
5754            Self::ENTITY_KINDS
5755        }
5756        fn handlers(&self) -> &'static [HandlerDef] {
5757            Self::HANDLERS
5758        }
5759        fn requires(&self) -> &'static [&'static str] {
5760            Self::REQUIRES
5761        }
5762        async fn dispatch(
5763            &self,
5764            verb: &str,
5765            _: Value,
5766            _: &VerbRegistry,
5767            _: &NamespaceToken,
5768        ) -> Result<Value, RuntimeError> {
5769            Err(RuntimeError::InvalidInput(format!(
5770                "MemoryDepPack has no verbs: {verb}"
5771            )))
5772        }
5773    }
5774
5775    #[async_trait]
5776    impl PackRuntime for ADepPack {
5777        fn name(&self) -> &str {
5778            Self::NAME
5779        }
5780        fn note_kinds(&self) -> &'static [&'static str] {
5781            Self::NOTE_KINDS
5782        }
5783        fn entity_kinds(&self) -> &'static [&'static str] {
5784            Self::ENTITY_KINDS
5785        }
5786        fn handlers(&self) -> &'static [HandlerDef] {
5787            Self::HANDLERS
5788        }
5789        fn requires(&self) -> &'static [&'static str] {
5790            Self::REQUIRES
5791        }
5792        async fn dispatch(
5793            &self,
5794            verb: &str,
5795            _: Value,
5796            _: &VerbRegistry,
5797            _: &NamespaceToken,
5798        ) -> Result<Value, RuntimeError> {
5799            Err(RuntimeError::InvalidInput(format!(
5800                "ADepPack has no verbs: {verb}"
5801            )))
5802        }
5803    }
5804
5805    #[async_trait]
5806    impl PackRuntime for BDepPack {
5807        fn name(&self) -> &str {
5808            Self::NAME
5809        }
5810        fn note_kinds(&self) -> &'static [&'static str] {
5811            Self::NOTE_KINDS
5812        }
5813        fn entity_kinds(&self) -> &'static [&'static str] {
5814            Self::ENTITY_KINDS
5815        }
5816        fn handlers(&self) -> &'static [HandlerDef] {
5817            Self::HANDLERS
5818        }
5819        fn requires(&self) -> &'static [&'static str] {
5820            Self::REQUIRES
5821        }
5822        async fn dispatch(
5823            &self,
5824            verb: &str,
5825            _: Value,
5826            _: &VerbRegistry,
5827            _: &NamespaceToken,
5828        ) -> Result<Value, RuntimeError> {
5829            Err(RuntimeError::InvalidInput(format!(
5830                "BDepPack has no verbs: {verb}"
5831            )))
5832        }
5833    }
5834
5835    #[test]
5836    fn test_pack_deps_happy_path() {
5837        let mut builder = VerbRegistryBuilder::new();
5838        builder.register(MemoryDepPack);
5839        builder.register(KgDepPack);
5840        let reg = builder
5841            .build()
5842            .expect("kg_dep satisfies memory_dep dependency");
5843        assert_eq!(reg.pack_requires("memory_dep").unwrap(), &["kg_dep"]);
5844        let names = reg.pack_names();
5845        let kg_pos = names.iter().position(|&n| n == "kg_dep").unwrap();
5846        let mem_pos = names.iter().position(|&n| n == "memory_dep").unwrap();
5847        assert!(
5848            kg_pos < mem_pos,
5849            "kg_dep must be loaded before memory_dep; order: {names:?}"
5850        );
5851    }
5852
5853    #[test]
5854    fn test_pack_deps_missing() {
5855        let mut builder = VerbRegistryBuilder::new();
5856        builder.register(MemoryDepPack);
5857        let err = match builder.build() {
5858            Ok(_) => panic!("expected Err, got Ok"),
5859            Err(e) => e,
5860        };
5861        assert!(
5862            matches!(err, RuntimeError::MissingPackDependency(_)),
5863            "expected MissingPackDependency, got {err:?}"
5864        );
5865        let msg = err.to_string();
5866        assert!(
5867            msg.contains("memory_dep"),
5868            "error must name the dependent pack: {msg}"
5869        );
5870        assert!(
5871            msg.contains("kg_dep"),
5872            "error must name the missing dep: {msg}"
5873        );
5874    }
5875
5876    #[test]
5877    fn test_pack_deps_circular() {
5878        let mut builder = VerbRegistryBuilder::new();
5879        builder.register(ADepPack);
5880        builder.register(BDepPack);
5881        let err = match builder.build() {
5882            Ok(_) => panic!("expected Err, got Ok"),
5883            Err(e) => e,
5884        };
5885        assert!(
5886            matches!(err, RuntimeError::CircularPackDependency(_)),
5887            "expected CircularPackDependency, got {err:?}"
5888        );
5889        let msg = err.to_string();
5890        assert!(msg.contains("pack_a"), "error must name pack_a: {msg}");
5891        assert!(msg.contains("pack_b"), "error must name pack_b: {msg}");
5892    }
5893
5894    #[test]
5895    fn test_pack_deps_no_deps() {
5896        struct NoDepsA;
5897        struct NoDepsB;
5898
5899        impl Pack for NoDepsA {
5900            const NAME: &'static str = "no_deps_a";
5901            const NOTE_KINDS: &'static [&'static str] = &[];
5902            const ENTITY_KINDS: &'static [&'static str] = &[];
5903            const HANDLERS: &'static [HandlerDef] = &[];
5904        }
5905
5906        impl Pack for NoDepsB {
5907            const NAME: &'static str = "no_deps_b";
5908            const NOTE_KINDS: &'static [&'static str] = &[];
5909            const ENTITY_KINDS: &'static [&'static str] = &[];
5910            const HANDLERS: &'static [HandlerDef] = &[];
5911        }
5912
5913        #[async_trait]
5914        impl PackRuntime for NoDepsA {
5915            fn name(&self) -> &str {
5916                Self::NAME
5917            }
5918            fn note_kinds(&self) -> &'static [&'static str] {
5919                Self::NOTE_KINDS
5920            }
5921            fn entity_kinds(&self) -> &'static [&'static str] {
5922                Self::ENTITY_KINDS
5923            }
5924            fn handlers(&self) -> &'static [HandlerDef] {
5925                Self::HANDLERS
5926            }
5927            async fn dispatch(
5928                &self,
5929                verb: &str,
5930                _: Value,
5931                _: &VerbRegistry,
5932                _: &NamespaceToken,
5933            ) -> Result<Value, RuntimeError> {
5934                Err(RuntimeError::InvalidInput(format!("NoDepsA: {verb}")))
5935            }
5936        }
5937
5938        #[async_trait]
5939        impl PackRuntime for NoDepsB {
5940            fn name(&self) -> &str {
5941                Self::NAME
5942            }
5943            fn note_kinds(&self) -> &'static [&'static str] {
5944                Self::NOTE_KINDS
5945            }
5946            fn entity_kinds(&self) -> &'static [&'static str] {
5947                Self::ENTITY_KINDS
5948            }
5949            fn handlers(&self) -> &'static [HandlerDef] {
5950                Self::HANDLERS
5951            }
5952            async fn dispatch(
5953                &self,
5954                verb: &str,
5955                _: Value,
5956                _: &VerbRegistry,
5957                _: &NamespaceToken,
5958            ) -> Result<Value, RuntimeError> {
5959                Err(RuntimeError::InvalidInput(format!("NoDepsB: {verb}")))
5960            }
5961        }
5962
5963        let mut builder = VerbRegistryBuilder::new();
5964        builder.register(NoDepsA);
5965        builder.register(NoDepsB);
5966        let reg = builder.build().expect("packs with REQUIRES=&[] build");
5967        assert_eq!(reg.pack_requires("no_deps_a").unwrap(), &[] as &[&str]);
5968        assert_eq!(reg.pack_requires("no_deps_b").unwrap(), &[] as &[&str]);
5969    }
5970}
5971
5972// ── Dispatch hook tests ─────────────────────────────────────────
5973
5974#[cfg(test)]
5975mod hook_tests {
5976    use super::*;
5977    use async_trait::async_trait;
5978    use khive_types::Pack;
5979    use std::sync::atomic::{AtomicUsize, Ordering};
5980    use std::sync::Mutex as StdMutex;
5981
5982    struct SimplePack;
5983
5984    impl Pack for SimplePack {
5985        const NAME: &'static str = "simple";
5986        const NOTE_KINDS: &'static [&'static str] = &[];
5987        const ENTITY_KINDS: &'static [&'static str] = &[];
5988        const HANDLERS: &'static [HandlerDef] = &[HandlerDef {
5989            name: "ping",
5990            description: "ping",
5991            visibility: Visibility::Verb,
5992            category: VerbCategory::Assertive,
5993            params: &[],
5994        }];
5995    }
5996
5997    #[async_trait]
5998    impl PackRuntime for SimplePack {
5999        fn name(&self) -> &str {
6000            SimplePack::NAME
6001        }
6002        fn note_kinds(&self) -> &'static [&'static str] {
6003            SimplePack::NOTE_KINDS
6004        }
6005        fn entity_kinds(&self) -> &'static [&'static str] {
6006            SimplePack::ENTITY_KINDS
6007        }
6008        fn handlers(&self) -> &'static [HandlerDef] {
6009            SimplePack::HANDLERS
6010        }
6011        async fn dispatch(
6012            &self,
6013            verb: &str,
6014            _params: Value,
6015            _registry: &VerbRegistry,
6016            _token: &NamespaceToken,
6017        ) -> Result<Value, RuntimeError> {
6018            Ok(serde_json::json!({ "verb": verb }))
6019        }
6020    }
6021
6022    /// Hook that counts calls and records the last verb seen.
6023    #[derive(Default)]
6024    struct CountingHook {
6025        calls: AtomicUsize,
6026        last_verb: StdMutex<String>,
6027    }
6028
6029    #[async_trait]
6030    impl DispatchHook for CountingHook {
6031        async fn on_dispatch(&self, view: &EventView) {
6032            self.calls.fetch_add(1, Ordering::SeqCst);
6033            *self.last_verb.lock().unwrap() = view.event.verb.clone();
6034        }
6035    }
6036
6037    #[tokio::test]
6038    async fn dispatch_hook_fires_on_successful_dispatch() {
6039        let hook = Arc::new(CountingHook::default());
6040        let mut builder = VerbRegistryBuilder::new();
6041        builder.register(SimplePack);
6042        builder.with_dispatch_hook(hook.clone());
6043        let reg = builder.build().expect("registry builds");
6044
6045        reg.dispatch("ping", Value::Null).await.unwrap();
6046
6047        assert_eq!(
6048            hook.calls.load(Ordering::SeqCst),
6049            1,
6050            "hook must fire once per successful dispatch"
6051        );
6052        assert_eq!(
6053            hook.last_verb.lock().unwrap().as_str(),
6054            "ping",
6055            "hook event must carry the dispatched verb"
6056        );
6057    }
6058
6059    #[tokio::test]
6060    async fn dispatch_hook_fires_multiple_times() {
6061        let hook = Arc::new(CountingHook::default());
6062        let mut builder = VerbRegistryBuilder::new();
6063        builder.register(SimplePack);
6064        builder.with_dispatch_hook(hook.clone());
6065        let reg = builder.build().expect("registry builds");
6066
6067        reg.dispatch("ping", Value::Null).await.unwrap();
6068        reg.dispatch("ping", Value::Null).await.unwrap();
6069        reg.dispatch("ping", Value::Null).await.unwrap();
6070
6071        assert_eq!(
6072            hook.calls.load(Ordering::SeqCst),
6073            3,
6074            "hook must fire once per successful dispatch"
6075        );
6076    }
6077
6078    #[tokio::test]
6079    async fn dispatch_hook_does_not_fire_on_unknown_verb() {
6080        let hook = Arc::new(CountingHook::default());
6081        let mut builder = VerbRegistryBuilder::new();
6082        builder.register(SimplePack);
6083        builder.with_dispatch_hook(hook.clone());
6084        let reg = builder.build().expect("registry builds");
6085
6086        let _ = reg.dispatch("nonexistent", Value::Null).await;
6087
6088        assert_eq!(
6089            hook.calls.load(Ordering::SeqCst),
6090            0,
6091            "hook must NOT fire for unknown verb (dispatch returns error)"
6092        );
6093    }
6094
6095    #[tokio::test]
6096    async fn dispatch_hook_does_not_fire_on_gate_deny() {
6097        use khive_gate::{Gate, GateDecision, GateError};
6098
6099        #[derive(Debug)]
6100        struct AlwaysDenyGate;
6101        impl Gate for AlwaysDenyGate {
6102            fn check(&self, _req: &GateRequest) -> Result<GateDecision, GateError> {
6103                Ok(GateDecision::deny("test deny"))
6104            }
6105        }
6106
6107        let hook = Arc::new(CountingHook::default());
6108        let mut builder = VerbRegistryBuilder::new();
6109        builder.register(SimplePack);
6110        builder.with_gate(Arc::new(AlwaysDenyGate));
6111        builder.with_dispatch_hook(hook.clone());
6112        let reg = builder.build().expect("registry builds");
6113
6114        let err = reg.dispatch("ping", Value::Null).await.unwrap_err();
6115        assert!(matches!(err, RuntimeError::PermissionDenied { .. }));
6116
6117        assert_eq!(
6118            hook.calls.load(Ordering::SeqCst),
6119            0,
6120            "hook must NOT fire when gate denies dispatch"
6121        );
6122    }
6123
6124    #[tokio::test]
6125    async fn dispatch_hook_event_carries_namespace_from_params() {
6126        let hook = Arc::new(CountingHook::default());
6127
6128        #[derive(Default)]
6129        struct NsCapturingHook {
6130            ns: StdMutex<String>,
6131        }
6132
6133        #[async_trait]
6134        impl DispatchHook for NsCapturingHook {
6135            async fn on_dispatch(&self, view: &EventView) {
6136                *self.ns.lock().unwrap() = view.event.namespace.clone();
6137            }
6138        }
6139
6140        let ns_hook = Arc::new(NsCapturingHook::default());
6141        let mut builder = VerbRegistryBuilder::new();
6142        builder.register(SimplePack);
6143        builder.with_dispatch_hook(ns_hook.clone());
6144        let reg = builder.build().expect("registry builds");
6145
6146        reg.dispatch("ping", serde_json::json!({"namespace": "tenant-abc"}))
6147            .await
6148            .unwrap();
6149
6150        assert_eq!(
6151            ns_hook.ns.lock().unwrap().as_str(),
6152            "tenant-abc",
6153            "dispatch hook event must carry the resolved namespace"
6154        );
6155
6156        // Suppress unused-variable warning from the outer hook.
6157        drop(hook);
6158    }
6159
6160    #[tokio::test]
6161    async fn no_dispatch_hook_configured_dispatch_succeeds() {
6162        // Regression: registries without a hook must still work.
6163        let mut builder = VerbRegistryBuilder::new();
6164        builder.register(SimplePack);
6165        // No with_dispatch_hook call.
6166        let reg = builder.build().expect("registry builds");
6167
6168        let res = reg.dispatch("ping", Value::Null).await.unwrap();
6169        assert_eq!(res["verb"], "ping");
6170    }
6171}
6172
6173// ── help=true tests ──────────────────────────────────────────────
6174
6175#[cfg(test)]
6176mod help_tests {
6177    use super::*;
6178    use async_trait::async_trait;
6179    use khive_types::Pack;
6180    use std::sync::{
6181        atomic::{AtomicUsize, Ordering},
6182        Arc,
6183    };
6184
6185    // ── HelpPack: a minimal pack with one handler that records invocation count.
6186    //
6187    // Used to verify that help=true never reaches the pack's dispatch method.
6188
6189    static CREATE_PARAMS: [ParamDef; 2] = [
6190        ParamDef {
6191            name: "kind",
6192            param_type: "string",
6193            required: true,
6194            description: "Granular kind (concept | document | ...).",
6195        },
6196        ParamDef {
6197            name: "name",
6198            param_type: "string",
6199            required: false,
6200            description: "Human-readable name.",
6201        },
6202    ];
6203
6204    static RECALL_PARAMS: [ParamDef; 2] = [
6205        ParamDef {
6206            name: "query",
6207            param_type: "string",
6208            required: true,
6209            description: "Semantic recall query.",
6210        },
6211        ParamDef {
6212            name: "limit",
6213            param_type: "integer",
6214            required: false,
6215            description: "Maximum memories to return.",
6216        },
6217    ];
6218
6219    // A subhandler with no params — mirrors recall.embed / brain.emit / etc.
6220    // Used to test that help=true on a Subhandler returns callable_via_mcp: false.
6221    static EMBED_PARAMS: [ParamDef; 0] = [];
6222
6223    struct HelpPack {
6224        invocations: Arc<AtomicUsize>,
6225    }
6226
6227    impl Pack for HelpPack {
6228        const NAME: &'static str = "helptest";
6229        const NOTE_KINDS: &'static [&'static str] = &[];
6230        const ENTITY_KINDS: &'static [&'static str] = &[];
6231        const HANDLERS: &'static [HandlerDef] = &[
6232            HandlerDef {
6233                name: "create",
6234                description: "Create an entity or note",
6235                visibility: Visibility::Verb,
6236                category: VerbCategory::Commissive,
6237                params: &CREATE_PARAMS,
6238            },
6239            HandlerDef {
6240                name: "recall",
6241                description: "Recall memory notes with decay-aware hybrid ranking",
6242                visibility: Visibility::Verb,
6243                category: VerbCategory::Assertive,
6244                params: &RECALL_PARAMS,
6245            },
6246            // A Subhandler used to test that help=true returns
6247            // callable_via_mcp: false for internal verbs.
6248            HandlerDef {
6249                name: "recall.embed",
6250                description: "Return the embedding vector used by memory recall",
6251                visibility: Visibility::Subhandler,
6252                category: VerbCategory::Assertive,
6253                params: &EMBED_PARAMS,
6254            },
6255            HandlerDef {
6256                name: "link",
6257                description: "Create a typed directed edge",
6258                visibility: Visibility::Verb,
6259                category: VerbCategory::Commissive,
6260                params: &[],
6261            },
6262        ];
6263    }
6264
6265    // A pack-declared additive edge rule (mirrors the GTD pack's real
6266    // task-to-task `depends_on` rule), used to verify `link(help=true)`
6267    // surfaces pack-composed rules alongside the base entity table. The
6268    // second entry declares a rule for a special relation
6269    // (`supersedes`) that the validator's dedicated special-relation
6270    // branch never consults `pack_rule_allows` for — it must NOT be
6271    // advertised (see `test_link_help_true_matches_special_relation_validator_set`).
6272    static HELP_EDGE_RULES: [EdgeEndpointRule; 2] = [
6273        EdgeEndpointRule {
6274            relation: khive_types::EdgeRelation::DependsOn,
6275            source: EndpointKind::NoteOfKind("task"),
6276            target: EndpointKind::NoteOfKind("task"),
6277        },
6278        EdgeEndpointRule {
6279            relation: khive_types::EdgeRelation::Supersedes,
6280            source: EndpointKind::NoteOfKind("task"),
6281            target: EndpointKind::NoteOfKind("task"),
6282        },
6283    ];
6284
6285    #[async_trait]
6286    impl PackRuntime for HelpPack {
6287        fn name(&self) -> &str {
6288            HelpPack::NAME
6289        }
6290        fn note_kinds(&self) -> &'static [&'static str] {
6291            HelpPack::NOTE_KINDS
6292        }
6293        fn entity_kinds(&self) -> &'static [&'static str] {
6294            HelpPack::ENTITY_KINDS
6295        }
6296        fn handlers(&self) -> &'static [HandlerDef] {
6297            HelpPack::HANDLERS
6298        }
6299        fn edge_rules(&self) -> &'static [EdgeEndpointRule] {
6300            &HELP_EDGE_RULES
6301        }
6302        async fn dispatch(
6303            &self,
6304            verb: &str,
6305            _params: Value,
6306            _registry: &VerbRegistry,
6307            _token: &NamespaceToken,
6308        ) -> Result<Value, RuntimeError> {
6309            self.invocations.fetch_add(1, Ordering::SeqCst);
6310            Ok(serde_json::json!({ "pack": "helptest", "verb": verb }))
6311        }
6312    }
6313
6314    fn build_help_registry(invocations: Arc<AtomicUsize>) -> VerbRegistry {
6315        let mut builder = VerbRegistryBuilder::new();
6316        builder.register(HelpPack { invocations });
6317        builder.build().expect("help registry builds")
6318    }
6319
6320    /// help=true on `create` returns a schema envelope with the correct verb name,
6321    /// pack name, description, and at least the required `kind` parameter.
6322    #[tokio::test]
6323    async fn test_help_true_returns_schema_for_kg_create() {
6324        let invocations = Arc::new(AtomicUsize::new(0));
6325        let reg = build_help_registry(invocations.clone());
6326
6327        let result = reg
6328            .dispatch("create", serde_json::json!({ "help": true }))
6329            .await
6330            .expect("help=true must succeed for a known verb");
6331
6332        // Shape checks.
6333        assert_eq!(result["verb"], "create", "envelope must name the verb");
6334        assert_eq!(
6335            result["pack"], "helptest",
6336            "envelope must name the owning pack"
6337        );
6338        assert!(
6339            result["description"].as_str().is_some(),
6340            "description must be a string"
6341        );
6342
6343        // Params array must be present and non-empty.
6344        let params = result["params"]
6345            .as_array()
6346            .expect("params must be a JSON array");
6347        assert!(!params.is_empty(), "params array must not be empty");
6348
6349        // The required `kind` param must appear.
6350        let kind_param = params.iter().find(|p| p["name"] == "kind");
6351        assert!(
6352            kind_param.is_some(),
6353            "params array must include the 'kind' parameter"
6354        );
6355        let kind_param = kind_param.unwrap();
6356        assert_eq!(
6357            kind_param["required"],
6358            serde_json::json!(true),
6359            "'kind' must be required"
6360        );
6361        assert_eq!(kind_param["type"], "string", "'kind' type must be 'string'");
6362    }
6363
6364    /// help=true on `recall` returns a schema envelope including the `query` param.
6365    #[tokio::test]
6366    async fn test_help_true_returns_schema_for_recall() {
6367        let invocations = Arc::new(AtomicUsize::new(0));
6368        let reg = build_help_registry(invocations.clone());
6369
6370        let result = reg
6371            .dispatch("recall", serde_json::json!({ "help": true }))
6372            .await
6373            .expect("help=true must succeed for recall");
6374
6375        assert_eq!(result["verb"], "recall");
6376        assert_eq!(result["pack"], "helptest");
6377
6378        let params = result["params"]
6379            .as_array()
6380            .expect("params must be a JSON array");
6381
6382        // `query` must be present and required.
6383        let query_param = params.iter().find(|p| p["name"] == "query");
6384        assert!(query_param.is_some(), "params must include 'query'");
6385        let query_param = query_param.unwrap();
6386        assert_eq!(
6387            query_param["required"],
6388            serde_json::json!(true),
6389            "'query' must be required"
6390        );
6391
6392        // `limit` must be present and optional.
6393        let limit_param = params.iter().find(|p| p["name"] == "limit");
6394        assert!(limit_param.is_some(), "params must include 'limit'");
6395        let limit_param = limit_param.unwrap();
6396        assert_eq!(
6397            limit_param["required"],
6398            serde_json::json!(false),
6399            "'limit' must be optional"
6400        );
6401    }
6402
6403    /// `link(help=true)` (issue #964) surfaces the composed per-relation
6404    /// endpoint allowlist: the base entity-to-entity table, every loaded
6405    /// pack's additive `EDGE_RULES`, and the `annotates` note-to-any rule —
6406    /// so a batch caller can defer to the kernel's own table instead of
6407    /// re-implementing it.
6408    #[tokio::test]
6409    async fn test_link_help_true_exposes_endpoint_rules() {
6410        let invocations = Arc::new(AtomicUsize::new(0));
6411        let reg = build_help_registry(invocations.clone());
6412
6413        let result = reg
6414            .dispatch("link", serde_json::json!({ "help": true }))
6415            .await
6416            .expect("help=true must succeed for link");
6417
6418        assert_eq!(result["verb"], "link");
6419        let rules = result["endpoint_rules"]
6420            .as_array()
6421            .expect("link help must include an endpoint_rules array");
6422        assert!(!rules.is_empty(), "endpoint_rules must not be empty");
6423
6424        // A base entity-to-entity rule (khive-runtime's own table) must appear.
6425        assert!(
6426            rules.iter().any(|r| r["relation"] == "contains"
6427                && r["source"] == "entity:concept"
6428                && r["target"] == "entity:concept"),
6429            "endpoint_rules must include the base 'contains' entity rule; got {rules:#?}"
6430        );
6431
6432        // The pack-declared additive rule (HelpPack's task->task depends_on) must appear.
6433        assert!(
6434            rules.iter().any(|r| r["relation"] == "depends_on"
6435                && r["source"] == "note:task"
6436                && r["target"] == "note:task"),
6437            "endpoint_rules must include the pack-declared depends_on rule; got {rules:#?}"
6438        );
6439
6440        // The annotates note-to-any special case must appear.
6441        assert!(
6442            rules
6443                .iter()
6444                .any(|r| r["relation"] == "annotates" && r["source"] == "note:*"),
6445            "endpoint_rules must document the annotates note-to-any rule; got {rules:#?}"
6446        );
6447
6448        // help=true must remain side-effect-free.
6449        assert_eq!(
6450            invocations.load(Ordering::SeqCst),
6451            0,
6452            "link(help=true) must not invoke pack dispatch"
6453        );
6454    }
6455
6456    /// `link(help=true)`'s `endpoint_rules` must match, set-for-set, every
6457    /// endpoint pair `validate_edge_relation_endpoints`
6458    /// (`crates/khive-runtime/src/operations.rs`) actually accepts for the
6459    /// three special relations (`supersedes` / `supports` / `refutes`):
6460    ///
6461    /// - a `note -> note` row for each of the three relations (the
6462    ///   validator's dedicated special-relation branch accepts any
6463    ///   `Resolved::Note(_), Resolved::Note(_)` pair unconditionally,
6464    ///   `operations.rs:1338` / `:1527` — before `pack_rule_allows` is ever
6465    ///   reached);
6466    /// - the base entity->entity rows for the three relations
6467    ///   (`base_entity_endpoint_rules`, e.g. `concept -[supersedes]-> concept`);
6468    /// - and, critically, NOT a row for `HelpPack`'s pack-declared
6469    ///   `supersedes` rule on `note:task -> note:task`
6470    ///   (`HELP_EDGE_RULES[1]`) — because the validator's special-relation
6471    ///   branch returns before `pack_rule_allows` is consulted, that pack
6472    ///   rule is never actually enforced, so advertising it would be a false
6473    ///   promise (the exact defect this test guards against, issue #991).
6474    #[tokio::test]
6475    async fn test_link_help_true_matches_special_relation_validator_set() {
6476        let invocations = Arc::new(AtomicUsize::new(0));
6477        let reg = build_help_registry(invocations.clone());
6478
6479        let result = reg
6480            .dispatch("link", serde_json::json!({ "help": true }))
6481            .await
6482            .expect("help=true must succeed for link");
6483
6484        let rules = result["endpoint_rules"]
6485            .as_array()
6486            .expect("link help must include an endpoint_rules array");
6487
6488        for relation in ["supersedes", "supports", "refutes"] {
6489            // The unconditional note -> note row must appear.
6490            assert!(
6491                rules.iter().any(|r| r["relation"] == relation
6492                    && r["source"] == "note:*"
6493                    && r["target"] == "note:*"),
6494                "endpoint_rules must include the note:*->note:* row for '{relation}' \
6495                 (validator accepts any note->note pair unconditionally); got {rules:#?}"
6496            );
6497
6498            // HelpPack's pack-declared rule for this relation on note:task->note:task
6499            // (only Supersedes is declared in HELP_EDGE_RULES) must NOT be advertised
6500            // as a distinct entity — the validator never reaches pack_rule_allows for
6501            // special relations, so no note:task->note:task row should exist for it.
6502            assert!(
6503                !rules.iter().any(|r| r["relation"] == relation
6504                    && r["source"] == "note:task"
6505                    && r["target"] == "note:task"),
6506                "endpoint_rules must NOT advertise a pack EDGE_RULES row for special \
6507                 relation '{relation}' — validate_edge_relation_endpoints never consults \
6508                 pack_rule_allows for supersedes/supports/refutes; got {rules:#?}"
6509            );
6510        }
6511
6512        // Base entity->entity rows for the three relations (from
6513        // base_entity_endpoint_rules) must still appear alongside the note rows.
6514        for (relation, kind) in [
6515            ("supersedes", "concept"),
6516            ("supports", "concept"),
6517            ("refutes", "concept"),
6518        ] {
6519            assert!(
6520                rules.iter().any(|r| r["relation"] == relation
6521                    && r["source"] == format!("entity:{kind}")
6522                    && r["target"] == "entity:concept"),
6523                "endpoint_rules must include the base entity:{kind}->entity:concept row \
6524                 for '{relation}'; got {rules:#?}"
6525            );
6526        }
6527    }
6528
6529    /// help=true is intercepted before pack dispatch — the pack's dispatch method
6530    /// must never be invoked when help=true is in the params.
6531    #[tokio::test]
6532    async fn test_help_true_does_not_execute_the_verb() {
6533        let invocations = Arc::new(AtomicUsize::new(0));
6534        let reg = build_help_registry(invocations.clone());
6535
6536        // Call both verbs with help=true.
6537        reg.dispatch("create", serde_json::json!({ "help": true }))
6538            .await
6539            .expect("help=true must succeed");
6540        reg.dispatch("recall", serde_json::json!({ "help": true }))
6541            .await
6542            .expect("help=true must succeed");
6543
6544        assert_eq!(
6545            invocations.load(Ordering::SeqCst),
6546            0,
6547            "pack dispatch MUST NOT be invoked when help=true; \
6548             got {} invocation(s)",
6549            invocations.load(Ordering::SeqCst)
6550        );
6551
6552        // Confirm that a normal call (without help=true) DOES invoke dispatch.
6553        reg.dispatch("create", serde_json::json!({}))
6554            .await
6555            .expect("normal dispatch must succeed");
6556        assert_eq!(
6557            invocations.load(Ordering::SeqCst),
6558            1,
6559            "pack dispatch must fire exactly once for a normal call"
6560        );
6561    }
6562
6563    // ── Subhandler help-schema regressions ─────────────────────────────────
6564    //
6565    // Subhandler verbs must return `callable_via_mcp: false` in their help
6566    // schema so agents who read help=true before probing see accurate
6567    // availability — not a "looks callable" schema followed by permission denied.
6568
6569    /// help=true on a `Visibility::Subhandler` verb returns `callable_via_mcp: false`
6570    /// and `visibility: "internal"` rather than a plain callable-looking envelope.
6571    #[tokio::test]
6572    async fn help_true_on_subhandler_returns_callable_via_mcp_false() {
6573        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6574
6575        let result = reg
6576            .dispatch("recall.embed", serde_json::json!({ "help": true }))
6577            .await
6578            .expect("help=true on subhandler must succeed (no permission check on help path)");
6579
6580        assert_eq!(
6581            result["callable_via_mcp"],
6582            serde_json::json!(false),
6583            "subhandler help must carry callable_via_mcp: false"
6584        );
6585        assert_eq!(
6586            result["visibility"], "internal",
6587            "subhandler help must carry visibility: internal"
6588        );
6589        // The verb and pack fields must still be present so the caller knows
6590        // what the schema belongs to.
6591        assert_eq!(result["verb"], "recall.embed");
6592        assert_eq!(result["pack"], "helptest");
6593    }
6594
6595    /// Public Verb-visibility handlers must NOT have `callable_via_mcp: false`.
6596    #[tokio::test]
6597    async fn help_true_on_public_verb_does_not_have_callable_via_mcp_false() {
6598        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6599
6600        let result = reg
6601            .dispatch("create", serde_json::json!({ "help": true }))
6602            .await
6603            .expect("help=true on public verb must succeed");
6604
6605        // callable_via_mcp must be absent or true for public verbs.
6606        assert_ne!(
6607            result.get("callable_via_mcp"),
6608            Some(&serde_json::json!(false)),
6609            "public verb help must NOT carry callable_via_mcp: false"
6610        );
6611        // visibility must be absent or 'public' (never 'internal') for public verbs.
6612        assert_ne!(
6613            result.get("visibility"),
6614            Some(&serde_json::json!("internal")),
6615            "public verb help must NOT carry visibility: internal"
6616        );
6617    }
6618
6619    /// help=true on an unknown verb returns an error (same behavior as normal dispatch).
6620    #[tokio::test]
6621    async fn help_true_on_unknown_verb_returns_error() {
6622        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6623
6624        let err = reg
6625            .dispatch("nonexistent_verb", serde_json::json!({ "help": true }))
6626            .await
6627            .unwrap_err();
6628
6629        assert!(
6630            matches!(err, RuntimeError::InvalidInput(_)),
6631            "help=true on unknown verb must return InvalidInput, got {err:?}"
6632        );
6633        let msg = err.to_string();
6634        assert!(
6635            msg.contains("nonexistent_verb"),
6636            "error must name the unknown verb: {msg}"
6637        );
6638    }
6639
6640    /// Subhandler help must include params: [] even when the verb has no params.
6641    #[tokio::test]
6642    async fn help_true_on_subhandler_includes_params_field() {
6643        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6644
6645        let result = reg
6646            .dispatch("recall.embed", serde_json::json!({ "help": true }))
6647            .await
6648            .expect("help=true on subhandler must succeed");
6649
6650        // params must always be present (consistent shape).
6651        let params = result
6652            .get("params")
6653            .expect("subhandler help must include 'params' field");
6654        assert!(
6655            params.is_array(),
6656            "subhandler help params must be a JSON array"
6657        );
6658    }
6659
6660    // ── Unknown-verb error must not leak subhandler names ─────────
6661
6662    /// `describe_verb` on an unknown verb must list only Verb-visibility names
6663    /// in the "available" list: never subhandler names like `recall.embed`.
6664    #[tokio::test]
6665    async fn help_true_unknown_verb_available_list_excludes_subhandlers() {
6666        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6667
6668        let err = reg
6669            .dispatch("not_a_verb", serde_json::json!({ "help": true }))
6670            .await
6671            .unwrap_err();
6672
6673        let msg = err.to_string();
6674        // `recall.embed` is a Subhandler in HelpPack — must NOT appear in the
6675        // "available" list of an unknown-verb error.
6676        assert!(
6677            !msg.contains("recall.embed"),
6678            "unknown-verb help error must not advertise subhandler recall.embed: {msg}"
6679        );
6680        // Public verbs must still appear so the agent knows what to call.
6681        assert!(
6682            msg.contains("create"),
6683            "unknown-verb help error must still list public verb 'create': {msg}"
6684        );
6685        assert!(
6686            msg.contains("recall"),
6687            "unknown-verb help error must still list public verb 'recall': {msg}"
6688        );
6689    }
6690
6691    /// Normal dispatch on an unknown verb must also not leak subhandler names.
6692    #[tokio::test]
6693    async fn dispatch_unknown_verb_available_list_excludes_subhandlers() {
6694        let reg = build_help_registry(Arc::new(AtomicUsize::new(0)));
6695
6696        let err = reg
6697            .dispatch("not_a_verb", serde_json::json!({}))
6698            .await
6699            .unwrap_err();
6700
6701        let msg = err.to_string();
6702        // `recall.embed` is a Subhandler in HelpPack — must NOT appear in the
6703        // "available" list of an unknown-verb dispatch error.
6704        assert!(
6705            !msg.contains("recall.embed"),
6706            "dispatch unknown-verb error must not advertise subhandler recall.embed: {msg}"
6707        );
6708        // Public verbs must still appear so the agent knows what to call.
6709        assert!(
6710            msg.contains("create"),
6711            "dispatch unknown-verb error must still list public verb 'create': {msg}"
6712        );
6713        assert!(
6714            msg.contains("recall"),
6715            "dispatch unknown-verb error must still list public verb 'recall': {msg}"
6716        );
6717    }
6718
6719    // ── ADR-028 multi-backend schema routing tests ───────────────────────────
6720
6721    /// A test pack that returns a real SchemaPlan so we can assert routing.
6722    struct SchemaPack {
6723        pack_name: &'static str,
6724        statements: &'static [&'static str],
6725    }
6726
6727    impl Pack for SchemaPack {
6728        const NAME: &'static str = "schema-pack";
6729        const NOTE_KINDS: &'static [&'static str] = &[];
6730        const ENTITY_KINDS: &'static [&'static str] = &[];
6731        const HANDLERS: &'static [HandlerDef] = &[];
6732    }
6733
6734    #[async_trait]
6735    impl PackRuntime for SchemaPack {
6736        fn name(&self) -> &str {
6737            self.pack_name
6738        }
6739        fn note_kinds(&self) -> &'static [&'static str] {
6740            &[]
6741        }
6742        fn entity_kinds(&self) -> &'static [&'static str] {
6743            &[]
6744        }
6745        fn handlers(&self) -> &'static [HandlerDef] {
6746            &[]
6747        }
6748        fn schema_plan(&self) -> SchemaPlan {
6749            SchemaPlan {
6750                pack: self.pack_name,
6751                statements: self.statements,
6752            }
6753        }
6754        async fn dispatch(
6755            &self,
6756            verb: &str,
6757            _params: Value,
6758            _registry: &VerbRegistry,
6759            _token: &NamespaceToken,
6760        ) -> Result<Value, RuntimeError> {
6761            Ok(serde_json::json!({ "pack": self.pack_name, "verb": verb }))
6762        }
6763    }
6764
6765    // ADR-028: all_schema_plans_named returns (pack_name, SchemaPlan) pairs
6766    // where pack_name comes from SchemaPlan::pack (always &'static str).
6767    #[test]
6768    fn all_schema_plans_named_returns_correct_pairs() {
6769        let mut builder = VerbRegistryBuilder::new();
6770        builder.register_boxed(Box::new(SchemaPack {
6771            pack_name: "alpha",
6772            statements: &["CREATE TABLE IF NOT EXISTS t_alpha (id INTEGER PRIMARY KEY)"],
6773        }));
6774        builder.register_boxed(Box::new(SchemaPack {
6775            pack_name: "beta",
6776            statements: &[],
6777        }));
6778        let reg = builder.build().expect("registry builds");
6779
6780        let named = reg.all_schema_plans_named();
6781        assert_eq!(named.len(), 2);
6782
6783        let alpha_entry = named.iter().find(|(n, _)| *n == "alpha");
6784        let beta_entry = named.iter().find(|(n, _)| *n == "beta");
6785
6786        assert!(alpha_entry.is_some(), "alpha must appear in named plans");
6787        assert!(beta_entry.is_some(), "beta must appear in named plans");
6788
6789        let (_, alpha_plan) = alpha_entry.unwrap();
6790        assert_eq!(alpha_plan.statements.len(), 1);
6791        assert!(!alpha_plan.is_empty());
6792
6793        let (_, beta_plan) = beta_entry.unwrap();
6794        assert!(beta_plan.is_empty());
6795    }
6796
6797    // ADR-028: apply_schema_plans_with_map routes non-empty plans to the
6798    // correct per-pack backend instead of the default.
6799    //
6800    // Verification: apply DDL to routed backend, then confirm the table is
6801    // present on pack_backend and absent on default_backend by attempting to
6802    // apply the same DDL again — if the table already exists on pack_backend
6803    // the idempotent CREATE IF NOT EXISTS succeeds; applying to default_backend
6804    // would only matter if the table were routed there.  We verify isolation
6805    // by applying the plan and then running a targeted DDL on each backend
6806    // that would fail if the table did not already exist (CREATE without
6807    // IF NOT EXISTS on a duplicate raises an error), combined with a no-error
6808    // path on the correct backend.
6809    //
6810    // Simpler approach: confirm the plan applies without error (routing is
6811    // correct) and that the opposite backend returns an error when we try to
6812    // INSERT into the routed table (table-not-found = SQLITE_ERROR).
6813    #[tokio::test]
6814    async fn apply_schema_plans_with_map_routes_to_correct_backend() {
6815        use khive_storage::types::{SqlStatement, SqlValue};
6816
6817        let default_backend = khive_db::StorageBackend::memory().expect("default memory backend");
6818        let pack_backend =
6819            khive_db::StorageBackend::memory().expect("pack-specific memory backend");
6820
6821        let mut builder = VerbRegistryBuilder::new();
6822        builder.register_boxed(Box::new(SchemaPack {
6823            pack_name: "routed",
6824            statements: &["CREATE TABLE IF NOT EXISTS t_routed (id INTEGER PRIMARY KEY)"],
6825        }));
6826        let reg = builder.build().expect("registry builds");
6827
6828        let mut backend_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
6829        backend_map.insert("routed", &pack_backend);
6830
6831        reg.apply_schema_plans_with_map(&backend_map, &default_backend)
6832            .expect("schema application must not collide");
6833
6834        // On pack_backend: INSERT must succeed (table exists).
6835        let mut writer = pack_backend.sql().writer().await.expect("writer");
6836        let result = writer
6837            .execute(SqlStatement {
6838                sql: "INSERT INTO t_routed (id) VALUES (?1)".into(),
6839                params: vec![SqlValue::Integer(1)],
6840                label: None,
6841            })
6842            .await;
6843        assert!(
6844            result.is_ok(),
6845            "t_routed must exist on pack_backend after routing: {result:?}"
6846        );
6847
6848        // On default_backend: INSERT must fail (table not there).
6849        let mut default_writer = default_backend.sql().writer().await.expect("writer");
6850        let default_result = default_writer
6851            .execute(SqlStatement {
6852                sql: "INSERT INTO t_routed (id) VALUES (?1)".into(),
6853                params: vec![SqlValue::Integer(2)],
6854                label: None,
6855            })
6856            .await;
6857        assert!(
6858            default_result.is_err(),
6859            "t_routed must NOT exist on default_backend (table should not be there)"
6860        );
6861    }
6862
6863    // ADR-028: apply_schema_plans_with_map uses default backend for packs
6864    // absent from the map.
6865    #[tokio::test]
6866    async fn apply_schema_plans_with_map_falls_back_to_default_for_unmapped_packs() {
6867        use khive_storage::types::{SqlStatement, SqlValue};
6868
6869        let default_backend = khive_db::StorageBackend::memory().expect("default memory backend");
6870
6871        let mut builder = VerbRegistryBuilder::new();
6872        builder.register_boxed(Box::new(SchemaPack {
6873            pack_name: "unmapped",
6874            statements: &["CREATE TABLE IF NOT EXISTS t_unmapped (id INTEGER PRIMARY KEY)"],
6875        }));
6876        let reg = builder.build().expect("registry builds");
6877
6878        let backend_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
6879        reg.apply_schema_plans_with_map(&backend_map, &default_backend)
6880            .expect("schema application must not collide");
6881
6882        // On default_backend: INSERT must succeed (table fell back here).
6883        let mut writer = default_backend.sql().writer().await.expect("writer");
6884        let result = writer
6885            .execute(SqlStatement {
6886                sql: "INSERT INTO t_unmapped (id) VALUES (?1)".into(),
6887                params: vec![SqlValue::Integer(1)],
6888                label: None,
6889            })
6890            .await;
6891        assert!(
6892            result.is_ok(),
6893            "t_unmapped must exist on default_backend for unmapped pack: {result:?}"
6894        );
6895    }
6896
6897    // ADR-028: two packs declaring the same auxiliary table on the same
6898    // backend must cause apply_schema_plans_with_map to return an error that
6899    // names both packs and the table: it is a boot-time failure, not a
6900    // silent DDL race.
6901    #[test]
6902    fn apply_schema_plans_with_map_collision_is_an_error() {
6903        let backend = khive_db::StorageBackend::memory().expect("memory backend");
6904        let empty_map: HashMap<&str, &khive_db::StorageBackend> = HashMap::new();
6905
6906        let mut builder = VerbRegistryBuilder::new();
6907        builder.register_boxed(Box::new(SchemaPack {
6908            pack_name: "pack_alpha",
6909            statements: &["CREATE TABLE IF NOT EXISTS collision_table (id INTEGER PRIMARY KEY)"],
6910        }));
6911        builder.register_boxed(Box::new(SchemaPack {
6912            pack_name: "pack_beta",
6913            statements: &["CREATE TABLE IF NOT EXISTS collision_table (id INTEGER PRIMARY KEY)"],
6914        }));
6915        let registry = builder.build().expect("registry builds");
6916
6917        let result = registry.apply_schema_plans_with_map(&empty_map, &backend);
6918        assert!(
6919            result.is_err(),
6920            "two packs declaring the same table on the same backend must produce a collision error"
6921        );
6922        let err = result.unwrap_err();
6923        let msg = err.to_string();
6924        assert!(
6925            msg.contains("pack_alpha"),
6926            "collision error must name first pack; got: {msg}"
6927        );
6928        assert!(
6929            msg.contains("pack_beta"),
6930            "collision error must name second pack; got: {msg}"
6931        );
6932        assert!(
6933            msg.contains("collision_table"),
6934            "collision error must name the table; got: {msg}"
6935        );
6936    }
6937}