Skip to main content

khive_pack_brain/
handlers.rs

1//! Verb handler implementations for `BrainPack`.
2
3use std::time::Instant;
4
5use async_trait::async_trait;
6use chrono::Utc;
7use serde::Deserialize;
8use serde_json::{json, Value};
9
10use khive_runtime::{
11    micros_to_iso, DispatchHook, EventView, KhiveRuntime, NamespaceToken, RuntimeError,
12    VerbRegistry,
13};
14use khive_storage::event::{Event, EventFilter};
15use khive_storage::types::PageRequest;
16use khive_types::HandlerDef;
17
18use crate::event::interpret;
19use crate::{sync_balanced_recall_record, BrainPack, ENTITY_CACHE_CAPACITY};
20use khive_brain_core::derive_deterministic_weights;
21#[cfg(feature = "lattice-router")]
22use khive_brain_core::BetaPosterior;
23use khive_brain_core::{
24    ConsumerKind, FeedbackEventKind, ProfileBinding, ProfileLifecycle, ProfileRecord,
25    SectionPosteriorState, SectionType, DEFAULT_ESS_CAP,
26};
27
28// ── Adapter revision sentinel ─────────────────────────────────────────────────
29
30/// Base model revision that registered adapters must target; overridable via
31/// KHIVE_BRAIN_BASE_MODEL_REVISION so both accept and reject paths are testable.
32pub(crate) const DEFAULT_BASE_MODEL_REVISION: &str = "base-v0";
33
34// ── Handler table ─────────────────────────────────────────────────────────────
35
36/// Brain pack verb surface. Visibility::Verb = exposed on the MCP `request` tool.
37/// Visibility::Subhandler = internal / operator-only. Illocutionary classification applied.
38pub(crate) static BRAIN_HANDLERS: &[HandlerDef] = &[
39    // ── Assertive (read) verbs ────────────────────────────────────────────
40    HandlerDef {
41        name: "brain.state",
42        description: "Return current BrainState snapshot for inspection",
43        visibility: khive_types::Visibility::Subhandler,
44        category: khive_types::VerbCategory::Assertive,
45        params: &[],
46    },
47    HandlerDef {
48        name: "brain.config",
49        description: "Return projected config for a named pack parameter",
50        visibility: khive_types::Visibility::Subhandler,
51        category: khive_types::VerbCategory::Assertive,
52        params: &[khive_types::ParamDef {
53            name: "parameter",
54            param_type: "string",
55            required: false,
56            description: "Specific parameter to query: \"recall::relevance_weight\" | \"recall::salience_weight\" | \"recall::temporal_weight\". Omit to return all.",
57        }],
58    },
59    HandlerDef {
60        name: "brain.events",
61        description: "List recent brain-relevant events for debugging",
62        visibility: khive_types::Visibility::Subhandler,
63        category: khive_types::VerbCategory::Assertive,
64        params: &[khive_types::ParamDef {
65            name: "limit",
66            param_type: "integer",
67            required: false,
68            description: "Maximum events to return (default 20, max 100).",
69        }],
70    },
71    HandlerDef {
72        name: "brain.event_counts",
73        description: "Windowed event counts grouped by kind, actor, and verb over the event \
74            plane (ADR-103 Stage 1, #724 Ask A); feedback_explicit events additionally split by \
75            served_by_profile_id; events carrying a work_class (today: phase_started / \
76            phase_completed / phase_cancelled payloads, checked before any future \
77            payload.resource.work_class) split by counts_by_work_class. Events carrying \
78            payload.resource.cost_unit (ADR-103 Amendment 1; stamped on every successful verb \
79            dispatch since PR #927) sum into total_cost_unit and cost_unit_by_verb; events with \
80            no resource.cost_unit (pre-Amendment-1 events, or errored/denied dispatches) simply \
81            do not contribute. Both fields are omitted, not zero-filled, when no event in the \
82            window carries cost_unit. When truncated=true, both sums are over the fetched page \
83            only, same as the other counts_by_* fields.",
84        visibility: khive_types::Visibility::Verb,
85        category: khive_types::VerbCategory::Assertive,
86        params: &[
87            khive_types::ParamDef {
88                name: "since",
89                param_type: "string",
90                required: true,
91                description: "Window start, ISO-8601/RFC-3339 datetime (e.g. \"2026-07-01T00:00:00Z\"). Inclusive.",
92            },
93            khive_types::ParamDef {
94                name: "until",
95                param_type: "string",
96                required: false,
97                description: "Window end, ISO-8601/RFC-3339 datetime. Exclusive. Defaults to now.",
98            },
99            khive_types::ParamDef {
100                name: "actor",
101                param_type: "string",
102                required: false,
103                description: "Filter to a single actor. Stored actor strings are prefixed \
104                    (e.g. \"actor:lambda:khive\"); pass either the bare seat form \
105                    (\"lambda:khive\") or the stored prefixed form — both match. Omit for all \
106                    actors.",
107            },
108            khive_types::ParamDef {
109                name: "kind",
110                param_type: "string",
111                required: false,
112                description: "Filter to a single EventKind (e.g. \"recall_executed\", \"feedback_explicit\"). Omit for all kinds.",
113            },
114        ],
115    },
116    HandlerDef {
117        name: "brain.profiles",
118        description: "List profiles, optionally filtered by lifecycle",
119        visibility: khive_types::Visibility::Verb,
120        category: khive_types::VerbCategory::Assertive,
121        params: &[khive_types::ParamDef {
122            name: "lifecycle",
123            param_type: "string",
124            required: false,
125            description: "Filter profiles by lifecycle state: \"active\" | \"inactive\" | \"archived\". Omit to return all.",
126        }],
127    },
128    HandlerDef {
129        name: "brain.profile",
130        description: "Profile metadata, latest snapshot, current state summary",
131        visibility: khive_types::Visibility::Verb,
132        category: khive_types::VerbCategory::Assertive,
133        params: &[khive_types::ParamDef {
134            name: "profile_id",
135            param_type: "string",
136            required: true,
137            description: "Profile ID string (e.g. \"balanced-recall-v1\"). NOT a UUID — use the string identifier. Alias: id.",
138        }],
139    },
140    HandlerDef {
141        name: "brain.resolve",
142        description: "Show which profile would serve a caller context",
143        visibility: khive_types::Visibility::Verb,
144        category: khive_types::VerbCategory::Assertive,
145        params: &[
146            khive_types::ParamDef {
147                name: "consumer_kind",
148                param_type: "string",
149                required: true,
150                description: "Verb or operation type the caller is about to perform (e.g. \"recall\").",
151            },
152            khive_types::ParamDef {
153                name: "actor",
154                param_type: "string",
155                required: false,
156                description: "Caller actor identifier. Defaults to the caller's dispatch identity; anonymous callers match only wildcard bindings. Pass explicitly to query another identity.",
157            },
158            khive_types::ParamDef {
159                name: "namespace",
160                param_type: "string",
161                required: false,
162                description: "Namespace for binding resolution. Defaults to \"*\" wildcard match.",
163            },
164        ],
165    },
166    // ── Commissive (write state) verbs ────────────────────────────────────
167    HandlerDef {
168        name: "brain.activate",
169        description: "Move a profile to Active (lifecycle transition; serving reads profile state per-request)",
170        visibility: khive_types::Visibility::Verb,
171        category: khive_types::VerbCategory::Commissive,
172        params: &[khive_types::ParamDef {
173            name: "profile_id",
174            param_type: "string",
175            required: true,
176            description: "Profile ID to activate (e.g. \"balanced-recall-v1\").",
177        }],
178    },
179    HandlerDef {
180        name: "brain.deactivate",
181        description: "Move a profile to Inactive (stop live updates, retain state)",
182        visibility: khive_types::Visibility::Verb,
183        category: khive_types::VerbCategory::Commissive,
184        params: &[khive_types::ParamDef {
185            name: "profile_id",
186            param_type: "string",
187            required: true,
188            description: "Profile ID to deactivate.",
189        }],
190    },
191    HandlerDef {
192        name: "brain.archive",
193        description: "Move a profile to Archived (read-only, audit-retained)",
194        visibility: khive_types::Visibility::Verb,
195        category: khive_types::VerbCategory::Declaration,
196        params: &[khive_types::ParamDef {
197            name: "profile_id",
198            param_type: "string",
199            required: true,
200            description: "Profile ID to archive.",
201        }],
202    },
203    HandlerDef {
204        name: "brain.reset",
205        description: "Reset posteriors to priors (preserves event history; increments exploration_epoch, which counts resets and nothing else)",
206        visibility: khive_types::Visibility::Verb,
207        category: khive_types::VerbCategory::Declaration,
208        params: &[khive_types::ParamDef {
209            name: "profile_id",
210            param_type: "string",
211            required: false,
212            description: "Profile ID to reset (must exist and be active). Defaults to \"balanced-recall-v1\". Use brain.profiles() to list profiles.",
213        }],
214    },
215    HandlerDef {
216        name: "brain.feedback",
217        description: "Emit a FeedbackExplicit event into the shared log",
218        visibility: khive_types::Visibility::Verb,
219        category: khive_types::VerbCategory::Commissive,
220        params: &[
221            khive_types::ParamDef {
222                name: "target_id",
223                param_type: "uuid",
224                required: true,
225                description: "UUID of the memory note or entity the feedback applies to.",
226            },
227            khive_types::ParamDef {
228                name: "signal",
229                param_type: "string",
230                required: true,
231                description: "Feedback signal: \"useful\" | \"not_useful\" | \"wrong\" | \"explicit_positive\" | \"explicit_negative\" | \"implicit_positive\" | \"implicit_negative\" | \"correction\".",
232            },
233            khive_types::ParamDef {
234                name: "served_by_profile_id",
235                param_type: "string",
236                required: false,
237                description: "Profile ID that served the result being rated. Recorded in the event payload.",
238            },
239            khive_types::ParamDef {
240                name: "section_signals",
241                param_type: "object",
242                required: false,
243                description: "Per-section feedback signals: {\"section_name\": \"useful\"|\"not_useful\"|\"wrong\"}. For knowledge_compose profiles.",
244            },
245            khive_types::ParamDef {
246                name: "scorer_run_id",
247                param_type: "string",
248                required: false,
249                description: "ADR-081: scorer pass identifier, half of the (scorer_run_id, serve_ledger_id) dedup key. Must be supplied together with serve_ledger_id.",
250            },
251            khive_types::ParamDef {
252                name: "serve_ledger_id",
253                param_type: "string",
254                required: false,
255                description: "ADR-081: id of the brain_serve_ledger row being graded. Must be supplied together with scorer_run_id; backfills the row's grade and gates dedup.",
256            },
257        ],
258    },
259    HandlerDef {
260        name: "brain.auto_feedback",
261        description: "Emit implicit feedback for recall results supplied by an agent. \
262            Convenience verb: agents call this after memory.recall instead of constructing \
263            a brain.feedback call manually. Keeps memory and brain packs decoupled (#517).",
264        visibility: khive_types::Visibility::Verb,
265        category: khive_types::VerbCategory::Commissive,
266        params: &[
267            khive_types::ParamDef {
268                name: "query",
269                param_type: "string",
270                required: true,
271                description: "Recall query that produced the results.",
272            },
273            khive_types::ParamDef {
274                name: "results",
275                param_type: "array",
276                required: true,
277                description: "Recall result objects; the first object's id is credited.",
278            },
279            khive_types::ParamDef {
280                name: "signal",
281                param_type: "string",
282                required: false,
283                description: "Feedback signal. Defaults to \"implicit_positive\".",
284            },
285            khive_types::ParamDef {
286                name: "served_by_profile_id",
287                param_type: "string",
288                required: false,
289                description: "Profile ID that served the recall. Defaults like brain.feedback.",
290            },
291            khive_types::ParamDef {
292                name: "scorer_run_id",
293                param_type: "string",
294                required: false,
295                description: "ADR-081: forwarded verbatim to brain.feedback. Must be supplied together with serve_ledger_id.",
296            },
297            khive_types::ParamDef {
298                name: "serve_ledger_id",
299                param_type: "string",
300                required: false,
301                description: "ADR-081: forwarded verbatim to brain.feedback. Must be supplied together with scorer_run_id.",
302            },
303        ],
304    },
305    HandlerDef {
306        name: "brain.record_serve",
307        description: "ADR-081 §4/§5: append cross-session serve-ledger rows for a batch of \
308            recall targets. Internal-only — memory.recall dispatches this off its response \
309            path after resolving served_by_profile_id via the ADR-035 three-tier discipline.",
310        visibility: khive_types::Visibility::Subhandler,
311        category: khive_types::VerbCategory::Commissive,
312        params: &[
313            khive_types::ParamDef {
314                name: "consumer_kind",
315                param_type: "string",
316                required: true,
317                description: "Consumer kind that served these results, e.g. \"recall\".",
318            },
319            khive_types::ParamDef {
320                name: "served_by_profile_id",
321                param_type: "string",
322                required: false,
323                description: "Profile ID resolved at serve time. Omitted when unresolved.",
324            },
325            khive_types::ParamDef {
326                name: "target_ids",
327                param_type: "array",
328                required: true,
329                description: "Note/entity ids that were served; one ledger row per id.",
330            },
331            khive_types::ParamDef {
332                name: "query_raw",
333                param_type: "string",
334                required: true,
335                description: "Raw query text; query_class is derived from this deterministically.",
336            },
337            khive_types::ParamDef {
338                name: "served_at",
339                param_type: "integer",
340                required: false,
341                description: "Serve timestamp in epoch microseconds. Defaults to now.",
342            },
343        ],
344    },
345    // ── Declaration verbs ─────────────────────────────────────────────────
346    HandlerDef {
347        name: "brain.bind",
348        description: "Write a row in the profile resolution table",
349        visibility: khive_types::Visibility::Verb,
350        category: khive_types::VerbCategory::Declaration,
351        params: &[
352            khive_types::ParamDef {
353                name: "profile_id",
354                param_type: "string",
355                required: true,
356                description: "Profile ID to bind (must exist).",
357            },
358            khive_types::ParamDef {
359                name: "actor",
360                param_type: "string",
361                required: false,
362                description: "Actor identifier to match. Default \"*\" (all actors). Cannot contain \"*\" inside a real value.",
363            },
364            khive_types::ParamDef {
365                name: "namespace",
366                param_type: "string",
367                required: false,
368                description: "Namespace to match. Default \"*\" (all namespaces).",
369            },
370            khive_types::ParamDef {
371                name: "consumer_kind",
372                param_type: "string",
373                required: false,
374                description: "Verb / operation kind to match. Default \"*\" (all kinds).",
375            },
376            khive_types::ParamDef {
377                name: "priority",
378                param_type: "integer",
379                required: false,
380                description: "Binding priority; higher wins when multiple bindings match (default 0).",
381            },
382        ],
383    },
384    HandlerDef {
385        name: "brain.unbind",
386        description: "Remove rows from the profile resolution table. At least one filter (profile_id, actor, namespace, or consumer_kind) is required.",
387        visibility: khive_types::Visibility::Verb,
388        category: khive_types::VerbCategory::Declaration,
389        params: &[
390            khive_types::ParamDef {
391                name: "profile_id",
392                param_type: "string",
393                required: false,
394                description: "Remove bindings for this profile ID. All filters use AND semantics. At least one filter is required.",
395            },
396            khive_types::ParamDef {
397                name: "actor",
398                param_type: "string",
399                required: false,
400                description: "Remove bindings for this actor.",
401            },
402            khive_types::ParamDef {
403                name: "namespace",
404                param_type: "string",
405                required: false,
406                description: "Remove bindings for this namespace.",
407            },
408            khive_types::ParamDef {
409                name: "consumer_kind",
410                param_type: "string",
411                required: false,
412                description: "Remove bindings for this consumer_kind.",
413            },
414        ],
415    },
416    HandlerDef {
417        name: "brain.bindings",
418        description: "List rows in the profile resolution table, optionally filtered",
419        visibility: khive_types::Visibility::Verb,
420        category: khive_types::VerbCategory::Assertive,
421        params: &[
422            khive_types::ParamDef {
423                name: "profile_id",
424                param_type: "string",
425                required: false,
426                description: "Filter bindings by profile ID.",
427            },
428            khive_types::ParamDef {
429                name: "actor",
430                param_type: "string",
431                required: false,
432                description: "Filter bindings by actor.",
433            },
434            khive_types::ParamDef {
435                name: "namespace",
436                param_type: "string",
437                required: false,
438                description: "Filter bindings by namespace.",
439            },
440            khive_types::ParamDef {
441                name: "consumer_kind",
442                param_type: "string",
443                required: false,
444                description: "Filter bindings by consumer_kind.",
445            },
446        ],
447    },
448    HandlerDef {
449        name: "brain.create_profile",
450        description: "Create a new brain profile with given name and optional seed priors",
451        visibility: khive_types::Visibility::Verb,
452        category: khive_types::VerbCategory::Declaration,
453        params: &[
454            khive_types::ParamDef {
455                name: "name",
456                param_type: "string",
457                required: true,
458                description: "Profile ID / name (alphanumeric, hyphens allowed, e.g. \"my-profile-v1\"). Must be unique.",
459            },
460            khive_types::ParamDef {
461                name: "description",
462                param_type: "string",
463                required: false,
464                description: "Human-readable description for this profile.",
465            },
466            khive_types::ParamDef {
467                name: "consumer_kind",
468                param_type: "string",
469                required: false,
470                description: "Operation kind this profile targets (e.g. \"recall\"). Default \"recall\".",
471            },
472            khive_types::ParamDef {
473                name: "seed_priors",
474                param_type: "object",
475                required: false,
476                description: "Seed priors object. For knowledge_compose: {\"section_posteriors\": {\"overview\": {\"alpha\": 2.0, \"beta\": 2.0}, ...}}. For recall: {\"relevance\": {\"alpha\": 7.0, \"beta\": 3.0}, ...}.",
477            },
478        ],
479    },
480    HandlerDef {
481        name: "brain.register_adapter",
482        description: "Register an adapter integrity record so the router only composes \
483            adapters matching the active base model revision",
484        visibility: khive_types::Visibility::Verb,
485        category: khive_types::VerbCategory::Declaration,
486        params: &[
487            khive_types::ParamDef {
488                name: "adapter_id",
489                param_type: "string",
490                required: true,
491                description: "Stable identifier for the adapter (used as the entity name).",
492            },
493            khive_types::ParamDef {
494                name: "content_hash",
495                param_type: "string",
496                required: true,
497                description: "Content hash of the adapter weights for integrity verification.",
498            },
499            khive_types::ParamDef {
500                name: "base_model_revision",
501                param_type: "string",
502                required: true,
503                description: "Base model revision the adapter was trained against. Must match the active revision or registration is rejected.",
504            },
505            khive_types::ParamDef {
506                name: "metadata",
507                param_type: "object",
508                required: false,
509                description: "Optional additional metadata merged into entity properties.",
510            },
511        ],
512    },
513    // ── Legacy / internal ─────────────────────────────────────────────────
514    HandlerDef {
515        name: "brain.emit",
516        description: "Manually emit a feedback event (deprecated; use brain.feedback)",
517        visibility: khive_types::Visibility::Subhandler,
518        category: khive_types::VerbCategory::Commissive,
519        params: &[
520            khive_types::ParamDef {
521                name: "target_id",
522                param_type: "uuid",
523                required: true,
524                description: "UUID of the record the feedback applies to.",
525            },
526            khive_types::ParamDef {
527                name: "signal",
528                param_type: "string",
529                required: true,
530                description: "Feedback signal: \"useful\" | \"not_useful\" | \"wrong\". Deprecated: use brain.feedback instead.",
531            },
532            khive_types::ParamDef {
533                name: "served_by_profile_id",
534                param_type: "string",
535                required: false,
536                description: "Profile ID that served the result.",
537            },
538        ],
539    },
540];
541
542// ── BrainPack handler impl ────────────────────────────────────────────────────
543
544impl BrainPack {
545    // ── brain.state ───────────────────────────────────────────────────────
546
547    pub(crate) async fn handle_state(&self, _params: Value) -> Result<Value, RuntimeError> {
548        let state = self.state.lock().unwrap();
549        let snapshot = state.to_snapshot();
550        serde_json::to_value(&snapshot).map_err(|e| RuntimeError::InvalidInput(e.to_string()))
551    }
552
553    // ── brain.config ──────────────────────────────────────────────────────
554
555    pub(crate) async fn handle_config(&self, params: Value) -> Result<Value, RuntimeError> {
556        #[derive(Deserialize)]
557        #[serde(deny_unknown_fields)]
558        struct ConfigParams {
559            parameter: Option<String>,
560        }
561        let p: ConfigParams = serde_json::from_value(params)
562            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
563
564        let state = self.state.lock().unwrap();
565        let br = &state.balanced_recall;
566
567        let param_map = [
568            ("recall::relevance_weight", &br.relevance),
569            ("recall::salience_weight", &br.salience),
570            ("recall::temporal_weight", &br.temporal),
571        ];
572
573        match p.parameter {
574            Some(key) => {
575                let posterior = param_map
576                    .iter()
577                    .find(|(k, _)| *k == key)
578                    .map(|(_, p)| *p)
579                    .ok_or_else(|| {
580                        RuntimeError::NotFound(format!(
581                            "parameter {key:?}; valid: {}",
582                            param_map
583                                .iter()
584                                .map(|(k, _)| *k)
585                                .collect::<Vec<_>>()
586                                .join(", ")
587                        ))
588                    })?;
589                Ok(json!({
590                    "parameter": key,
591                    "mean": posterior.mean(),
592                    "variance": posterior.variance(),
593                    "ess": posterior.effective_sample_size(),
594                    "alpha": posterior.alpha(),
595                    "beta": posterior.beta(),
596                }))
597            }
598            None => {
599                let configs: serde_json::Map<String, Value> = param_map
600                    .iter()
601                    .map(|(k, p)| {
602                        (
603                            (*k).to_owned(),
604                            json!({
605                                "mean": p.mean(),
606                                "variance": p.variance(),
607                                "ess": p.effective_sample_size(),
608                            }),
609                        )
610                    })
611                    .collect();
612                Ok(Value::Object(configs))
613            }
614        }
615    }
616
617    // ── brain.events ──────────────────────────────────────────────────────
618
619    pub(crate) async fn handle_events(
620        &self,
621        token: &NamespaceToken,
622        params: Value,
623    ) -> Result<Value, RuntimeError> {
624        #[derive(Deserialize)]
625        #[serde(deny_unknown_fields)]
626        struct EventsParams {
627            limit: Option<u32>,
628        }
629        let p: EventsParams = serde_json::from_value(params)
630            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
631
632        let limit = p.limit.unwrap_or(20).min(100);
633        let ns = token.namespace().as_str().to_string();
634
635        let store = self.runtime.events(token)?;
636        let filter = EventFilter {
637            verbs: vec![
638                "recall".into(),
639                "search".into(),
640                "brain.feedback".into(),
641                "brain.emit".into(), // retained for backward-compat queries
642                "get".into(),
643                "remember".into(),
644            ],
645            ..EventFilter::default()
646        };
647        let _ = ns;
648        let page = store
649            .query_events(filter, PageRequest { offset: 0, limit })
650            .await
651            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
652
653        let events: Vec<Value> = page
654            .items
655            .iter()
656            .map(|e| {
657                json!({
658                    "id": e.id.to_string(),
659                    "verb": e.verb,
660                    "outcome": e.outcome,
661                    "target_id": e.target_id.map(|t| t.to_string()),
662                    "duration_us": e.duration_us,
663                    "created_at": micros_to_iso(e.created_at),
664                    "payload": e.payload,
665                })
666            })
667            .collect();
668
669        Ok(json!({
670            "count": events.len(),
671            "events": events,
672        }))
673    }
674
675    // ── brain.event_counts ──────────────────────────────────────────────────
676    //
677    // ADR-103 Stage 1 (#724 Ask A): a windowed, per-actor, per-kind, per-verb
678    // read verb over the event plane, additionally surfacing `work_class`
679    // (ADR-103-resource-attribution-model.md:303) via `counts_by_work_class`.
680    // `work_class` is read from `payload.work_class` first — the shape the
681    // three Phase* events (`PhaseStarted`/`PhaseCompleted`/`PhaseCancelled`,
682    // `khive_storage::telemetry`) use today — falling back to
683    // `payload.resource.work_class` for the not-yet-emitted future shape
684    // where a generic audit row carries a `resource` sub-object (ADR-103
685    // Stage 1's "resource payload enrichment", still open). Events with
686    // neither path present are simply not counted in the split.
687    // `cost_unit` (ADR-103 Amendment 1, `payload.resource.cost_unit`,
688    // stamped on every successful verb dispatch since PR #927 via
689    // `khive_runtime::cost_unit::resource_payload`) sums into
690    // `total_cost_unit` and `cost_unit_by_verb`, both omitted rather than
691    // zero-filled when no event in the window carries it. Amendment 1's
692    // "absence has exactly two meanings" rule means a missing
693    // `resource.cost_unit` is either a pre-Amendment-1 event or an
694    // errored/denied dispatch — this verb does not distinguish the two, it
695    // simply excludes the event from the sum, matching the `work_class`
696    // split's existing exclusion convention.
697    //
698    // Named `brain.event_counts` rather than the literal
699    // `brain.events(...)` shape sketched on the issue — that name is already
700    // taken by the `Visibility::Subhandler` debug-listing verb above, which an
701    // MCP-boundary test (`subhandler_verbs_are_blocked_at_mcp_boundary`)
702    // requires to stay internal-only, and which several existing tests depend
703    // on for its `{count, events}` shape. Reusing the name would either break
704    // that invariant or overload one verb with two incompatible response
705    // shapes; a second, public verb name keeps both surfaces intact.
706
707    /// Cap on events aggregated by a single `brain.event_counts` window query.
708    /// `EventStore` has no SQL-side grouped-count primitive today (`count_events`
709    /// returns one scalar per filter, not a GROUP BY), so this verb fetches a
710    /// bounded page via `query_events` and aggregates by kind/actor/verb/profile
711    /// in the handler. A window wider than this cap still returns counts up to the
712    /// cap and sets `truncated: true` plus the true `window_event_total` rather
713    /// than silently reporting a partial total as complete. At current event
714    /// volumes this bound is not expected to bind in practice; widening the
715    /// storage trait with a grouped-count method was intentionally deferred
716    /// rather than contorting `EventStore` for a case that is not yet load-bearing.
717    const MAX_WINDOW_EVENTS: u32 = 50_000;
718
719    pub(crate) async fn handle_event_counts(
720        &self,
721        token: &NamespaceToken,
722        params: Value,
723    ) -> Result<Value, RuntimeError> {
724        #[derive(Deserialize)]
725        #[serde(deny_unknown_fields)]
726        struct EventCountsParams {
727            actor: Option<String>,
728            kind: Option<String>,
729            // `Option`, not a required `String`: a bare-missing `since` must go through
730            // the same named-field-plus-example-format error as a malformed one, not
731            // serde's generic "missing field `since`" message.
732            since: Option<String>,
733            until: Option<String>,
734        }
735        let p: EventCountsParams = serde_json::from_value(params)
736            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
737
738        let since_raw = p.since.as_deref().ok_or_else(|| {
739            RuntimeError::InvalidInput(
740                "missing `since`: required ISO-8601/RFC-3339 datetime; expected e.g. \
741                 \"2026-07-01T00:00:00Z\""
742                    .to_string(),
743            )
744        })?;
745        let since_us = parse_rfc3339_micros("since", since_raw)?;
746        let until_us = match p.until.as_deref() {
747            Some(u) => parse_rfc3339_micros("until", u)?,
748            None => Utc::now().timestamp_micros(),
749        };
750
751        let kind_filter = match p.kind.as_deref() {
752            Some(k) => Some(
753                k.parse::<khive_types::EventKind>()
754                    .map_err(|e| RuntimeError::InvalidInput(format!("invalid `kind`: {e}")))?,
755            ),
756            None => None,
757        };
758
759        // Stored actor strings are prefixed (`actor:<kind>:<id>`). Callers naturally pass the
760        // bare seat form (e.g. "lambda:khive"), which would silently match nothing against an
761        // exact-match filter. Match either spelling by expanding the filter to both forms —
762        // `EventFilter.actors` is an IN-list, so this is a pure OR, never a guess. A caller who
763        // already passes the stored `actor:`-prefixed form keeps exact-match behavior.
764        let actor_filters: Vec<String> = match p.actor.as_deref() {
765            Some(a) if a.starts_with("actor:") => vec![a.to_string()],
766            Some(a) => vec![a.to_string(), format!("actor:{a}")],
767            None => Vec::new(),
768        };
769
770        let store = self.runtime.events(token)?;
771        let filter = EventFilter {
772            actors: actor_filters,
773            kinds: kind_filter.into_iter().collect(),
774            // Half-open window [since, until): `EventFilter.after` is a strict
775            // `created_at > after`, so subtracting one microsecond from `since`
776            // makes the boundary itself inclusive; `before` is already a strict
777            // `created_at < before`, which is exactly the exclusive `until` we want.
778            after: Some(since_us - 1),
779            before: Some(until_us),
780            ..EventFilter::default()
781        };
782
783        let page = store
784            .query_events(
785                filter,
786                PageRequest {
787                    offset: 0,
788                    limit: Self::MAX_WINDOW_EVENTS,
789                },
790            )
791            .await
792            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
793
794        let window_event_total = page.total.unwrap_or(page.items.len() as u64);
795        let truncated = window_event_total > page.items.len() as u64;
796
797        let mut counts_by_kind: std::collections::BTreeMap<String, u64> =
798            std::collections::BTreeMap::new();
799        let mut counts_by_actor: std::collections::BTreeMap<String, u64> =
800            std::collections::BTreeMap::new();
801        let mut counts_by_verb: std::collections::BTreeMap<String, u64> =
802            std::collections::BTreeMap::new();
803        let mut by_profile: std::collections::BTreeMap<String, u64> =
804            std::collections::BTreeMap::new();
805        let mut counts_by_work_class: std::collections::BTreeMap<String, u64> =
806            std::collections::BTreeMap::new();
807        let mut total_cost_unit: i64 = 0;
808        let mut cost_unit_by_verb: std::collections::BTreeMap<String, i64> =
809            std::collections::BTreeMap::new();
810
811        for event in &page.items {
812            *counts_by_kind
813                .entry(event.kind.name().to_string())
814                .or_insert(0) += 1;
815            *counts_by_actor.entry(event.actor.clone()).or_insert(0) += 1;
816            *counts_by_verb.entry(event.verb.clone()).or_insert(0) += 1;
817            if event.kind == khive_types::EventKind::FeedbackExplicit {
818                let profile = event
819                    .payload
820                    .get("served_by_profile_id")
821                    .and_then(Value::as_str)
822                    .unwrap_or("unspecified")
823                    .to_string();
824                *by_profile.entry(profile).or_insert(0) += 1;
825            }
826            if let Some(work_class) = event_work_class(&event.payload) {
827                *counts_by_work_class
828                    .entry(work_class.to_string())
829                    .or_insert(0) += 1;
830            }
831            if let Some(cost_unit) = event_cost_unit(&event.payload) {
832                total_cost_unit = total_cost_unit.saturating_add(cost_unit);
833                let entry = cost_unit_by_verb.entry(event.verb.clone()).or_insert(0);
834                *entry = entry.saturating_add(cost_unit);
835            }
836        }
837
838        let mut result = json!({
839            "since": micros_to_iso(since_us),
840            "until": micros_to_iso(until_us),
841            "actor": p.actor,
842            "kind": p.kind,
843            "total": page.items.len() as u64,
844            "counts_by_kind": counts_by_kind,
845            "counts_by_actor": counts_by_actor,
846            "counts_by_verb": counts_by_verb,
847            "truncated": truncated,
848            "window_event_total": window_event_total,
849        });
850        if !by_profile.is_empty() {
851            result["by_profile"] = json!(by_profile);
852        }
853        if !counts_by_work_class.is_empty() {
854            result["counts_by_work_class"] = json!(counts_by_work_class);
855        }
856        if !cost_unit_by_verb.is_empty() {
857            result["total_cost_unit"] = json!(total_cost_unit);
858            result["cost_unit_by_verb"] = json!(cost_unit_by_verb);
859        }
860        Ok(result)
861    }
862
863    // ── brain.profiles ────────────────────────────────────────────────────
864
865    pub(crate) async fn handle_profiles(&self, params: Value) -> Result<Value, RuntimeError> {
866        #[derive(Deserialize)]
867        #[serde(deny_unknown_fields)]
868        struct ProfilesParams {
869            lifecycle: Option<String>,
870        }
871        let p: ProfilesParams = serde_json::from_value(params)
872            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
873
874        let state = self.state.lock().unwrap();
875        // Only expose the three public-facing lifecycle states.
876        // `defined` and `registered` are internal implementation states that
877        // callers cannot filter on; listing them in the error would confuse users.
878        let filter_lc: Option<ProfileLifecycle> = p
879            .lifecycle
880            .as_deref()
881            .map(|s| match s {
882                "active" => Ok(ProfileLifecycle::Active),
883                "inactive" => Ok(ProfileLifecycle::Inactive),
884                "archived" => Ok(ProfileLifecycle::Archived),
885                other => Err(RuntimeError::InvalidInput(format!(
886                    "invalid lifecycle {other:?}; expected one of 'active', 'inactive', 'archived'"
887                ))),
888            })
889            .transpose()?;
890
891        // Sort by id for deterministic output regardless of HashMap iteration order.
892        let mut profiles: Vec<&ProfileRecord> = state
893            .profiles
894            .values()
895            .filter(|r| filter_lc.as_ref().is_none_or(|lc| &r.lifecycle == lc))
896            .collect();
897        profiles.sort_by(|a, b| a.id.cmp(&b.id));
898
899        let items: Vec<Value> = profiles
900            .iter()
901            .map(|r| {
902                json!({
903                    "id": r.id,
904                    "description": r.description,
905                    "consumer_kind": r.consumer_kind,
906                    "state_class": r.state_class,
907                    "lifecycle": r.lifecycle,
908                    "total_events": r.total_events,
909                    "exploration_epoch": r.exploration_epoch,
910                    "created_at": r.created_at,
911                })
912            })
913            .collect();
914
915        Ok(json!({ "count": items.len(), "profiles": items }))
916    }
917
918    // ── brain.profile ─────────────────────────────────────────────────────
919
920    pub(crate) async fn handle_profile(&self, params: Value) -> Result<Value, RuntimeError> {
921        // Accept both `profile_id` (canonical) and `id` (legacy alias).
922        #[derive(Deserialize)]
923        #[serde(deny_unknown_fields)]
924        struct ProfileParams {
925            profile_id: Option<String>,
926            id: Option<String>,
927        }
928        let p: ProfileParams = serde_json::from_value(params)
929            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
930        let profile_id = p
931            .profile_id
932            .or(p.id)
933            .ok_or_else(|| RuntimeError::InvalidInput("missing field `profile_id`".into()))?;
934
935        let state = self.state.lock().unwrap();
936        let record = state
937            .profiles
938            .get(&profile_id)
939            .ok_or_else(|| RuntimeError::NotFound(format!("profile {:?}", profile_id)))?;
940
941        // Build per-section posterior summary for the response.
942        let section_summary = if let Some(ss) = state.section_states.get(&profile_id) {
943            let weights = derive_deterministic_weights(ss);
944            let mut sections_json: serde_json::Map<String, Value> =
945                serde_json::Map::with_capacity(ss.posteriors.len());
946            for (section, posterior) in &ss.posteriors {
947                let w = weights.get(section).copied().unwrap_or(0.0);
948                sections_json.insert(
949                    section.as_str().to_owned(),
950                    json!({
951                        "alpha": posterior.alpha(),
952                        "beta": posterior.beta(),
953                        "mean": posterior.mean(),
954                        "variance": posterior.variance(),
955                        "ess": posterior.effective_sample_size(),
956                        "weight": w,
957                    }),
958                );
959            }
960            Value::Object(sections_json)
961        } else {
962            Value::Null
963        };
964
965        Ok(json!({
966            "id": record.id,
967            "description": record.description,
968            "consumer_kind": record.consumer_kind,
969            "state_class": record.state_class,
970            "lifecycle": record.lifecycle,
971            "total_events": record.total_events,
972            "exploration_epoch": record.exploration_epoch,
973            "created_at": record.created_at,
974            "state_snapshot": record.state_snapshot,
975            "section_posteriors": section_summary,
976        }))
977    }
978
979    // ── brain.resolve ─────────────────────────────────────────────────────
980
981    pub(crate) async fn handle_resolve(
982        &self,
983        token: &NamespaceToken,
984        params: Value,
985    ) -> Result<Value, RuntimeError> {
986        #[derive(Deserialize)]
987        #[serde(deny_unknown_fields)]
988        struct ResolveParams {
989            actor: Option<String>,
990            namespace: Option<String>,
991            consumer_kind: String,
992        }
993        let p: ResolveParams = serde_json::from_value(params)
994            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
995
996        // #741: an omitted `actor` defaults to the caller's dispatch identity so
997        // this introspection verb reports what the serve path (#708) actually
998        // does. Anonymous callers stay `None` and match only wildcard bindings;
999        // an explicit `actor` param wins so evaluation tooling can query other
1000        // identities.
1001        let actor = match p.actor.as_deref() {
1002            Some(a) => Some(a),
1003            None => token.actor().binding_id(),
1004        };
1005
1006        let state = self.state.lock().unwrap();
1007        // Return requested_consumer_kind + matched_consumer_kind + matched_binding so
1008        // callers can distinguish a real binding match from a system-default fallback.
1009        // ADR-035 tier-2 requires matched_binding=true; a false value means tier-3 fires.
1010        match state.resolve_with_match(actor, p.namespace.as_deref(), &p.consumer_kind) {
1011            Some((record, matched_kind, matched_binding)) => Ok(json!({
1012                "resolved_profile_id": record.id,
1013                "lifecycle": record.lifecycle,
1014                "requested_consumer_kind": p.consumer_kind,
1015                "matched_consumer_kind": matched_kind,
1016                "matched_binding": matched_binding,
1017            })),
1018            None => Err(RuntimeError::NotFound(format!(
1019                "no profile resolved for consumer_kind={:?}",
1020                p.consumer_kind
1021            ))),
1022        }
1023    }
1024
1025    // ── brain.activate / deactivate / archive ─────────────────────────────
1026
1027    pub(crate) async fn handle_activate(
1028        &self,
1029        token: &NamespaceToken,
1030        params: Value,
1031    ) -> Result<Value, RuntimeError> {
1032        self.set_lifecycle(token, params, ProfileLifecycle::Active)
1033            .await
1034    }
1035
1036    pub(crate) async fn handle_deactivate(
1037        &self,
1038        token: &NamespaceToken,
1039        params: Value,
1040    ) -> Result<Value, RuntimeError> {
1041        self.set_lifecycle(token, params, ProfileLifecycle::Inactive)
1042            .await
1043    }
1044
1045    pub(crate) async fn handle_archive(
1046        &self,
1047        token: &NamespaceToken,
1048        params: Value,
1049    ) -> Result<Value, RuntimeError> {
1050        self.set_lifecycle(token, params, ProfileLifecycle::Archived)
1051            .await
1052    }
1053
1054    /// Route a profile lifecycle transition through the durable-write helper
1055    /// (issue #457): the transition is validated and applied against a
1056    /// proposed state copy, and only takes effect in `self.state` after the
1057    /// brain event-log append + snapshot upsert commit in one transaction.
1058    pub(crate) async fn set_lifecycle(
1059        &self,
1060        token: &NamespaceToken,
1061        params: Value,
1062        lifecycle: ProfileLifecycle,
1063    ) -> Result<Value, RuntimeError> {
1064        #[derive(Deserialize)]
1065        #[serde(deny_unknown_fields)]
1066        struct LifecycleParams {
1067            profile_id: String,
1068        }
1069        let p: LifecycleParams = serde_json::from_value(params)
1070            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
1071
1072        let event_kind = match lifecycle {
1073            ProfileLifecycle::Active => "brain.activate",
1074            ProfileLifecycle::Inactive => "brain.deactivate",
1075            ProfileLifecycle::Archived => "brain.archive",
1076            // set_lifecycle is only ever invoked (via handle_activate /
1077            // handle_deactivate / handle_archive) with one of the three
1078            // targets above; other lifecycle values are registry-only states
1079            // that no handler transitions to here.
1080            ProfileLifecycle::Defined | ProfileLifecycle::Registered => "brain.set_lifecycle",
1081        };
1082        let profile_id = p.profile_id;
1083        let event_payload = json!({ "profile_id": profile_id, "lifecycle": lifecycle.clone() });
1084
1085        crate::persist::persist_brain_state_mutation(
1086            self.runtime.sql().as_ref(),
1087            token,
1088            &self.persistence,
1089            &self.state,
1090            crate::persist::BrainMutationEvent {
1091                profile_id: profile_id.clone(),
1092                event_kind: event_kind.to_string(),
1093                payload: event_payload,
1094            },
1095            ENTITY_CACHE_CAPACITY,
1096            move |state: &mut khive_brain_core::BrainState| -> Result<Value, RuntimeError> {
1097                let record = state
1098                    .profiles
1099                    .get_mut(&profile_id)
1100                    .ok_or_else(|| RuntimeError::NotFound(format!("profile {:?}", profile_id)))?;
1101
1102                // Lifecycle DAG: defined → registered → active ⟷ inactive → archived
1103                // Terminal state: archived is read-only/audit-retained.
1104                // No transition OUT of archived is legal.
1105                // Active → archived is also illegal; must deactivate first.
1106                match (&record.lifecycle, &lifecycle) {
1107                    // archived is terminal — nothing leaves it
1108                    (ProfileLifecycle::Archived, _) => {
1109                        return Err(RuntimeError::InvalidInput(format!(
1110                            "profile {:?} is archived; archive is terminal and no transition is permitted",
1111                            profile_id
1112                        )));
1113                    }
1114                    // active → archived is illegal (must go through inactive first)
1115                    (ProfileLifecycle::Active, ProfileLifecycle::Archived) => {
1116                        return Err(RuntimeError::InvalidInput(format!(
1117                            "profile {:?} is active; deactivate it before archiving",
1118                            profile_id
1119                        )));
1120                    }
1121                    // all other transitions are permitted
1122                    _ => {}
1123                }
1124
1125                record.lifecycle = lifecycle.clone();
1126                Ok(json!({
1127                    "profile_id": profile_id,
1128                    "lifecycle": lifecycle,
1129                }))
1130            },
1131        )
1132        .await
1133    }
1134
1135    // ── brain.reset ───────────────────────────────────────────────────────
1136
1137    pub(crate) async fn handle_reset(
1138        &self,
1139        token: &NamespaceToken,
1140        params: Value,
1141    ) -> Result<Value, RuntimeError> {
1142        // profile_id is optional; defaults to "balanced-recall-v1".
1143        #[derive(Deserialize)]
1144        #[serde(deny_unknown_fields)]
1145        struct ResetParams {
1146            profile_id: Option<String>,
1147        }
1148        let p: ResetParams = serde_json::from_value(params)
1149            .map_err(|e| RuntimeError::InvalidInput(format!("brain.reset: {e}")))?;
1150        let profile_id = p
1151            .profile_id
1152            .unwrap_or_else(|| "balanced-recall-v1".to_string());
1153        let event_payload = json!({ "profile_id": profile_id });
1154
1155        // Route through the durable-write helper (issue #457): reset is
1156        // applied to a proposed state copy and only takes effect after the
1157        // brain event-log append + snapshot upsert commit.
1158        crate::persist::persist_brain_state_mutation(
1159            self.runtime.sql().as_ref(),
1160            token,
1161            &self.persistence,
1162            &self.state,
1163            crate::persist::BrainMutationEvent {
1164                profile_id: profile_id.clone(),
1165                event_kind: "brain.reset".to_string(),
1166                payload: event_payload,
1167            },
1168            ENTITY_CACHE_CAPACITY,
1169            move |state: &mut khive_brain_core::BrainState| -> Result<Value, RuntimeError> {
1170                // Validate profile exists.
1171                let lifecycle = state
1172                    .profiles
1173                    .get(&profile_id)
1174                    .ok_or_else(|| RuntimeError::NotFound(format!("profile {:?}", profile_id)))?
1175                    .lifecycle
1176                    .clone();
1177
1178                // Reject archived profiles — archive is terminal and read-only.
1179                if lifecycle == ProfileLifecycle::Archived {
1180                    return Err(RuntimeError::InvalidInput(format!(
1181                        "profile {:?} is archived; archive is terminal and reset is not permitted",
1182                        profile_id
1183                    )));
1184                }
1185
1186                if profile_id == "balanced-recall-v1" {
1187                    state.reset_posteriors();
1188                    // Sync profile record after reset so brain.profile reflects the restored
1189                    // domain-informed priors, not stale pre-reset values.
1190                    sync_balanced_recall_record(state);
1191                } else if state.profile_states.contains_key(&profile_id) {
1192                    // User-created Bayesian profile — reset its own posteriors.
1193                    state.reset_profile_posteriors(&profile_id);
1194                } else {
1195                    // Profile exists in registry but has no live state (e.g. non-Bayesian).
1196                    // Increment exploration_epoch on the record to mark the reset event.
1197                    if let Some(record) = state.profiles.get_mut(&profile_id) {
1198                        record.exploration_epoch += 1;
1199                    }
1200                }
1201
1202                let epoch = if profile_id == "balanced-recall-v1" {
1203                    state.balanced_recall.exploration_epoch
1204                } else {
1205                    state.profiles[&profile_id].exploration_epoch
1206                };
1207
1208                Ok(json!({
1209                    "reset": true,
1210                    "profile_id": profile_id,
1211                    "exploration_epoch": epoch,
1212                }))
1213            },
1214        )
1215        .await
1216    }
1217
1218    /// Resolve the effective serving profile for feedback attribution
1219    /// (ADR-035 tiers 1-2, #697): explicit `served_by_profile_id` wins outright;
1220    /// otherwise resolve an actor+namespace-scoped binding via the same
1221    /// `resolve_with_match` table `brain.resolve` uses, before falling back to
1222    /// the system default. `brain.auto_feedback` forwards into `handle_feedback`
1223    /// unresolved so it inherits this without duplicating the logic.
1224    ///
1225    /// Consumer kind is fixed to `recall`: feedback routed directly through
1226    /// `brain.feedback`/`brain.auto_feedback` (as opposed to `memory.feedback`,
1227    /// which resolves its own tier and passes an explicit profile) always
1228    /// originates from the recall serve loop, so it must resolve against the
1229    /// same binding bucket the serve path used.
1230    /// Returns the profile id plus a resolution marker (`explicit` |
1231    /// `binding` | `default`) so event payloads can record call-site
1232    /// discipline separately from attribution (#1016).
1233    fn resolve_effective_feedback_profile(
1234        &self,
1235        token: &NamespaceToken,
1236        explicit: Option<&str>,
1237    ) -> (String, &'static str) {
1238        if let Some(profile_id) = explicit {
1239            return (profile_id.to_string(), "explicit");
1240        }
1241        let actor = token.actor().binding_id();
1242        let namespace = token.namespace().as_str();
1243        let state = self.state.lock().unwrap();
1244        match state.resolve_with_match(actor, Some(namespace), ConsumerKind::Recall.as_str()) {
1245            Some((record, _matched_kind, true)) => (record.id.clone(), "binding"),
1246            _ => ("balanced-recall-v1".to_string(), "default"),
1247        }
1248    }
1249
1250    // ── brain.feedback ────────────────────────────────────────────────────
1251
1252    pub(crate) async fn handle_feedback(
1253        &self,
1254        token: &NamespaceToken,
1255        params: Value,
1256    ) -> Result<Value, RuntimeError> {
1257        let feedback_start = Instant::now();
1258
1259        #[derive(Deserialize)]
1260        #[serde(deny_unknown_fields)]
1261        struct FeedbackParams {
1262            target_id: String,
1263            signal: String,
1264            served_by_profile_id: Option<String>,
1265            section_signals: Option<serde_json::Value>,
1266            // ADR-081 §6: scorer provenance, additive and optional. Must be
1267            // supplied together (validated just below) — one without the other
1268            // is rejected rather than silently coerced.
1269            scorer_run_id: Option<String>,
1270            serve_ledger_id: Option<String>,
1271        }
1272        let p: FeedbackParams = serde_json::from_value(params)
1273            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
1274
1275        // ADR-081 §6: together-or-rejected. Absent-both is the unchanged,
1276        // ordinary feedback path; present-both enables dedup + ledger backfill
1277        // below. Exactly one present is invalid input.
1278        if p.scorer_run_id.is_some() != p.serve_ledger_id.is_some() {
1279            return Err(RuntimeError::InvalidInput(
1280                "scorer_run_id and serve_ledger_id must be supplied together".to_string(),
1281            ));
1282        }
1283
1284        let target: uuid::Uuid =
1285            resolve_auto_feedback_target(&self.runtime, token, &p.target_id).await?;
1286
1287        let signal = match p.signal.as_str() {
1288            "useful" => "useful",
1289            "not_useful" => "not_useful",
1290            "wrong" => "wrong",
1291            // Semantic event taxonomy names are also valid.
1292            "explicit_positive" => "explicit_positive",
1293            "explicit_negative" => "explicit_negative",
1294            "implicit_positive" => "implicit_positive",
1295            "implicit_negative" => "implicit_negative",
1296            "correction" => "correction",
1297            other => {
1298                return Err(RuntimeError::InvalidInput(format!(
1299                    "unknown signal {other:?}; valid: useful | not_useful | wrong | \
1300                     explicit_positive | explicit_negative | implicit_positive | \
1301                     implicit_negative | correction"
1302                )))
1303            }
1304        };
1305
1306        // Resolve the target by UUID with no namespace filter (ADR-007 Rule 2 /
1307        // PR-A1: by-ID ops are namespace-agnostic; authorization is the Gate's,
1308        // not a post-fetch namespace check). Rule 3b recall fans out actor-stamped
1309        // memories from other namespaces by design, so a primary-only check would
1310        // reject the feedback loop's own recalled targets. NotFound only for a
1311        // genuinely absent UUID.
1312        use khive_runtime::Resolved;
1313        // ADR-041 permits both entity and note signal targets; the resolved
1314        // substrate is threaded onto the emitted event below (#831) so the decoder can tell entity and
1315        // note signal observations apart instead of hard-coding entity.
1316        let target_substrate = match self
1317            .runtime
1318            .resolve_by_id(token, target)
1319            .await
1320            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?
1321        {
1322            Some(Resolved::Entity(_)) => khive_types::SubstrateKind::Entity,
1323            Some(Resolved::Note(_)) => khive_types::SubstrateKind::Note,
1324            _ => {
1325                return Err(RuntimeError::NotFound(format!(
1326                    "target_id {target:?} not found"
1327                )));
1328            }
1329        };
1330
1331        // Compute the effective serving profile (explicit, else a matching
1332        // actor+namespace binding, else the system default — #697), then
1333        // validate that it exists in the registry and is not Archived.
1334        let (effective_profile, profile_resolution) =
1335            self.resolve_effective_feedback_profile(token, p.served_by_profile_id.as_deref());
1336        let effective_profile = effective_profile.as_str();
1337        {
1338            let state = self.state.lock().unwrap();
1339            match state.profiles.get(effective_profile) {
1340                None => {
1341                    return Err(RuntimeError::NotFound(format!(
1342                        "serving profile {:?} not found in profile registry",
1343                        effective_profile
1344                    )));
1345                }
1346                Some(rec) if rec.lifecycle == khive_brain_core::ProfileLifecycle::Archived => {
1347                    return Err(RuntimeError::InvalidInput(format!(
1348                        "serving profile {:?} is archived; feedback cannot credit archived profiles",
1349                        effective_profile
1350                    )));
1351                }
1352                Some(_) => {}
1353            }
1354        }
1355
1356        // Validate section_signals up front using the shared validator.
1357        // Malformed input is rejected here rather than silently dropped during replay.
1358        if let Some(ref ss) = p.section_signals {
1359            crate::validate_section_signals(ss)?;
1360        }
1361
1362        let sql = self.runtime.sql();
1363        let now_us = Utc::now().timestamp_micros();
1364
1365        // ADR-081 §6: when both scorer_run_id and serve_ledger_id are supplied,
1366        // resolve dedup + the accounting_profile_id fail-safe against the serve
1367        // ledger before the fold gate runs.
1368        let mut forced_zero_weight = false;
1369        if let (Some(scorer_run_id), Some(serve_ledger_id)) =
1370            (p.scorer_run_id.as_deref(), p.serve_ledger_id.as_deref())
1371        {
1372            match crate::serve_ledger::resolve(sql.as_ref(), serve_ledger_id, scorer_run_id).await?
1373            {
1374                crate::serve_ledger::ServeLedgerResolution::AlreadyGraded => {
1375                    return Ok(json!({
1376                        "emitted": false,
1377                        "deduped": true,
1378                        "verb": "brain.feedback",
1379                        "signal": signal,
1380                        "target_id": target.to_string(),
1381                        "serve_ledger_id": serve_ledger_id,
1382                        "scorer_run_id": scorer_run_id,
1383                    }));
1384                }
1385                crate::serve_ledger::ServeLedgerResolution::NotFound => {
1386                    return Err(RuntimeError::NotFound(format!(
1387                        "serve_ledger_id {:?} not found",
1388                        serve_ledger_id
1389                    )));
1390                }
1391                crate::serve_ledger::ServeLedgerResolution::Proceed {
1392                    accounting_profile_id,
1393                } => {
1394                    // ADR-081 §4 fail-safe: an implicit event whose serve row has
1395                    // no resolvable profile is recorded at zero weight — never
1396                    // folded under a guessed profile.
1397                    if accounting_profile_id.is_none() {
1398                        forced_zero_weight = true;
1399                    }
1400                }
1401            }
1402        }
1403
1404        // ADR-081 §2/§6: when both scorer fields are present, the dedup claim
1405        // is made atomically inside the SAME `BEGIN IMMEDIATE` transaction as
1406        // the fold gate's mass check-and-write AND
1407        // the durable feedback event append (fold_gate.rs) — the `resolve`
1408        // check above is a non-atomic fast path only (it still handles the
1409        // common sequential case and the NotFound / forced-zero-weight
1410        // determination); this is the authoritative correctness mechanism
1411        // under concurrent duplicate submissions.
1412        let dedup_key: Option<(&str, &str)> =
1413            match (p.scorer_run_id.as_deref(), p.serve_ledger_id.as_deref()) {
1414                (Some(r), Some(l)) => Some((r, l)),
1415                _ => None,
1416            };
1417
1418        // Base feedback payload, shared by every signal kind. The gated
1419        // (implicit) path below adds a "gate" key inside the atomic unit;
1420        // the ungated path (explicit/correction) adds nothing further.
1421        // #1016: always stamp the resolved profile (not just the
1422        // caller-explicit value) so event_counts attribution and ADR-032
1423        // replay match what the posterior fold actually credited; the
1424        // marker records how it was resolved.
1425        let mut base_data = json!({
1426            "signal": signal,
1427            "served_by_profile_id": effective_profile,
1428            "profile_resolution": profile_resolution,
1429        });
1430        if let Some(ref ss) = p.section_signals {
1431            base_data["section_signals"] = ss.clone();
1432        }
1433        if let (Some(ref scorer_run_id), Some(ref serve_ledger_id)) =
1434            (p.scorer_run_id.as_ref(), p.serve_ledger_id.as_ref())
1435        {
1436            base_data["scorer_run_id"] = json!(scorer_run_id);
1437            base_data["serve_ledger_id"] = json!(serve_ledger_id);
1438        }
1439
1440        // ADR-081 §2: the bounded-mass fold gate applies only to implicit
1441        // signals. Explicit/correction signals are never gated (they are the
1442        // clamp's own comparator, ADR-081 §1) and — ADR-081 §6 — need not
1443        // join the event append into the fold transaction: there is no
1444        // dedup claim to keep consistent for them, so their append path
1445        // below is unchanged from before this fix.
1446        let is_gated_implicit = matches!(
1447            FeedbackEventKind::from_signal_str(signal),
1448            Some(FeedbackEventKind::ImplicitPositive) | Some(FeedbackEventKind::ImplicitNegative)
1449        );
1450
1451        let event = if is_gated_implicit {
1452            let nominal_weight = FeedbackEventKind::from_signal_str(signal)
1453                .expect("is_gated_implicit implies from_signal_str is Some")
1454                .update_weight();
1455            // The forced-zero fail-safe path now
1456            // runs through the SAME atomic claim+append unit as the nominal
1457            // path below — only the mass fold write itself is skipped — so
1458            // it participates in the dedup claim instead of bypassing it.
1459            let gate_mode = if forced_zero_weight {
1460                crate::fold_gate::FeedbackGateMode::ForcedZero
1461            } else {
1462                crate::fold_gate::FeedbackGateMode::Nominal(nominal_weight)
1463            };
1464
1465            let namespace = token.namespace().as_str().to_string();
1466            // ADR-096 per-request identity: stamp the resolved caller actor,
1467            // not a hardcoded pack name — matches the `kind:id` convention
1468            // the generic Audit event already uses (khive-runtime pack.rs
1469            // `build_audit_storage_event`). `ActorRef::anonymous()` resolves
1470            // to the explicit `"anonymous:local"` string rather than
1471            // silently mislabeling the event, so unresolved-actor calls are
1472            // still distinguishable from configured caller attribution.
1473            let actor_label = format!("{}:{}", token.actor().kind, token.actor().id);
1474            // `apply_fold_gate_and_append_event`'s `build_event` closure is
1475            // now required to be `'static` (ADR-067 Component A, Fork C
1476            // slice 2 — it is boxed into an `AtomicUnitOp` and may run
1477            // inside the writer task's `spawn_blocking`), so it must own
1478            // its captures rather than borrow `namespace`/`base_data` — a
1479            // separate `namespace_for_event` clone avoids conflicting with
1480            // the `&namespace` borrow passed as this call's own argument.
1481            let namespace_for_event = namespace.clone();
1482            let actor_label_for_event = actor_label.clone();
1483            let base_data_for_event = base_data.clone();
1484            let outcome = crate::fold_gate::apply_fold_gate_and_append_event(
1485                sql.as_ref(),
1486                &namespace,
1487                effective_profile,
1488                &target.to_string(),
1489                gate_mode,
1490                now_us,
1491                dedup_key,
1492                move |fold_outcome, forced_zero| {
1493                    let mut data = base_data_for_event;
1494                    let (effective_weight, mass_before, mass_after) = match fold_outcome {
1495                        Some(o) => (o.effective_weight, o.mass_before, o.mass_after),
1496                        None => (0.0, 0.0, 0.0),
1497                    };
1498                    data["gate"] = json!({
1499                        "effective_weight": effective_weight,
1500                        "mass_before": mass_before,
1501                        "mass_after": mass_after,
1502                        "forced_zero_weight": forced_zero,
1503                    });
1504                    let duration_us = feedback_start.elapsed().as_micros().max(1) as i64;
1505                    Event::new(
1506                        namespace_for_event,
1507                        "brain.feedback",
1508                        khive_types::EventKind::FeedbackExplicit,
1509                        target_substrate,
1510                        actor_label_for_event,
1511                    )
1512                    .with_target(target)
1513                    .with_payload(data)
1514                    .with_duration_us(duration_us)
1515                },
1516            )
1517            .await?;
1518
1519            match outcome {
1520                // This is now the ONLY place a scorer-tagged
1521                // implicit call returns `deduped` — reached only when the
1522                // atomic unit's claim conflicted, meaning either a prior call
1523                // already committed claim+fold+event together, or (never)
1524                // a partially-failed prior attempt, since a failed attempt
1525                // rolls back its claim too.
1526                crate::fold_gate::GateAndAppendOutcome::Deduped => {
1527                    return Ok(json!({
1528                        "emitted": false,
1529                        "deduped": true,
1530                        "verb": "brain.feedback",
1531                        "signal": signal,
1532                        "target_id": target.to_string(),
1533                        "serve_ledger_id": p.serve_ledger_id,
1534                        "scorer_run_id": p.scorer_run_id,
1535                    }));
1536                }
1537                crate::fold_gate::GateAndAppendOutcome::Applied(result) => result.event,
1538            }
1539        } else {
1540            // Unchanged: explicit/correction signals (and non-scorer implicit
1541            // feedback would also land here if it ever reached this branch,
1542            // but `is_gated_implicit` already routes all implicit signals
1543            // above) append through the ordinary `EventStore` path, in its
1544            // own transaction — ADR-081 §6 does not require these to join
1545            // the fold gate's atomic unit.
1546            let duration_us = feedback_start.elapsed().as_micros().max(1) as i64;
1547            let event = Event::new(
1548                token.namespace().as_str().to_string(),
1549                "brain.feedback",
1550                khive_types::EventKind::FeedbackExplicit,
1551                target_substrate,
1552                format!("{}:{}", token.actor().kind, token.actor().id),
1553            )
1554            .with_target(target)
1555            .with_payload(base_data)
1556            .with_duration_us(duration_us);
1557
1558            let store = self.runtime.events(token)?;
1559            store
1560                .append_event(event.clone())
1561                .await
1562                .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
1563            event
1564        };
1565
1566        let serving_profile_owned = effective_profile.to_string();
1567
1568        let brain_signal = interpret(&event);
1569
1570        // Issue #458: apply the posterior mutation and make its durability
1571        // part of the success contract. `persist_brain_state_mutation` runs
1572        // this closure against a *proposed* state copy, then commits the
1573        // brain event-log append + snapshot upsert in one transaction —
1574        // `self.state` is only replaced with the proposed copy after that
1575        // commit succeeds. If persistence fails, this returns `Err` and
1576        // `self.state` is left completely untouched (no phantom in-memory
1577        // posterior update that vanishes on restart).
1578        crate::persist::persist_brain_state_mutation(
1579            self.runtime.sql().as_ref(),
1580            token,
1581            &self.persistence,
1582            &self.state,
1583            crate::persist::BrainMutationEvent {
1584                profile_id: serving_profile_owned.clone(),
1585                event_kind: event.verb.clone(),
1586                payload: serde_json::to_value(&event)
1587                    .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?,
1588            },
1589            ENTITY_CACHE_CAPACITY,
1590            {
1591                let brain_signal = brain_signal.clone();
1592                let serving_profile_owned = serving_profile_owned.clone();
1593                move |state: &mut khive_brain_core::BrainState| -> Result<(), RuntimeError> {
1594                    let serving_profile = serving_profile_owned.as_str();
1595
1596                    if serving_profile == "balanced-recall-v1" {
1597                        state.balanced_recall.apply_signal(&brain_signal);
1598                        sync_balanced_recall_record(state);
1599                    } else if state.profile_states.contains_key(serving_profile) {
1600                        let ps = state
1601                            .profile_states
1602                            .get_mut(serving_profile)
1603                            .expect("key checked above");
1604                        ps.apply_signal(&brain_signal);
1605                        let snap = serde_json::to_value(ps.to_snapshot()).ok();
1606                        let total = ps.total_events;
1607                        if let Some(record) = state.profiles.get_mut(serving_profile) {
1608                            record.total_events = total;
1609                            record.state_snapshot = snap;
1610                        }
1611                    } else {
1612                        state.balanced_recall.apply_signal(&brain_signal);
1613                        sync_balanced_recall_record(state);
1614                    }
1615
1616                    // Seed and backfill section posteriors, then apply the signal.
1617                    // Shared contract with the replay path — see `ensure_section_state_seeded`.
1618                    {
1619                        let section_state = crate::ensure_section_state_seeded(
1620                            &mut state.section_states,
1621                            &serving_profile_owned,
1622                        );
1623                        section_state.apply_signal(&brain_signal);
1624                    }
1625
1626                    Ok(())
1627                }
1628            },
1629        )
1630        .await?;
1631
1632        // lattice-router: build the context vector from the now-published live
1633        // state and forward through the fann network. This is a best-effort
1634        // routing computation independent of durability, so it reads from
1635        // `self.state` only after the mutation above has durably committed.
1636        // Consumption of routed weights lands with the engine route() seam (#343) and the
1637        // compose value-gate (#346); no per-event stdout/stderr spam.
1638        #[cfg(feature = "lattice-router")]
1639        {
1640            let state = self.state.lock().unwrap();
1641            let serving_profile = serving_profile_owned.as_str();
1642            let (rel, sal, temp) = if serving_profile == "balanced-recall-v1" {
1643                (
1644                    state.balanced_recall.relevance.clone(),
1645                    state.balanced_recall.salience.clone(),
1646                    state.balanced_recall.temporal.clone(),
1647                )
1648            } else if let Some(ps) = state.profile_states.get(serving_profile) {
1649                (
1650                    ps.relevance.clone(),
1651                    ps.salience.clone(),
1652                    ps.temporal.clone(),
1653                )
1654            } else {
1655                (
1656                    state.balanced_recall.relevance.clone(),
1657                    state.balanced_recall.salience.clone(),
1658                    state.balanced_recall.temporal.clone(),
1659                )
1660            };
1661            let sec = state
1662                .section_states
1663                .get(serving_profile)
1664                .map(|ss| ss.posteriors.clone());
1665            drop(state);
1666            let ctx = build_context_vector(&rel, &sal, &temp, sec.as_ref());
1667            let _routed = route_via_fann(&ctx);
1668        }
1669
1670        // ADR-081 §6: "the fold... backfills the ledger row's grade." Runs after
1671        // the fold itself so a ledger-write failure never blocks the feedback
1672        // event from landing — intentionally non-fatal, unlike the fail-closed
1673        // brain-state persistence above (#458).
1674        if let (Some(scorer_run_id), Some(serve_ledger_id)) =
1675            (p.scorer_run_id.as_deref(), p.serve_ledger_id.as_deref())
1676        {
1677            if let Err(e) = crate::serve_ledger::backfill_grade(
1678                sql.as_ref(),
1679                serve_ledger_id,
1680                signal,
1681                now_us,
1682                scorer_run_id,
1683            )
1684            .await
1685            {
1686                eprintln!("[brain] serve ledger grade backfill failed (non-fatal): {e}");
1687            }
1688        }
1689
1690        Ok(json!({
1691            "emitted": true,
1692            "event_id": event.id.to_string(),
1693            "verb": "brain.feedback",
1694            "signal": signal,
1695            "target_id": target.to_string(),
1696        }))
1697    }
1698
1699    // ── brain.auto_feedback ───────────────────────────────────────────────
1700
1701    /// Emit implicit feedback for the first `memory.recall` result (accepts 8-char prefix or full UUID).
1702    pub(crate) async fn handle_auto_feedback(
1703        &self,
1704        token: &NamespaceToken,
1705        params: Value,
1706    ) -> Result<Value, RuntimeError> {
1707        #[derive(Deserialize)]
1708        #[serde(deny_unknown_fields)]
1709        struct AutoFeedbackParams {
1710            query: String,
1711            results: Vec<AutoFeedbackResult>,
1712            signal: Option<String>,
1713            served_by_profile_id: Option<String>,
1714            // ADR-081 §6: forwarded verbatim to brain.feedback, which owns the
1715            // together-or-rejected validation and the dedup/fold-gate logic.
1716            scorer_run_id: Option<String>,
1717            serve_ledger_id: Option<String>,
1718        }
1719
1720        #[derive(Deserialize)]
1721        struct AutoFeedbackResult {
1722            id: String,
1723        }
1724
1725        let p: AutoFeedbackParams = serde_json::from_value(params)
1726            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
1727        if p.query.trim().is_empty() {
1728            return Err(RuntimeError::InvalidInput(
1729                "auto_feedback: `query` must not be empty".into(),
1730            ));
1731        }
1732
1733        let Some(first) = p.results.first() else {
1734            return Ok(json!({
1735                "emitted": false,
1736                "verb": "brain.auto_feedback",
1737                "reason": "no_results",
1738            }));
1739        };
1740
1741        let target = resolve_auto_feedback_target(&self.runtime, token, &first.id).await?;
1742
1743        let mut feedback_params = json!({
1744            "target_id": target.to_string(),
1745            "signal": p.signal.as_deref().unwrap_or("implicit_positive"),
1746        });
1747        if let Some(ref profile_id) = p.served_by_profile_id {
1748            feedback_params["served_by_profile_id"] = json!(profile_id);
1749        }
1750        if let Some(ref scorer_run_id) = p.scorer_run_id {
1751            feedback_params["scorer_run_id"] = json!(scorer_run_id);
1752        }
1753        if let Some(ref serve_ledger_id) = p.serve_ledger_id {
1754            feedback_params["serve_ledger_id"] = json!(serve_ledger_id);
1755        }
1756
1757        let mut out = self.handle_feedback(token, feedback_params).await?;
1758        out["verb"] = json!("brain.auto_feedback");
1759        out["feedback_verb"] = json!("brain.feedback");
1760        out["result_count"] = json!(p.results.len());
1761        Ok(out)
1762    }
1763
1764    // ── brain.record_serve ────────────────────────────────────────────────
1765
1766    /// ADR-081 §4/§5 (#394): append one serve-ledger row per target id. Never
1767    /// propagates a per-target write failure as a batch error — a serve-ledger
1768    /// append is best-effort accounting, not the recall response itself, so a
1769    /// single row's failure (or an exact-key duplicate, tolerated as `skipped`)
1770    /// must not poison the rest of the batch.
1771    pub(crate) async fn handle_record_serve(
1772        &self,
1773        token: &NamespaceToken,
1774        params: Value,
1775    ) -> Result<Value, RuntimeError> {
1776        #[derive(Deserialize)]
1777        #[serde(deny_unknown_fields)]
1778        struct RecordServeParams {
1779            consumer_kind: String,
1780            served_by_profile_id: Option<String>,
1781            target_ids: Vec<String>,
1782            query_raw: String,
1783            served_at: Option<i64>,
1784        }
1785        let p: RecordServeParams = serde_json::from_value(params)
1786            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
1787
1788        if p.target_ids.is_empty() {
1789            return Ok(json!({
1790                "ok": true,
1791                "written": 0,
1792                "skipped": 0,
1793                "verb": "brain.record_serve",
1794            }));
1795        }
1796
1797        let namespace = token.namespace().as_str().to_string();
1798        let query_class = crate::serve_ledger::compute_query_class(&p.query_raw);
1799        let served_at = p.served_at.unwrap_or_else(|| Utc::now().timestamp_micros());
1800        let sql = self.runtime.sql();
1801
1802        let mut written = 0usize;
1803        let mut skipped = 0usize;
1804        for target_id in &p.target_ids {
1805            let row_id = uuid::Uuid::new_v4().to_string();
1806            match crate::serve_ledger::record_serve(
1807                sql.as_ref(),
1808                &row_id,
1809                &namespace,
1810                &p.consumer_kind,
1811                p.served_by_profile_id.as_deref(),
1812                None,
1813                None,
1814                target_id,
1815                &query_class,
1816                &p.query_raw,
1817                served_at,
1818            )
1819            .await
1820            {
1821                Ok(true) => written += 1,
1822                Ok(false) => skipped += 1,
1823                Err(e) => {
1824                    eprintln!(
1825                        "[brain] serve ledger write failed for target {target_id} (non-fatal): {e}"
1826                    );
1827                }
1828            }
1829        }
1830
1831        Ok(json!({
1832            "ok": true,
1833            "written": written,
1834            "skipped": skipped,
1835            "query_class": query_class,
1836            "verb": "brain.record_serve",
1837        }))
1838    }
1839
1840    // ── brain.emit (deprecated) ───────────────────────────────────────────
1841
1842    /// Deprecated alias for `brain.feedback`; routes to `handle_feedback`.
1843    pub(crate) async fn handle_emit(
1844        &self,
1845        token: &NamespaceToken,
1846        params: Value,
1847    ) -> Result<Value, RuntimeError> {
1848        self.handle_feedback(token, params).await
1849    }
1850
1851    // ── brain.bind ────────────────────────────────────────────────────────
1852
1853    pub(crate) async fn handle_bind(
1854        &self,
1855        token: &NamespaceToken,
1856        params: Value,
1857    ) -> Result<Value, RuntimeError> {
1858        #[derive(Deserialize)]
1859        #[serde(deny_unknown_fields)]
1860        struct BindParams {
1861            profile_id: String,
1862            actor: Option<String>,
1863            namespace: Option<String>,
1864            consumer_kind: Option<String>,
1865            priority: Option<i32>,
1866        }
1867        let p: BindParams = serde_json::from_value(params)
1868            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
1869
1870        let actor = p.actor.unwrap_or_else(|| "*".into());
1871        let namespace = p.namespace.unwrap_or_else(|| "*".into());
1872        let consumer_kind = p.consumer_kind.unwrap_or_else(|| "*".into());
1873
1874        // Validate that '*' is not used as a real value ('*' is reserved as the wildcard sentinel)
1875        for (field, val) in [
1876            ("actor", &actor),
1877            ("namespace", &namespace),
1878            ("consumer_kind", &consumer_kind),
1879        ] {
1880            if val.as_str() != "*" && val.contains('*') {
1881                return Err(RuntimeError::InvalidInput(format!(
1882                    "{field}: '*' is reserved as the wildcard sentinel and cannot appear inside a real value"
1883                )));
1884            }
1885        }
1886
1887        // Secret gate: scan arbitrary text fields before writing.
1888        // Wildcard sentinel `*` is safe; real values are scanned.
1889        if actor != "*" {
1890            khive_runtime::secret_gate::check(&actor)?;
1891        }
1892        if namespace != "*" {
1893            khive_runtime::secret_gate::check(&namespace)?;
1894        }
1895        if consumer_kind != "*" {
1896            khive_runtime::secret_gate::check(&consumer_kind)?;
1897        }
1898
1899        let profile_id = p.profile_id;
1900        let priority = p.priority.unwrap_or(0);
1901        let event_payload = json!({
1902            "profile_id": profile_id,
1903            "actor": actor,
1904            "namespace": namespace,
1905            "consumer_kind": consumer_kind,
1906            "priority": priority,
1907        });
1908
1909        // Route through the durable-write helper (issue #457): the binding
1910        // change is applied to a proposed state copy and only takes effect
1911        // after the brain event-log append + snapshot upsert commit.
1912        crate::persist::persist_brain_state_mutation(
1913            self.runtime.sql().as_ref(),
1914            token,
1915            &self.persistence,
1916            &self.state,
1917            crate::persist::BrainMutationEvent {
1918                profile_id: profile_id.clone(),
1919                event_kind: "brain.bind".to_string(),
1920                payload: event_payload,
1921            },
1922            ENTITY_CACHE_CAPACITY,
1923            move |state: &mut khive_brain_core::BrainState| -> Result<Value, RuntimeError> {
1924                // Verify the profile exists and is not archived (archived = terminal, no new bindings).
1925                match state.profiles.get(&profile_id) {
1926                    None => {
1927                        return Err(RuntimeError::NotFound(format!("profile {:?}", profile_id)));
1928                    }
1929                    Some(record) if record.lifecycle == ProfileLifecycle::Archived => {
1930                        return Err(RuntimeError::InvalidInput(format!(
1931                            "profile {:?} is archived; bindings to archived profiles are not permitted",
1932                            profile_id
1933                        )));
1934                    }
1935                    Some(_) => {}
1936                }
1937
1938                // Remove any existing binding for the same (actor, namespace, consumer_kind)
1939                state.bindings.retain(|b| {
1940                    !(b.actor == actor
1941                        && b.namespace == namespace
1942                        && b.consumer_kind == consumer_kind)
1943                });
1944
1945                state.bindings.push(ProfileBinding {
1946                    actor: actor.clone(),
1947                    namespace: namespace.clone(),
1948                    consumer_kind: consumer_kind.clone(),
1949                    profile_id: profile_id.clone(),
1950                    priority,
1951                    created_at: Utc::now(),
1952                });
1953
1954                Ok(json!({
1955                    "bound": true,
1956                    "profile_id": profile_id,
1957                    "actor": actor,
1958                    "namespace": namespace,
1959                    "consumer_kind": consumer_kind,
1960                }))
1961            },
1962        )
1963        .await
1964    }
1965
1966    // ── brain.unbind ──────────────────────────────────────────────────────
1967
1968    pub(crate) async fn handle_unbind(
1969        &self,
1970        token: &NamespaceToken,
1971        params: Value,
1972    ) -> Result<Value, RuntimeError> {
1973        #[derive(Deserialize)]
1974        #[serde(deny_unknown_fields)]
1975        struct UnbindParams {
1976            profile_id: Option<String>,
1977            actor: Option<String>,
1978            namespace: Option<String>,
1979            consumer_kind: Option<String>,
1980        }
1981        let p: UnbindParams = serde_json::from_value(params)
1982            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
1983
1984        // Require at least one non-null filter to prevent accidental bulk-delete.
1985        if p.profile_id.is_none()
1986            && p.actor.is_none()
1987            && p.namespace.is_none()
1988            && p.consumer_kind.is_none()
1989        {
1990            return Err(RuntimeError::InvalidInput(
1991                "unbind requires at least one filter; pass profile_id, actor, namespace, or consumer_kind".into(),
1992            ));
1993        }
1994
1995        let event_payload = json!({
1996            "profile_id": p.profile_id,
1997            "actor": p.actor,
1998            "namespace": p.namespace,
1999            "consumer_kind": p.consumer_kind,
2000        });
2001        let event_profile_id = p.profile_id.clone().unwrap_or_else(|| "*".to_string());
2002
2003        // Route through the durable-write helper (issue #457): the binding
2004        // removal is applied to a proposed state copy and only takes effect
2005        // after the brain event-log append + snapshot upsert commit.
2006        crate::persist::persist_brain_state_mutation(
2007            self.runtime.sql().as_ref(),
2008            token,
2009            &self.persistence,
2010            &self.state,
2011            crate::persist::BrainMutationEvent {
2012                profile_id: event_profile_id,
2013                event_kind: "brain.unbind".to_string(),
2014                payload: event_payload,
2015            },
2016            ENTITY_CACHE_CAPACITY,
2017            move |state: &mut khive_brain_core::BrainState| -> Result<Value, RuntimeError> {
2018                let before = state.bindings.len();
2019
2020                state.bindings.retain(|b| {
2021                    let pid_match = p.profile_id.as_ref().is_none_or(|id| &b.profile_id == id);
2022                    let actor_match = p.actor.as_ref().is_none_or(|a| &b.actor == a);
2023                    let ns_match = p.namespace.as_ref().is_none_or(|n| &b.namespace == n);
2024                    let kind_match = p
2025                        .consumer_kind
2026                        .as_ref()
2027                        .is_none_or(|k| &b.consumer_kind == k);
2028                    // Retain if this binding does NOT match ALL of the provided filters.
2029                    // A filter that is absent (None) matches everything — only bindings
2030                    // satisfying every supplied criterion are removed.
2031                    !(pid_match && actor_match && ns_match && kind_match)
2032                });
2033
2034                let removed = before - state.bindings.len();
2035                Ok(json!({ "unbound": removed }))
2036            },
2037        )
2038        .await
2039    }
2040
2041    // ── brain.bindings ────────────────────────────────────────────────────
2042
2043    pub(crate) async fn handle_bindings(&self, params: Value) -> Result<Value, RuntimeError> {
2044        // Inspection verb — list binding rows, optionally filtered.
2045        #[derive(Deserialize)]
2046        struct BindingsParams {
2047            profile_id: Option<String>,
2048            actor: Option<String>,
2049            namespace: Option<String>,
2050            consumer_kind: Option<String>,
2051        }
2052        let p: BindingsParams = serde_json::from_value(params)
2053            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
2054
2055        let state = self.state.lock().unwrap();
2056        let rows: Vec<Value> = state
2057            .bindings
2058            .iter()
2059            .filter(|b| {
2060                p.profile_id.as_ref().is_none_or(|id| &b.profile_id == id)
2061                    && p.actor.as_ref().is_none_or(|a| &b.actor == a)
2062                    && p.namespace.as_ref().is_none_or(|n| &b.namespace == n)
2063                    && p.consumer_kind
2064                        .as_ref()
2065                        .is_none_or(|k| &b.consumer_kind == k)
2066            })
2067            .map(|b| {
2068                json!({
2069                    "profile_id": b.profile_id,
2070                    "actor": b.actor,
2071                    "namespace": b.namespace,
2072                    "consumer_kind": b.consumer_kind,
2073                    "priority": b.priority,
2074                    "created_at": b.created_at,
2075                })
2076            })
2077            .collect();
2078
2079        Ok(json!({ "count": rows.len(), "bindings": rows }))
2080    }
2081
2082    // ── brain.create_profile ──────────────────────────────────────────────
2083
2084    pub(crate) async fn handle_create_profile(
2085        &self,
2086        token: &NamespaceToken,
2087        params: Value,
2088    ) -> Result<Value, RuntimeError> {
2089        // Allow external agents to create new profiles via MCP.
2090        // seed_priors: optional object for seeding section priors, or null for defaults.
2091        #[derive(Deserialize)]
2092        struct CreateProfileParams {
2093            name: String,
2094            description: Option<String>,
2095            consumer_kind: Option<String>,
2096            seed_priors: Option<serde_json::Value>,
2097        }
2098        let p: CreateProfileParams = serde_json::from_value(params)
2099            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
2100
2101        // Validate profile-id grammar: trim, reject empty, enforce ASCII alphanumeric + hyphen.
2102        let name = p.name.trim().to_string();
2103        if name.is_empty() {
2104            return Err(RuntimeError::InvalidInput(
2105                "name must not be empty or whitespace-only".into(),
2106            ));
2107        }
2108        if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-') {
2109            return Err(RuntimeError::InvalidInput(format!(
2110                "name {:?} is invalid; profile IDs must match [a-zA-Z0-9-]+ (alphanumeric and hyphens only)",
2111                name
2112            )));
2113        }
2114        let p_name = name;
2115
2116        // Validate consumer_kind — reject empty/whitespace and wildcard sentinel.
2117        let consumer_kind = p.consumer_kind.unwrap_or_else(|| "recall".into());
2118        let ck_trimmed = consumer_kind.trim();
2119        if ck_trimmed.is_empty() {
2120            return Err(RuntimeError::InvalidInput(
2121                "consumer_kind must not be empty or whitespace".into(),
2122            ));
2123        }
2124        if ck_trimmed == "*" {
2125            return Err(RuntimeError::InvalidInput(
2126                "consumer_kind '*' is the wildcard sentinel and is not permitted for profile creation; provide a specific operation kind (e.g. \"recall\", \"search\")"
2127                    .into(),
2128            ));
2129        }
2130
2131        let description = p
2132            .description
2133            .unwrap_or_else(|| format!("User-created profile: {}", p_name));
2134
2135        // Secret gate: scan caller-supplied text before any write.
2136        // `p_name` is already constrained to [a-zA-Z0-9-]+ and cannot carry a secret.
2137        khive_runtime::secret_gate::check(&description)?;
2138        khive_runtime::secret_gate::check(&consumer_kind)?;
2139        if let Some(ref seed) = p.seed_priors {
2140            khive_runtime::secret_gate::check_json(seed)?;
2141        }
2142        let seed_priors = p.seed_priors;
2143
2144        let event_payload = json!({
2145            "profile_id": p_name,
2146            "consumer_kind": consumer_kind,
2147            "description": description,
2148        });
2149
2150        // Route through the durable-write helper (issue #457): the new
2151        // profile/state/section rows are applied to a proposed state copy
2152        // and only take effect after the brain event-log append + snapshot
2153        // upsert commit — so a persistence failure never leaves a phantom
2154        // profile that vanishes on restart.
2155        crate::persist::persist_brain_state_mutation(
2156            self.runtime.sql().as_ref(),
2157            token,
2158            &self.persistence,
2159            &self.state,
2160            crate::persist::BrainMutationEvent {
2161                profile_id: p_name.clone(),
2162                event_kind: "brain.create_profile".to_string(),
2163                payload: event_payload,
2164            },
2165            ENTITY_CACHE_CAPACITY,
2166            move |state: &mut khive_brain_core::BrainState| -> Result<Value, RuntimeError> {
2167                if state.profiles.contains_key(&p_name) {
2168                    return Err(RuntimeError::InvalidInput(format!(
2169                        "profile {:?} already exists",
2170                        p_name
2171                    )));
2172                }
2173
2174                // Initialize live BalancedRecallState for this profile so that reset and
2175                // feedback can route to its actual posteriors rather than a metadata-only record.
2176                let ps = khive_brain_core::BalancedRecallState::new(ENTITY_CACHE_CAPACITY);
2177                let snap = serde_json::to_value(ps.to_snapshot()).ok();
2178
2179                let record = ProfileRecord {
2180                    id: p_name.clone(),
2181                    description: description.clone(),
2182                    consumer_kind: consumer_kind.clone(),
2183                    state_class: "Bayesian".into(),
2184                    lifecycle: ProfileLifecycle::Inactive,
2185                    created_at: Utc::now(),
2186                    state_snapshot: snap,
2187                    total_events: 0,
2188                    exploration_epoch: 0,
2189                };
2190
2191                // Seed section posteriors: parse explicit section_posteriors object if provided,
2192                // else use default informative priors.
2193                let section_state = if let Some(ref seed) = seed_priors {
2194                    if let Some(sp_obj) = seed.get("section_posteriors").and_then(|v| v.as_object())
2195                    {
2196                        let mut priors = std::collections::HashMap::new();
2197                        for (key, val) in sp_obj {
2198                            let st: SectionType = key.parse().map_err(|_| {
2199                                RuntimeError::InvalidInput(format!("unknown section type: {key:?}"))
2200                            })?;
2201                            let alpha =
2202                                val.get("alpha").and_then(|v| v.as_f64()).ok_or_else(|| {
2203                                    RuntimeError::InvalidInput(format!(
2204                                        "missing or invalid alpha for section {key:?}"
2205                                    ))
2206                                })?;
2207                            let beta =
2208                                val.get("beta").and_then(|v| v.as_f64()).ok_or_else(|| {
2209                                    RuntimeError::InvalidInput(format!(
2210                                        "missing or invalid beta for section {key:?}"
2211                                    ))
2212                                })?;
2213                            let posterior = khive_brain_core::BetaPosterior::try_new(alpha, beta)
2214                                .map_err(|e| {
2215                                RuntimeError::InvalidInput(format!(
2216                                    "invalid seed_priors for section {key:?}: {e}"
2217                                ))
2218                            })?;
2219                            let ess = posterior.effective_sample_size();
2220                            if ess > DEFAULT_ESS_CAP {
2221                                return Err(RuntimeError::InvalidInput(format!(
2222                                    "seed_priors for section {key:?}: alpha+beta ({ess}) exceeds \
2223                                     maximum allowed ESS ({DEFAULT_ESS_CAP}); use values where \
2224                                     alpha + beta <= {DEFAULT_ESS_CAP}"
2225                                )));
2226                            }
2227                            priors.insert(st, posterior);
2228                        }
2229                        SectionPosteriorState::from_priors(priors)
2230                    } else {
2231                        return Err(RuntimeError::InvalidInput(
2232                            "seed_priors must contain a 'section_posteriors' object".into(),
2233                        ));
2234                    }
2235                } else {
2236                    SectionPosteriorState::new()
2237                };
2238
2239                state.profiles.insert(p_name.clone(), record);
2240                state.profile_states.insert(p_name.clone(), ps);
2241                state.section_states.insert(p_name.clone(), section_state);
2242
2243                Ok(json!({
2244                    "created": true,
2245                    "profile_id": p_name,
2246                    "consumer_kind": consumer_kind,
2247                    "lifecycle": "inactive",
2248                    "description": description,
2249                }))
2250            },
2251        )
2252        .await
2253    }
2254
2255    // ── brain.register_adapter ────────────────────────────────────────────
2256
2257    pub(crate) async fn handle_register_adapter(
2258        &self,
2259        token: &NamespaceToken,
2260        params: Value,
2261    ) -> Result<Value, RuntimeError> {
2262        #[derive(Deserialize)]
2263        #[serde(deny_unknown_fields)]
2264        struct RegisterAdapterParams {
2265            adapter_id: String,
2266            content_hash: String,
2267            base_model_revision: String,
2268            metadata: Option<serde_json::Value>,
2269        }
2270        let p: RegisterAdapterParams = serde_json::from_value(params)
2271            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?;
2272
2273        let active_revision = std::env::var("KHIVE_BRAIN_BASE_MODEL_REVISION")
2274            .unwrap_or_else(|_| DEFAULT_BASE_MODEL_REVISION.to_string());
2275
2276        if p.base_model_revision != active_revision {
2277            return Err(RuntimeError::InvalidInput(format!(
2278                "base_model_revision mismatch: expected {:?}, got {:?}",
2279                active_revision, p.base_model_revision
2280            )));
2281        }
2282
2283        let mut props = serde_json::json!({
2284            "content_hash": p.content_hash,
2285            "base_model_revision": p.base_model_revision,
2286        });
2287        if let Some(serde_json::Value::Object(meta)) = p.metadata {
2288            let props_obj = props.as_object_mut().unwrap();
2289            for (k, v) in meta {
2290                if k != "content_hash" && k != "base_model_revision" {
2291                    props_obj.insert(k, v);
2292                }
2293            }
2294        }
2295
2296        self.runtime
2297            .create_entity(
2298                token,
2299                "artifact",
2300                Some("adapter"),
2301                &p.adapter_id,
2302                None,
2303                Some(props),
2304                vec![],
2305            )
2306            .await?;
2307
2308        Ok(json!({
2309            "registered": true,
2310            "adapter_id": p.adapter_id,
2311            "content_hash": p.content_hash,
2312            "base_model_revision": p.base_model_revision,
2313        }))
2314    }
2315}
2316
2317/// Parse an ISO-8601/RFC-3339 datetime string into a microsecond epoch,
2318/// naming the offending field/value in the error rather than a bare parse
2319/// failure (`brain.event_counts`, ADR-103 Stage 1).
2320fn parse_rfc3339_micros(field: &'static str, value: &str) -> Result<i64, RuntimeError> {
2321    chrono::DateTime::parse_from_rfc3339(value.trim())
2322        .map(|dt| dt.with_timezone(&Utc).timestamp_micros())
2323        .map_err(|e| {
2324            RuntimeError::InvalidInput(format!(
2325                "invalid `{field}`: {value:?} is not a valid ISO-8601/RFC-3339 datetime ({e}); \
2326                 expected e.g. \"2026-07-01T00:00:00Z\""
2327            ))
2328        })
2329}
2330
2331/// Extract `work_class` from an event payload for `brain.event_counts`'
2332/// `counts_by_work_class` split (ADR-103 Stage 1, ADR-103-resource-attribution-model.md:303).
2333///
2334/// Checks two paths, phase-payload first:
2335/// 1. `payload.work_class` — the shape `PhaseStarted`/`PhaseCompleted`/`PhaseCancelled`
2336///    events use today (`khive_storage::telemetry::{PhaseStartedPayload, ...}`).
2337/// 2. `payload.resource.work_class` — the shape successful-dispatch audit rows write:
2338///    `khive_runtime::cost_unit` constructs the resource object with `work_class` and the
2339///    runtime persists it on successful audit rows. Phase payloads keep precedence when
2340///    both paths are present.
2341///
2342/// Returns `None` (silently excluded from the split, not an error) when neither path is
2343/// present — the overwhelming majority of event kinds today.
2344fn event_work_class(payload: &Value) -> Option<&str> {
2345    payload
2346        .get("work_class")
2347        .and_then(Value::as_str)
2348        .or_else(|| {
2349            payload
2350                .get("resource")
2351                .and_then(|resource| resource.get("work_class"))
2352                .and_then(Value::as_str)
2353        })
2354}
2355
2356/// Extract `cost_unit` from an event payload for `brain.event_counts`'
2357/// `total_cost_unit` / `cost_unit_by_verb` aggregation (ADR-103 Amendment 1).
2358///
2359/// Reads `payload.resource.cost_unit` only — unlike `work_class`, `cost_unit`
2360/// has no top-level payload shape: it is emitted exclusively by
2361/// `khive_runtime::cost_unit::resource_payload` on the per-dispatch audit
2362/// row's `resource` sub-object, never by the Phase* background-work events.
2363///
2364/// Returns `None` (silently excluded from the sum, not an error or a zero)
2365/// when the field is absent, which per Amendment 1's "absence has exactly
2366/// two meanings" rule is either a pre-Amendment-1 event or an
2367/// errored/denied dispatch.
2368fn event_cost_unit(payload: &Value) -> Option<i64> {
2369    payload
2370        .get("resource")
2371        .and_then(|resource| resource.get("cost_unit"))
2372        .and_then(Value::as_i64)
2373}
2374
2375// ── lattice-router seam (#345 M1 / #352 M2) ──────────────────────────────────
2376
2377/// Context-vector dimension for the lattice-fann router seam.
2378/// Layout:
2379///   [0..2]  = relevance  {mean, ess}
2380///   [2..4]  = salience   {mean, ess}
2381///   [4..6]  = temporal   {mean, ess}
2382///   [6..16] = 10 section posterior means in `SectionType::all()` order
2383#[cfg(feature = "lattice-router")]
2384const ROUTER_CONTEXT_DIM: usize = 16;
2385
2386/// Build a 16-element context vector from live brain posteriors.
2387///
2388/// ESS values are stored as raw f32 casts.  Normalization is deferred to the
2389/// engine `route()` seam (#343) once the network is trained; documenting the
2390/// omission here so it is not forgotten.
2391///
2392/// When `sections` is `None` the 10 section slots are filled from
2393/// `SectionPosteriorState::default_priors()` means — deterministically neutral,
2394/// never arbitrary zeros.
2395#[cfg(feature = "lattice-router")]
2396fn build_context_vector(
2397    relevance: &BetaPosterior,
2398    salience: &BetaPosterior,
2399    temporal: &BetaPosterior,
2400    sections: Option<&std::collections::HashMap<SectionType, BetaPosterior>>,
2401) -> [f32; ROUTER_CONTEXT_DIM] {
2402    let mut v = [0.0f32; ROUTER_CONTEXT_DIM];
2403    v[0] = relevance.mean() as f32;
2404    v[1] = relevance.effective_sample_size() as f32;
2405    v[2] = salience.mean() as f32;
2406    v[3] = salience.effective_sample_size() as f32;
2407    v[4] = temporal.mean() as f32;
2408    v[5] = temporal.effective_sample_size() as f32;
2409    // Section slots: deterministic ordering via SectionType::all().
2410    // Pre-compute default priors unconditionally so that slots absent from a
2411    // caller-supplied partial map fall back to the configured prior mean, not
2412    // an arbitrary 0.5 that would disagree with the seeded posteriors.
2413    let default_priors = SectionPosteriorState::default_priors();
2414    let posteriors: &std::collections::HashMap<SectionType, BetaPosterior> =
2415        sections.unwrap_or(&default_priors);
2416    for (i, st) in SectionType::all().iter().enumerate() {
2417        let mean = posteriors
2418            .get(st)
2419            .or_else(|| default_priors.get(st))
2420            .map_or(0.5, |p| p.mean());
2421        v[6 + i] = mean as f32;
2422    }
2423    v
2424}
2425
2426/// M1 seam: links lattice-fann and produces routed mixture weights.
2427/// The real engine `route(context_vector, available_adapters)` (lattice mixture-runtime,
2428/// #343) swaps in here when shipped; M1 proves the dependency links and the seam runs.
2429#[cfg(feature = "lattice-router")]
2430fn route_via_fann(context: &[f32]) -> Vec<f32> {
2431    use lattice_fann::{Activation, NetworkBuilder};
2432    const ADAPTER_SLOTS: usize = 4;
2433    let mut network = match NetworkBuilder::new()
2434        .input(ROUTER_CONTEXT_DIM)
2435        .hidden(8, Activation::ReLU)
2436        .output(ADAPTER_SLOTS, Activation::Softmax)
2437        .build()
2438    {
2439        Ok(n) => n,
2440        Err(e) => {
2441            eprintln!("[brain] lattice-router: network build failed (non-fatal): {e}");
2442            return Vec::new();
2443        }
2444    };
2445    match network.forward(context) {
2446        Ok(weights) => weights.to_vec(),
2447        Err(e) => {
2448            eprintln!("[brain] lattice-router: forward failed (non-fatal): {e}");
2449            Vec::new()
2450        }
2451    }
2452}
2453
2454// ── brain.auto_feedback helpers ───────────────────────────────────────────────
2455
2456/// Resolve an `id` from `memory.recall` output to a full UUID.
2457///
2458/// Accepts a 36-char UUID directly, or an 8-char hex prefix (Agent-mode short
2459/// form). Returns `InvalidInput` if neither form matches or the prefix is
2460/// ambiguous.
2461pub(crate) async fn resolve_auto_feedback_target(
2462    runtime: &KhiveRuntime,
2463    token: &NamespaceToken,
2464    raw: &str,
2465) -> Result<uuid::Uuid, RuntimeError> {
2466    if let Ok(uuid) = raw.parse::<uuid::Uuid>() {
2467        return Ok(uuid);
2468    }
2469    if raw.len() >= 8 && raw.chars().all(|c| c.is_ascii_hexdigit()) {
2470        return runtime
2471            .resolve_prefix(token, raw)
2472            .await
2473            .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?
2474            .ok_or_else(|| {
2475                RuntimeError::InvalidInput(format!(
2476                    "auto_feedback: no record matches id prefix: {raw:?}"
2477                ))
2478            });
2479    }
2480    Err(RuntimeError::InvalidInput(format!(
2481        "auto_feedback: invalid id {raw:?}; expected full UUID or 8-char hex prefix"
2482    )))
2483}
2484
2485// ── DispatchHook impl ─────────────────────────────────────────────────────────
2486
2487/// `BrainPack` as a post-dispatch hook.
2488///
2489/// When registered via `VerbRegistryBuilder::with_dispatch_hook`, every
2490/// successful verb dispatch calls `on_dispatch` with a synthesized `Event`.
2491/// The event is fed into `BalancedRecallFold::reduce`, updating the brain's
2492/// posteriors in real time — no polling required.
2493#[async_trait]
2494impl DispatchHook for BrainPack {
2495    async fn on_dispatch(&self, view: &EventView) {
2496        if view.event.verb.starts_with("brain.") {
2497            return;
2498        }
2499
2500        let _gate = self.dispatch_gate.lock().await;
2501
2502        let signal = interpret(&view.event);
2503
2504        // Route the signal to the state bucket that owns view.event.namespace.
2505        // No event is silently dropped: cold and saved namespaces are updated
2506        // inside PersistenceTracker; only when the namespace is the active slot
2507        // do we need to apply to the shared BrainState.
2508        let target = {
2509            let mut tracker = self.persistence.lock().unwrap();
2510            tracker.route_signal(&view.event.namespace, &signal, ENTITY_CACHE_CAPACITY)
2511        };
2512
2513        if matches!(target, crate::persist::ApplyTarget::ActiveSlot) {
2514            let mut state = self.state.lock().unwrap();
2515            state.balanced_recall.apply_signal(&signal);
2516            sync_balanced_recall_record(&mut state);
2517        }
2518    }
2519}
2520
2521// ── PackRuntime dispatch ──────────────────────────────────────────────────────
2522
2523#[async_trait]
2524impl khive_runtime::pack::PackRuntime for BrainPack {
2525    fn name(&self) -> &str {
2526        <BrainPack as khive_types::Pack>::NAME
2527    }
2528
2529    fn note_kinds(&self) -> &'static [&'static str] {
2530        <BrainPack as khive_types::Pack>::NOTE_KINDS
2531    }
2532
2533    fn entity_kinds(&self) -> &'static [&'static str] {
2534        <BrainPack as khive_types::Pack>::ENTITY_KINDS
2535    }
2536
2537    fn handlers(&self) -> &'static [HandlerDef] {
2538        BRAIN_HANDLERS
2539    }
2540
2541    fn requires(&self) -> &'static [&'static str] {
2542        <BrainPack as khive_types::Pack>::REQUIRES
2543    }
2544
2545    async fn dispatch(
2546        &self,
2547        verb: &str,
2548        params: Value,
2549        _registry: &VerbRegistry,
2550        token: &NamespaceToken,
2551    ) -> Result<Value, RuntimeError> {
2552        // Serialise the (ensure_loaded → handler) pair under the dispatch gate
2553        // so no concurrent dispatch for a different namespace can swap the
2554        // single shared BrainState slot between the two steps.
2555        //
2556        // Lock order: dispatch_gate (outermost) → persistence → state.
2557        // Nothing inside ensure_loaded or any handler acquires dispatch_gate,
2558        // so there is no lock-order cycle.
2559        let _gate = self.dispatch_gate.lock().await;
2560
2561        self.ensure_loaded(token).await?;
2562
2563        // In test builds, fire the interleaving hook (if any) between
2564        // ensure_loaded returning and the handler acquiring self.state.
2565        // This lets tests prove that without the gate a concurrent namespace
2566        // swap would corrupt the handler's view.
2567        #[cfg(test)]
2568        {
2569            let hook = crate::pack::DISPATCH_INTERLEAVE_HOOK.lock().unwrap().take();
2570            if let Some(h) = hook {
2571                let _ = h.reached_tx.send(());
2572                let _ = h.proceed_rx.await;
2573            }
2574        }
2575
2576        match verb {
2577            // Assertive
2578            "brain.state" => self.handle_state(params).await,
2579            "brain.config" => self.handle_config(params).await,
2580            "brain.events" => self.handle_events(token, params).await,
2581            "brain.event_counts" => self.handle_event_counts(token, params).await,
2582            "brain.profiles" => self.handle_profiles(params).await,
2583            "brain.profile" => self.handle_profile(params).await,
2584            "brain.resolve" => self.handle_resolve(token, params).await,
2585            "brain.bindings" => self.handle_bindings(params).await,
2586            // Commissive
2587            "brain.activate" => self.handle_activate(token, params).await,
2588            "brain.deactivate" => self.handle_deactivate(token, params).await,
2589            "brain.archive" => self.handle_archive(token, params).await,
2590            "brain.reset" => self.handle_reset(token, params).await,
2591            "brain.feedback" => self.handle_feedback(token, params).await,
2592            "brain.auto_feedback" => self.handle_auto_feedback(token, params).await,
2593            "brain.record_serve" => self.handle_record_serve(token, params).await,
2594            // Declaration
2595            "brain.bind" => self.handle_bind(token, params).await,
2596            "brain.unbind" => self.handle_unbind(token, params).await,
2597            "brain.create_profile" => self.handle_create_profile(token, params).await,
2598            "brain.register_adapter" => self.handle_register_adapter(token, params).await,
2599            // Legacy
2600            "brain.emit" => self.handle_emit(token, params).await,
2601            _ => Err(RuntimeError::InvalidInput(format!(
2602                "brain pack does not handle verb {verb:?}"
2603            ))),
2604        }
2605    }
2606}
2607
2608#[cfg(all(test, feature = "lattice-router"))]
2609mod router_tests {
2610    use super::*;
2611
2612    #[test]
2613    fn route_via_fann_emits_normalized_mixture() {
2614        let rel = BetaPosterior::new(7.0, 3.0);
2615        let sal = BetaPosterior::new(2.0, 8.0);
2616        let temp = BetaPosterior::new(1.0, 9.0);
2617        let ctx = build_context_vector(&rel, &sal, &temp, None);
2618        let weights = route_via_fann(&ctx);
2619        assert_eq!(weights.len(), 4);
2620        let sum: f32 = weights.iter().sum();
2621        assert!(
2622            (sum - 1.0).abs() < 1e-3,
2623            "softmax weights must sum to ~1, got {sum}"
2624        );
2625    }
2626
2627    #[test]
2628    fn context_vector_reflects_posterior_state() {
2629        // High-relevance state.
2630        let rel_high = BetaPosterior::new(90.0, 10.0);
2631        let sal = BetaPosterior::new(2.0, 8.0);
2632        let temp = BetaPosterior::new(1.0, 9.0);
2633        let v_high = build_context_vector(&rel_high, &sal, &temp, None);
2634
2635        // Low-relevance state.
2636        let rel_low = BetaPosterior::new(1.0, 9.0);
2637        let v_low = build_context_vector(&rel_low, &sal, &temp, None);
2638
2639        // Vectors must differ.
2640        assert_ne!(
2641            v_high, v_low,
2642            "context vectors must differ for different posteriors"
2643        );
2644
2645        // Dimension must be 16.
2646        assert_eq!(v_high.len(), ROUTER_CONTEXT_DIM);
2647        assert_eq!(v_low.len(), ROUTER_CONTEXT_DIM);
2648
2649        // Slot 0 = relevance mean; slot 1 = relevance ESS — verify they mirror the input.
2650        assert!(
2651            (v_high[0] - rel_high.mean() as f32).abs() < 1e-5,
2652            "slot 0 must equal relevance mean (high); got {}",
2653            v_high[0]
2654        );
2655        assert!(
2656            (v_high[1] - rel_high.effective_sample_size() as f32).abs() < 1e-5,
2657            "slot 1 must equal relevance ESS; got {}",
2658            v_high[1]
2659        );
2660        assert!(
2661            (v_low[0] - rel_low.mean() as f32).abs() < 1e-5,
2662            "slot 0 must equal relevance mean (low); got {}",
2663            v_low[0]
2664        );
2665
2666        // Slot 0 must differ substantially between high and low states.
2667        assert!(
2668            (v_high[0] - v_low[0]).abs() > 0.5,
2669            "relevance mean slot must differ by >0.5; high={} low={}",
2670            v_high[0],
2671            v_low[0]
2672        );
2673
2674        // Section slots [6..16] filled from default priors when sections=None — never zero.
2675        for (i, &val) in v_high.iter().enumerate().skip(6) {
2676            assert!(
2677                val > 0.0,
2678                "section slot {i} must be non-zero (default priors); got {val}",
2679            );
2680        }
2681    }
2682
2683    /// Regression gate: a `Some(map)` that omits one SectionType must fill the
2684    /// missing slot from the default prior mean, never from the bare value 0.5.
2685    ///
2686    /// Formalism has `BetaPosterior(1.5, 4.0)` → mean ≈ 0.273, which differs
2687    /// from 0.5 by more than 0.1 and is easy to distinguish.
2688    #[test]
2689    fn build_context_vector_uses_prior_mean_for_missing_section_slot() {
2690        // Build a partial posteriors map that intentionally omits Formalism.
2691        let mut partial = SectionPosteriorState::default_priors();
2692        partial.remove(&SectionType::Formalism);
2693        assert!(
2694            !partial.contains_key(&SectionType::Formalism),
2695            "Formalism must be absent from the test map before calling the helper"
2696        );
2697
2698        let neutral = BetaPosterior::new(2.0, 2.0);
2699        let v = build_context_vector(&neutral, &neutral, &neutral, Some(&partial));
2700
2701        let formalism_idx = SectionType::all()
2702            .iter()
2703            .position(|st| *st == SectionType::Formalism)
2704            .expect("Formalism must be in SectionType::all()");
2705        let slot_val = v[6 + formalism_idx];
2706
2707        let prior_mean = SectionPosteriorState::default_priors()
2708            .get(&SectionType::Formalism)
2709            .expect("default_priors must include Formalism")
2710            .mean() as f32;
2711
2712        assert!(
2713            (slot_val - prior_mean).abs() < 1e-5,
2714            "missing Formalism slot must use prior mean ({prior_mean:.4}), not bare 0.5; \
2715             got {slot_val:.4}"
2716        );
2717        assert!(
2718            (slot_val - 0.5_f32).abs() > 0.1,
2719            "slot must NOT fall back to the bare 0.5 value; got {slot_val:.4}"
2720        );
2721    }
2722}