diff --git a/AGENTS.md b/AGENTS.md index f1df391ef..8869b2e7f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -108,23 +108,23 @@ Composite scores are always in [0,1]. Typical production floor: 0.3-0.7. ### Brain pack — 15 verbs (`brain.` prefix) -| Verb | What it does | When to use | -| ------------------------ | ---------------------------------------------------------------------- | ---------------------------------------------------------- | -| `brain.event_counts` | Windowed event counts by kind/actor/verb (+ feedback by_profile split) | Flywheel metrics, feedback-coverage reporting | -| `brain.profiles` | List profiles (optionally filtered by lifecycle) | See what profiles exist | -| `brain.profile` | Full detail: metadata, snapshot, state summary | Inspect a specific profile | -| `brain.create_profile` | Create a new profile with optional seed priors | Custom tuning for a new consumer | -| `brain.resolve` | Which profile serves a given consumer context? | Before recall — check active tuning | -| `brain.activate` | Start live update loop for a profile | Enable feedback-driven tuning | -| `brain.deactivate` | Stop live updates, retain state | Pause tuning without losing progress | -| `brain.archive` | Read-only, audit-retained | Retire a profile permanently | -| `brain.reset` | Reset posteriors to priors (preserves event history) | Start tuning fresh | -| `brain.feedback` | Emit explicit feedback event | Rate a recall result as useful/not_useful/wrong | -| `brain.auto_feedback` | Emit implicit feedback for recall results | Convenience: agents call after memory.recall | -| `brain.bind` | Bind a profile to an actor + consumer | Route a specific caller to a specific profile | -| `brain.unbind` | Remove a binding | Stop routing | -| `brain.bindings` | List binding rows | Audit profile routing | -| `brain.register_adapter` | Register an adapter integrity record | Gate adapter composition to the active base-model revision | +| Verb | What it does | When to use | +| ------------------------ | ---------------------------------------------------------------------------------------- | --------------------------------------------------------------- | +| `brain.event_counts` | Windowed event counts by kind/actor/verb (+ feedback by_profile split, cost_unit totals) | Flywheel metrics, feedback-coverage reporting, cost attribution | +| `brain.profiles` | List profiles (optionally filtered by lifecycle) | See what profiles exist | +| `brain.profile` | Full detail: metadata, snapshot, state summary | Inspect a specific profile | +| `brain.create_profile` | Create a new profile with optional seed priors | Custom tuning for a new consumer | +| `brain.resolve` | Which profile serves a given consumer context? | Before recall — check active tuning | +| `brain.activate` | Start live update loop for a profile | Enable feedback-driven tuning | +| `brain.deactivate` | Stop live updates, retain state | Pause tuning without losing progress | +| `brain.archive` | Read-only, audit-retained | Retire a profile permanently | +| `brain.reset` | Reset posteriors to priors (preserves event history) | Start tuning fresh | +| `brain.feedback` | Emit explicit feedback event | Rate a recall result as useful/not_useful/wrong | +| `brain.auto_feedback` | Emit implicit feedback for recall results | Convenience: agents call after memory.recall | +| `brain.bind` | Bind a profile to an actor + consumer | Route a specific caller to a specific profile | +| `brain.unbind` | Remove a binding | Stop routing | +| `brain.bindings` | List binding rows | Audit profile routing | +| `brain.register_adapter` | Register an adapter integrity record | Gate adapter composition to the active base-model revision | ### Comm pack — 7 verbs (`comm.` prefix) diff --git a/crates/khive-db/src/writer_task.rs b/crates/khive-db/src/writer_task.rs index c13590065..17738a9a5 100644 --- a/crates/khive-db/src/writer_task.rs +++ b/crates/khive-db/src/writer_task.rs @@ -420,6 +420,7 @@ async fn run_writer_task( mod tests { use super::*; use crate::pool::PoolConfig; + use serial_test::serial; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -432,7 +433,14 @@ mod tests { ConnectionPool::new(cfg).expect("pool open") } + // `#[serial(tx_registry)]`: `run_writer_task` registers a `writer_task_tx` + // handle in the process-wide `tx_registry` singleton for the life of each + // `BEGIN IMMEDIATE`. Tests that observe the registry (the checkpoint + // `tx_age_sweep_*` group) read `tx_registry::oldest()`; an un-serialized + // spawning test here would leak a longer-lived `writer_task_tx` into that + // read and make the sweep name the wrong transaction. Share the key. #[tokio::test] + #[serial(tx_registry)] async fn begin_immediate_failure_replies_error_without_running_op() { // Real lock contention, not a simulation: hold the database-level // write lock from the pool's own writer connection (the unmigrated @@ -504,7 +512,11 @@ mod tests { ); } + // `#[serial(tx_registry)]`: shares the key with the checkpoint + // `tx_age_sweep_*` tests — see the note on + // `begin_immediate_failure_replies_error_without_running_op`. #[tokio::test] + #[serial(tx_registry)] async fn writer_task_executes_op_and_commits() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("writer_task_commit.db"); @@ -627,7 +639,13 @@ mod tests { first.abort(); } + // `#[serial(tx_registry)]`: this test deliberately keeps a request (and + // thus its `writer_task_tx` registry handle) alive past a timeout, so it is + // the worst polluter of the checkpoint `tx_age_sweep_*` reads if left + // un-serialized. Shares the key — see the note on + // `begin_immediate_failure_replies_error_without_running_op`. #[tokio::test] + #[serial(tx_registry)] async fn send_with_timeout_returns_op_result_when_op_outlives_the_timeout() { // `send_with_timeout`'s timeout must bound ONLY the enqueue step — // never the reply-wait. An accepted request (channel not full) must diff --git a/crates/khive-mcp/src/serve.rs b/crates/khive-mcp/src/serve.rs index 6116fac2f..a9fd7a163 100644 --- a/crates/khive-mcp/src/serve.rs +++ b/crates/khive-mcp/src/serve.rs @@ -5048,11 +5048,21 @@ backend = "kg-backend" ); use clap::Parser; - let args = Args::parse_from(["mcp", "--config", config_path.to_str().expect("utf8 path")]); + let args = Args::parse_from([ + "mcp", + "--config", + config_path.to_str().expect("utf8 path"), + "--no-embed", + ]); let (server, schedule_rt) = build_server(&args).expect("build_server must succeed"); // No `[packs.schedule]` entry above, so it defaults to "main". let rt = schedule_rt.expect("schedule pack is loaded by default"); + assert!( + rt.config().embedding_model.is_none() + && rt.config().additional_embedding_models.is_empty(), + "the backend-routing fixture must remain independent of external embedding models" + ); let marker = "adr106-multi-backend-dispatch-marker"; let action_dsl = format!("create(kind=\"observation\", content=\"{marker}\")"); diff --git a/crates/khive-pack-brain/src/handlers.rs b/crates/khive-pack-brain/src/handlers.rs index b160401b7..adfb2fe13 100644 --- a/crates/khive-pack-brain/src/handlers.rs +++ b/crates/khive-pack-brain/src/handlers.rs @@ -74,9 +74,13 @@ pub(crate) static BRAIN_HANDLERS: &[HandlerDef] = &[ plane (ADR-103 Stage 1, #724 Ask A); feedback_explicit events additionally split by \ served_by_profile_id; events carrying a work_class (today: phase_started / \ phase_completed / phase_cancelled payloads, checked before any future \ - payload.resource.work_class) split by counts_by_work_class. cost_unit is not \ - surfaced: it does not exist on any event payload yet (ADR-103 Stage 0 design-only) \ - and will be added once a resource payload carries it.", + payload.resource.work_class) split by counts_by_work_class. Events carrying \ + payload.resource.cost_unit (ADR-103 Amendment 1; stamped on every successful verb \ + dispatch since PR #927) sum into total_cost_unit and cost_unit_by_verb; events with \ + no resource.cost_unit (pre-Amendment-1 events, or errored/denied dispatches) simply \ + do not contribute. Both fields are omitted, not zero-filled, when no event in the \ + window carries cost_unit. When truncated=true, both sums are over the fetched page \ + only, same as the other counts_by_* fields.", visibility: khive_types::Visibility::Verb, category: khive_types::VerbCategory::Assertive, params: &[ @@ -680,14 +684,16 @@ impl BrainPack { // where a generic audit row carries a `resource` sub-object (ADR-103 // Stage 1's "resource payload enrichment", still open). Events with // neither path present are simply not counted in the split. - // `cost_unit` is NOT surfaced: as of this PR no event payload carries it - // anywhere in the codebase (ADR-103 Stage 0 defines it; Stage 1's - // resource-payload work has not landed a `cost_unit` field on any - // producer yet). Surfacing a field that no event can ever populate today - // would be a documentation lie dressed as a response key; once a - // producer emits `cost_unit` (under `payload.resource.cost_unit`, by the - // same convention as `work_class`), this verb should sum it the same way - // `counts_by_work_class` groups `work_class`. + // `cost_unit` (ADR-103 Amendment 1, `payload.resource.cost_unit`, + // stamped on every successful verb dispatch since PR #927 via + // `khive_runtime::cost_unit::resource_payload`) sums into + // `total_cost_unit` and `cost_unit_by_verb`, both omitted rather than + // zero-filled when no event in the window carries it. Amendment 1's + // "absence has exactly two meanings" rule means a missing + // `resource.cost_unit` is either a pre-Amendment-1 event or an + // errored/denied dispatch — this verb does not distinguish the two, it + // simply excludes the event from the sum, matching the `work_class` + // split's existing exclusion convention. // // Named `brain.event_counts` rather than the literal // `brain.events(...)` shape sketched on the issue — that name is already @@ -798,6 +804,9 @@ impl BrainPack { std::collections::BTreeMap::new(); let mut counts_by_work_class: std::collections::BTreeMap = std::collections::BTreeMap::new(); + let mut total_cost_unit: i64 = 0; + let mut cost_unit_by_verb: std::collections::BTreeMap = + std::collections::BTreeMap::new(); for event in &page.items { *counts_by_kind @@ -819,6 +828,11 @@ impl BrainPack { .entry(work_class.to_string()) .or_insert(0) += 1; } + if let Some(cost_unit) = event_cost_unit(&event.payload) { + total_cost_unit = total_cost_unit.saturating_add(cost_unit); + let entry = cost_unit_by_verb.entry(event.verb.clone()).or_insert(0); + *entry = entry.saturating_add(cost_unit); + } } let mut result = json!({ @@ -839,6 +853,10 @@ impl BrainPack { if !counts_by_work_class.is_empty() { result["counts_by_work_class"] = json!(counts_by_work_class); } + if !cost_unit_by_verb.is_empty() { + result["total_cost_unit"] = json!(total_cost_unit); + result["cost_unit_by_verb"] = json!(cost_unit_by_verb); + } Ok(result) } @@ -2299,10 +2317,10 @@ fn parse_rfc3339_micros(field: &'static str, value: &str) -> Result Option<&str> { }) } +/// Extract `cost_unit` from an event payload for `brain.event_counts`' +/// `total_cost_unit` / `cost_unit_by_verb` aggregation (ADR-103 Amendment 1). +/// +/// Reads `payload.resource.cost_unit` only — unlike `work_class`, `cost_unit` +/// has no top-level payload shape: it is emitted exclusively by +/// `khive_runtime::cost_unit::resource_payload` on the per-dispatch audit +/// row's `resource` sub-object, never by the Phase* background-work events. +/// +/// Returns `None` (silently excluded from the sum, not an error or a zero) +/// when the field is absent, which per Amendment 1's "absence has exactly +/// two meanings" rule is either a pre-Amendment-1 event or an +/// errored/denied dispatch. +fn event_cost_unit(payload: &Value) -> Option { + payload + .get("resource") + .and_then(|resource| resource.get("cost_unit")) + .and_then(Value::as_i64) +} + // ── lattice-router seam (#345 M1 / #352 M2) ────────────────────────────────── /// Context-vector dimension for the lattice-fann router seam. diff --git a/crates/khive-pack-brain/src/tests.rs b/crates/khive-pack-brain/src/tests.rs index 67807ad01..ea3db9044 100644 --- a/crates/khive-pack-brain/src/tests.rs +++ b/crates/khive-pack-brain/src/tests.rs @@ -6554,13 +6554,12 @@ mod event_counts_tests { ); } - /// `cost_unit` does not exist on any event payload in the codebase today (ADR-103 - /// Stage 0 defines it; no Stage 1 producer emits it yet). This verb must not invent - /// a `cost_unit` key — asserts its literal absence from the response shape so a - /// future producer landing the field is a deliberate response-shape change, not a - /// silent no-op this test would miss. + /// Events with no `payload.resource.cost_unit` anywhere in the window (pre-#927 + /// events, or events like Phase* spans that never carry `cost_unit` at all) must + /// not synthesize `total_cost_unit` / `cost_unit_by_verb` out of thin air — both + /// keys must be entirely absent, not zero-filled. #[tokio::test] - async fn cost_unit_is_not_present_in_the_response() { + async fn cost_unit_fields_absent_when_no_event_carries_cost_unit() { let (pack, rt) = make_pack(); let registry = empty_registry(); let token = rt.authorize(Namespace::local()).unwrap(); @@ -6575,6 +6574,17 @@ mod event_counts_tests { json!({"work_class": "warm", "phase": "ann_warm", "wall_us": 1, "cpu_us": null}), ) .await; + // Pre-#927 audit row: work_class present, no cost_unit at all. + seed_event( + &rt, + &token, + "search", + EventKind::Audit, + "lambda:khive", + 1_000_000, + json!({"resource": {"work_class": "interactive"}}), + ) + .await; let result = pack .dispatch( @@ -6587,12 +6597,209 @@ mod event_counts_tests { .expect("brain.event_counts must succeed"); assert!( - result.get("cost_unit").is_none(), - "cost_unit must not appear anywhere at the top level: {result}" + result.get("total_cost_unit").is_none(), + "total_cost_unit must be omitted, not zero, when no event carries cost_unit: {result}" + ); + assert!( + result.get("cost_unit_by_verb").is_none(), + "cost_unit_by_verb must be omitted, not an empty object, when no event carries \ + cost_unit: {result}" + ); + } + + /// ADR-103 Amendment 1 / PR #927: `payload.resource.cost_unit` on successful + /// dispatch audit rows must sum into `total_cost_unit` and split per-verb into + /// `cost_unit_by_verb`, following the same aggregation style as + /// `counts_by_kind`/`counts_by_actor`/`counts_by_work_class`. + #[tokio::test] + async fn cost_unit_sums_total_and_per_verb() { + let (pack, rt) = make_pack(); + let registry = empty_registry(); + let token = rt.authorize(Namespace::local()).unwrap(); + + seed_event( + &rt, + &token, + "create", + EventKind::Audit, + "lambda:a", + 1_000_000, + json!({"resource": {"work_class": "interactive", "cost_unit": 4}}), + ) + .await; + seed_event( + &rt, + &token, + "create", + EventKind::Audit, + "lambda:a", + 1_000_000, + json!({"resource": {"work_class": "interactive", "cost_unit": 6}}), + ) + .await; + seed_event( + &rt, + &token, + "stats", + EventKind::Audit, + "lambda:b", + 1_000_000, + json!({"resource": {"work_class": "interactive", "cost_unit": 1}}), + ) + .await; + // No cost_unit at all — must not contribute to either sum. + seed_event( + &rt, + &token, + "search", + EventKind::SearchExecuted, + "lambda:a", + 1_000_000, + json!({}), + ) + .await; + // Errored dispatch: work_class present, cost_unit absent per Amendment 1's + // "absence has exactly two meanings" rule — must not contribute. + let mut errored = Event::new( + token.namespace().as_str(), + "create", + EventKind::Audit, + SubstrateKind::Event, + "lambda:a", + ); + errored.created_at = 1_000_000; + errored.outcome = khive_types::EventOutcome::Error; + errored.payload = json!({"resource": {"work_class": "interactive"}}); + rt.events(&token) + .expect("event store") + .append_event(errored) + .await + .expect("seed errored audit event"); + + let result = pack + .dispatch( + "brain.event_counts", + json!({"since": micros_to_iso(0)}), + ®istry, + &token, + ) + .await + .expect("brain.event_counts must succeed"); + + assert_eq!(result["total"], json!(5), "got: {result}"); + assert_eq!( + result["total_cost_unit"], + json!(11), + "total_cost_unit must sum only the events carrying cost_unit (4+6+1): {result}" + ); + assert_eq!( + result["cost_unit_by_verb"]["create"], + json!(10), + "got: {result}" + ); + assert_eq!( + result["cost_unit_by_verb"]["stats"], + json!(1), + "got: {result}" ); assert!( - result.get("counts_by_cost_unit").is_none(), - "no cost_unit split must be surfaced until a producer emits the field: {result}" + result["cost_unit_by_verb"].get("search").is_none(), + "a verb whose events carry no cost_unit must not appear in the split: {result}" + ); + } + + /// The window filter (`since`/`until`) must apply to `cost_unit` aggregation + /// exactly like every other count field — an event outside the window must not + /// contribute to `total_cost_unit`. + #[tokio::test] + async fn cost_unit_respects_window_filter() { + let (pack, rt) = make_pack(); + let registry = empty_registry(); + let token = rt.authorize(Namespace::local()).unwrap(); + + seed_event( + &rt, + &token, + "create", + EventKind::Audit, + "lambda:a", + 1_500_000, + json!({"resource": {"work_class": "interactive", "cost_unit": 9}}), + ) + .await; + // Outside the window entirely. + seed_event( + &rt, + &token, + "create", + EventKind::Audit, + "lambda:a", + 5_000_000, + json!({"resource": {"work_class": "interactive", "cost_unit": 100}}), + ) + .await; + + let result = pack + .dispatch( + "brain.event_counts", + json!({ + "since": micros_to_iso(1_000_000), + "until": micros_to_iso(2_000_000), + }), + ®istry, + &token, + ) + .await + .expect("brain.event_counts must succeed"); + + assert_eq!(result["total"], json!(1), "got: {result}"); + assert_eq!(result["total_cost_unit"], json!(9), "got: {result}"); + } + + /// Mirrors `cost_unit.rs`'s own overflow-clamp tests: the sum must saturate + /// rather than panic when individual `cost_unit` values are extreme. + #[tokio::test] + async fn cost_unit_total_saturates_on_overflow_instead_of_panicking() { + let (pack, rt) = make_pack(); + let registry = empty_registry(); + let token = rt.authorize(Namespace::local()).unwrap(); + + seed_event( + &rt, + &token, + "knowledge.index", + EventKind::Audit, + "lambda:a", + 1_000_000, + json!({"resource": {"work_class": "interactive", "cost_unit": i64::MAX}}), + ) + .await; + seed_event( + &rt, + &token, + "knowledge.index", + EventKind::Audit, + "lambda:a", + 1_000_000, + json!({"resource": {"work_class": "interactive", "cost_unit": 5}}), + ) + .await; + + let result = pack + .dispatch( + "brain.event_counts", + json!({"since": micros_to_iso(0)}), + ®istry, + &token, + ) + .await + .expect("brain.event_counts must succeed"); + + assert_eq!(result["total_cost_unit"], json!(i64::MAX), "got: {result}"); + assert_eq!( + result["cost_unit_by_verb"]["knowledge.index"], + json!(i64::MAX), + "got: {result}" ); } diff --git a/crates/kkernel/src/code_ingest.rs b/crates/kkernel/src/code_ingest.rs index 4d0bf5d5e..2d49613eb 100644 --- a/crates/kkernel/src/code_ingest.rs +++ b/crates/kkernel/src/code_ingest.rs @@ -104,6 +104,16 @@ pub async fn run_code_ingest(args: CodeIngestArgs) -> Result<()> { /// after validation has already succeeded and a real (non-dry-run) write is /// about to occur. async fn code_ingest_batch(args: CodeIngestArgs) -> Result { + code_ingest_batch_with_runtime_setup(args, |_| Ok(())).await +} + +async fn code_ingest_batch_with_runtime_setup( + args: CodeIngestArgs, + runtime_setup: F, +) -> Result +where + F: FnOnce(&KhiveRuntime) -> Result<()>, +{ let bytes = std::fs::read(&args.findings) .with_context(|| format!("failed to read {}", args.findings.display()))?; @@ -161,6 +171,7 @@ async fn code_ingest_batch(args: CodeIngestArgs) -> Result { } let runtime = KhiveRuntime::new(cfg).map_err(|e| anyhow::anyhow!("{e}"))?; + runtime_setup(&runtime)?; let resolved_ns = runtime.config().default_namespace.clone(); let token = runtime .authorize(resolved_ns) @@ -479,10 +490,63 @@ fn wal_sidecar_path(db_path: &Path) -> PathBuf { #[cfg(test)] mod tests { + use std::sync::Arc; + + use async_trait::async_trait; + use khive_runtime::{EmbedderProvider, RuntimeError}; + use lattice_embed::{EmbedError, EmbeddingModel, EmbeddingService}; use serial_test::serial; use super::*; + struct FixedEmbeddingService { + dimensions: usize, + } + + #[async_trait] + impl EmbeddingService for FixedEmbeddingService { + async fn embed( + &self, + texts: &[String], + _model: EmbeddingModel, + ) -> std::result::Result>, EmbedError> { + Ok(texts + .iter() + .map(|_| vec![1.0_f32; self.dimensions]) + .collect()) + } + + fn supports_model(&self, _model: EmbeddingModel) -> bool { + true + } + + fn name(&self) -> &'static str { + "code-ingest-test" + } + } + + struct FixedEmbeddingProvider { + name: String, + dimensions: usize, + } + + #[async_trait] + impl EmbedderProvider for FixedEmbeddingProvider { + fn name(&self) -> &str { + &self.name + } + + fn dimensions(&self) -> usize { + self.dimensions + } + + async fn build(&self) -> std::result::Result, RuntimeError> { + Ok(Arc::new(FixedEmbeddingService { + dimensions: self.dimensions, + })) + } + } + fn base_args(findings: PathBuf, db: PathBuf) -> CodeIngestArgs { CodeIngestArgs { findings, @@ -847,9 +911,20 @@ mod tests { let findings = write_valid_findings(tmp.path()); let db = tmp.path().join("scratch.db"); - code_ingest_batch(base_args(findings, db.clone())) - .await - .expect("ingest must succeed"); + code_ingest_batch_with_runtime_setup(base_args(findings, db.clone()), |runtime| { + let model_names = runtime.registered_embedding_model_names(); + assert!( + !model_names.is_empty(), + "test requires at least one configured embedding model" + ); + for name in model_names { + let dimensions = runtime.resolve_embedding_model(Some(&name))?.dimensions(); + runtime.register_embedder(FixedEmbeddingProvider { name, dimensions }); + } + Ok(()) + }) + .await + .expect("ingest must succeed"); let cfg = resolve_runtime_config(RuntimeConfigInputs { db: Some(db.to_str().expect("utf8 path")), diff --git a/docs/guide/api-reference.md b/docs/guide/api-reference.md index 29cd58c6a..228b3182e 100644 --- a/docs/guide/api-reference.md +++ b/docs/guide/api-reference.md @@ -817,10 +817,14 @@ caller. Optional; load with `KHIVE_PACKS=kg,brain`. Windowed event counts grouped by kind, actor, and verb over the event plane (ADR-103 Stage 1, #724 Ask A). `feedback_explicit` events are additionally split by `served_by_profile_id`. Events carrying a `work_class` (today: -`phase_started`/`phase_completed`/`phase_cancelled` payloads) split by -`counts_by_work_class`. `cost_unit` is not surfaced: it does not exist on any event -payload yet (ADR-103 Stage 0 is design-only) and will be added once a resource payload -carries it. +`phase_started`/`phase_completed`/`phase_cancelled` payloads, or `payload.resource.work_class` +on a dispatch audit row) split by `counts_by_work_class`. Events carrying +`payload.resource.cost_unit` (ADR-103 Amendment 1, stamped on every successful verb dispatch +since PR #927) sum into `total_cost_unit` and `cost_unit_by_verb`; both are omitted, not +zero-filled, when no event in the window carries `cost_unit`. Events without a `cost_unit` +(pre-Amendment-1 events, or errored/denied dispatches) simply do not contribute. When +`truncated` is `true`, these sums are computed over the fetched page only, same as the other +`counts_by_*` fields. | Param | Type | Required | Notes | | ------- | ------ | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------- |