From affa66679a5e89669c60cea4f19eddb23acb020b Mon Sep 17 00:00:00 2001 From: Leo Date: Mon, 13 Jul 2026 12:43:34 -0400 Subject: [PATCH 1/6] feat(pack-kg): served-result telemetry on search Commit-early sweep: implementer leg ended without committing (build-lane contention). Local gates deferred to CI. Co-Authored-By: Claude Fable 5 --- crates/khive-pack-kg/src/handlers/search.rs | 108 +++++++++++++- crates/khive-pack-kg/tests/integration.rs | 154 ++++++++++++++++++++ 2 files changed, 260 insertions(+), 2 deletions(-) diff --git a/crates/khive-pack-kg/src/handlers/search.rs b/crates/khive-pack-kg/src/handlers/search.rs index de6a4fcbf..ae3716024 100644 --- a/crates/khive-pack-kg/src/handlers/search.rs +++ b/crates/khive-pack-kg/src/handlers/search.rs @@ -12,10 +12,12 @@ use std::collections::HashMap; /// specific query text to keep target records near the top of the ranking. const FILTERED_SCAN_CAP: u32 = 500; -use serde_json::Value; +use std::time::Instant; + +use serde_json::{json, Value}; use uuid::Uuid; -use khive_runtime::{NamespaceToken, RuntimeError, VerbRegistry}; +use khive_runtime::{KhiveRuntime, NamespaceToken, RuntimeError, VerbRegistry}; use khive_storage::types::PageRequest; use khive_storage::EntityFilter; @@ -32,6 +34,7 @@ impl KgPack { params: Value, registry: &VerbRegistry, ) -> Result { + let search_start = Instant::now(); let p: SearchParams = deser(params)?; let limit = p.limit.unwrap_or(10).min(100); let spec = resolve_kind_spec(&p.kind, registry)?; @@ -151,6 +154,13 @@ impl KgPack { }) }) .collect(); + self.track_search_serve( + token, + &p.query, + "entity", + &result, + search_start.elapsed().as_micros() as i64, + ); to_json(&result) } KindSpec::Note { specific } => { @@ -257,6 +267,13 @@ impl KgPack { }) }) .collect(); + self.track_search_serve( + token, + &p.query, + "note", + &result, + search_start.elapsed().as_micros() as i64, + ); to_json(&result) } KindSpec::Edge => Err(RuntimeError::InvalidInput( @@ -270,4 +287,91 @@ impl KgPack { )), } } + + /// Fire-and-forget `search_executed` telemetry (ADR-103 event plane), + /// mirroring `memory.recall`'s `track_recall_serve` seam (#866): the + /// event append runs off the response path via `track_background_task` + /// so a slow or failing event store never affects a served search. + fn track_search_serve( + &self, + token: &NamespaceToken, + query_raw: &str, + result_kind: &'static str, + results: &[Value], + latency_us: i64, + ) { + let selected: Vec = results + .iter() + .filter_map(|r| r.get("id").and_then(Value::as_str).map(str::to_string)) + .collect(); + let result_count = selected.len(); + let query = query_raw.to_string(); + let actor = format!("{}:{}", token.actor().kind, token.actor().id); + let runtime = self.runtime.clone(); + let token = token.clone(); + + khive_runtime::track_background_task(async move { + emit_search_executed_event( + &runtime, + &token, + actor, + query, + result_kind, + selected, + result_count, + latency_us, + ) + .await; + }); + } +} + +/// Append best-effort search telemetry without affecting the search response. +#[allow(clippy::too_many_arguments)] +async fn emit_search_executed_event( + rt: &KhiveRuntime, + token: &NamespaceToken, + actor: String, + query: String, + result_kind: &'static str, + selected: Vec, + result_count: usize, + latency_us: i64, +) { + let store = match rt.events(token) { + Ok(store) => store, + Err(err) => { + tracing::warn!( + error = %err, + namespace = token.namespace().as_str(), + event_kind = "search_executed", + "search_executed event store acquisition failed; search result is unaffected" + ); + return; + } + }; + let payload = json!({ + "actor": actor, + "served_by_profile_id": Value::Null, + "query": query, + "result_kind": result_kind, + "result_count": result_count, + "selected": selected, + "latency_us": latency_us, + }); + let event = khive_storage::Event::new( + token.namespace().as_str(), + "search", + khive_types::EventKind::SearchExecuted, + khive_types::SubstrateKind::Event, + actor, + ) + .with_payload(payload) + .with_duration_us(latency_us); + if let Err(err) = store.append_event(event).await { + tracing::warn!( + error = %err, + "search_executed event append failed; search result is unaffected" + ); + } } diff --git a/crates/khive-pack-kg/tests/integration.rs b/crates/khive-pack-kg/tests/integration.rs index f14240c1b..2a9fc410d 100644 --- a/crates/khive-pack-kg/tests/integration.rs +++ b/crates/khive-pack-kg/tests/integration.rs @@ -10952,3 +10952,157 @@ async fn list_proposal_limit_over_cap_reports_effective_limit() { assert_eq!(response["effective_limit"], 500); assert_eq!(response["limit_clamped"], true); } + +// ── #806: `search_executed` event-plane emission ──────────────────────────── +// +// Mirrors memory.recall's `#866` `recall_executed` regression +// (khive-pack-memory/src/handlers/recall.rs's +// `recall_emits_exactly_one_recall_executed_event`): `search` must append +// exactly one `SearchExecuted` event per served search, off the response +// path via `track_background_task`, so poll briefly instead of assuming it +// has landed by the time `search` returns. + +async fn poll_search_executed_events( + store: &std::sync::Arc, +) -> Vec { + for _ in 0..100 { + let page = store + .query_events( + khive_storage::EventFilter { + kinds: vec![khive_types::EventKind::SearchExecuted], + ..Default::default() + }, + khive_storage::types::PageRequest { + limit: 50, + offset: 0, + }, + ) + .await + .expect("query_events"); + if !page.items.is_empty() { + return page.items; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + Vec::new() +} + +#[tokio::test] +async fn search_entity_emits_exactly_one_search_executed_event() { + let rt = KhiveRuntime::memory().expect("in-memory runtime must succeed"); + let ns = Namespace::local(); + let token = rt.authorize(ns.clone()).expect("authorize local"); + + rt.create_entity( + &token, + "concept", + None, + "806 search executed entity", + None, + None, + vec![], + ) + .await + .expect("create entity"); + + let mut builder = VerbRegistryBuilder::new(); + builder.register(KgPack::new(rt.clone())); + let registry = builder.build().expect("registry builds"); + + let result = registry + .dispatch( + "search", + json!({"kind": "entity", "query": "806 search executed entity"}), + ) + .await + .expect("search must succeed"); + let hits = result.as_array().expect("bare array result"); + assert!(!hits.is_empty(), "must find the seeded entity"); + + let store = rt.events(&token).expect("event store for local namespace"); + let search_events = poll_search_executed_events(&store).await; + + assert_eq!( + search_events.len(), + 1, + "exactly one search_executed event per served search, got: {search_events:?}" + ); + let event = &search_events[0]; + assert_eq!(event.kind, khive_types::EventKind::SearchExecuted); + assert_eq!(event.verb, "search"); + assert_eq!(event.payload["result_kind"], json!("entity")); + assert_eq!(event.payload["result_count"], json!(hits.len())); + assert_eq!(event.payload["query"], json!("806 search executed entity")); + assert_eq!( + event.payload["served_by_profile_id"], + Value::Null, + "kg search has no profile resolution — the field must stay present but null" + ); + assert!( + event.payload["latency_us"] + .as_i64() + .is_some_and(|us| us >= 0), + "latency_us must be a non-negative measured duration, got: {:?}", + event.payload["latency_us"] + ); + assert!( + event.payload["actor"] + .as_str() + .is_some_and(|s| !s.is_empty()), + "actor must be stamped, got: {:?}", + event.payload["actor"] + ); + let selected = event.payload["selected"] + .as_array() + .expect("selected must be a UUID array"); + assert_eq!( + selected.len(), + hits.len(), + "selected must carry the full served result-ID list" + ); +} + +#[tokio::test] +async fn search_note_emits_exactly_one_search_executed_event_with_note_result_kind() { + let rt = KhiveRuntime::memory().expect("in-memory runtime must succeed"); + let ns = Namespace::local(); + let token = rt.authorize(ns.clone()).expect("authorize local"); + + rt.create_note( + &token, + "observation", + None, + "806 search executed note unique_marker_9931", + None, + None, + vec![], + ) + .await + .expect("create note"); + + let mut builder = VerbRegistryBuilder::new(); + builder.register(KgPack::new(rt.clone())); + let registry = builder.build().expect("registry builds"); + + let result = registry + .dispatch( + "search", + json!({"kind": "note", "query": "unique_marker_9931"}), + ) + .await + .expect("search must succeed"); + let hits = result.as_array().expect("bare array result"); + assert!(!hits.is_empty(), "must find the seeded note"); + + let store = rt.events(&token).expect("event store for local namespace"); + let search_events = poll_search_executed_events(&store).await; + + assert_eq!( + search_events.len(), + 1, + "exactly one search_executed event per served search, got: {search_events:?}" + ); + let event = &search_events[0]; + assert_eq!(event.payload["result_kind"], json!("note")); + assert_eq!(event.payload["result_count"], json!(hits.len())); +} From ddfdbb1f3a48c2aae4305b82e67f8411c46191ab Mon Sep 17 00:00:00 2001 From: Leo Date: Mon, 13 Jul 2026 17:49:45 -0400 Subject: [PATCH 2/6] test(mcp): keep schedule routing regression hermetic --- crates/khive-mcp/src/serve.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) 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}\")"); From 7871897f1dde987658144b9a30e6ed37b94dfe46 Mon Sep 17 00:00:00 2001 From: Leo Date: Mon, 13 Jul 2026 17:02:23 -0400 Subject: [PATCH 3/6] test(khive-db): serialize writer-task tests on tx_registry to fix flaky tx-age sweep run_writer_task registers a writer_task_tx handle in the process-wide tx_registry singleton; the checkpoint tx_age_sweep_* tests read that singleton under #[serial(tx_registry)]. The writer_task.rs tests that spawn a writer task were missing from the group, leaking a longer-lived writer_task_tx into the sweep's oldest() read and intermittently failing the assertion on main CI. Join them to the serial group. No prod change. --- crates/khive-db/src/writer_task.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) 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 From 1e789bb75d19bdef651fe5f9b8ecd2b7cc8ee5d3 Mon Sep 17 00:00:00 2001 From: Leo Date: Mon, 13 Jul 2026 17:56:25 -0400 Subject: [PATCH 4/6] test(kkernel): make code-ingest vector test hermetic --- crates/kkernel/src/code_ingest.rs | 81 +++++++++++++++++++++++++++++-- 1 file changed, 78 insertions(+), 3 deletions(-) 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")), From 6dd931e75f1f8599b89710aec6e580254ee6b386 Mon Sep 17 00:00:00 2001 From: Leo Date: Tue, 14 Jul 2026 08:58:19 -0400 Subject: [PATCH 5/6] fix(kg): preserve typed search provenance (#973) --- crates/khive-db/src/stores/event.rs | 57 +++++++++-- crates/khive-db/src/stores/event_tests.rs | 23 +++++ crates/khive-pack-brain/src/tests.rs | 3 + crates/khive-pack-kg/src/handlers/search.rs | 1 + crates/khive-pack-kg/tests/integration.rs | 107 +++++++++++++++++++- crates/khive-runtime/tests/integration.rs | 1 + implementation_notes.md | 20 ++++ 7 files changed, 200 insertions(+), 12 deletions(-) create mode 100644 implementation_notes.md diff --git a/crates/khive-db/src/stores/event.rs b/crates/khive-db/src/stores/event.rs index f1a85337b..4e36990d4 100644 --- a/crates/khive-db/src/stores/event.rs +++ b/crates/khive-db/src/stores/event.rs @@ -438,7 +438,8 @@ pub async fn append_event_on_writer( fn decode_event_observations(event: &Event) -> Result, rusqlite::Error> { match event.kind { EventKind::RerankExecuted => decode_rerank_observations(event), - EventKind::RecallExecuted | EventKind::SearchExecuted => decode_recall_observations(event), + EventKind::RecallExecuted => decode_recall_observations(event), + EventKind::SearchExecuted => decode_search_observations(event), EventKind::LinkCreated => decode_link_observations(event), EventKind::EntityCreated | EventKind::EntityUpdated @@ -552,7 +553,10 @@ fn payload_uuid(event: &Event, field: &'static str) -> Result, rusq .map_err(|e| invalid_payload(event.kind, field, e)) } -fn decode_candidate_observations(event: &Event) -> Result, rusqlite::Error> { +fn decode_candidate_observations( + event: &Event, + referent_kind: ReferentKind, +) -> Result, rusqlite::Error> { let mut rows = Vec::new(); for (position, entity_id) in payload_uuid_array(event, "candidates")? @@ -569,7 +573,7 @@ fn decode_candidate_observations(event: &Event) -> Result, rows.push(EventObservation { event_id: event.id, entity_id, - referent_kind: ReferentKind::Note, + referent_kind, role: ObservationRole::Candidate, position: position_u32, }); @@ -581,6 +585,7 @@ fn decode_candidate_observations(event: &Event) -> Result, fn push_selected_observations( event: &Event, selected: Vec, + referent_kind: ReferentKind, rows: &mut Vec, ) -> Result<(), rusqlite::Error> { for (position, entity_id) in selected.into_iter().enumerate() { @@ -594,7 +599,7 @@ fn push_selected_observations( rows.push(EventObservation { event_id: event.id, entity_id, - referent_kind: ReferentKind::Note, + referent_kind, role: ObservationRole::Selected, position: position_u32, }); @@ -602,13 +607,43 @@ fn push_selected_observations( Ok(()) } -/// `RecallExecuted`/`SearchExecuted` payloads carry a flat `selected: Vec` -/// field (ADR-041 §"Projection rules"). These payloads are untyped JSON; the -/// ADR-041 projection contract makes `selected` the only field consulted here. +/// `RecallExecuted` payloads carry flat candidate and selected note UUID lists. fn decode_recall_observations(event: &Event) -> Result, rusqlite::Error> { - let mut rows = decode_candidate_observations(event)?; + let mut rows = decode_candidate_observations(event, ReferentKind::Note)?; + let selected = payload_uuid_array_opt(event, "selected")?.unwrap_or_default(); + push_selected_observations(event, selected, ReferentKind::Note, &mut rows)?; + Ok(rows) +} + +/// `SearchExecuted.result_kind` identifies which substrate owns every UUID in +/// the candidate and selected lists. Rejecting missing or unknown values keeps +/// the append-only projection from persisting an untyped reference. +fn decode_search_observations(event: &Event) -> Result, rusqlite::Error> { + let referent_kind = match event + .payload + .get("result_kind") + .and_then(|value| value.as_str()) + { + Some("entity") => ReferentKind::Entity, + Some("note") => ReferentKind::Note, + Some(_) => { + return Err(invalid_payload( + event.kind, + "result_kind", + "expected \"entity\" or \"note\"", + )); + } + None => { + return Err(invalid_payload( + event.kind, + "result_kind", + "expected string \"entity\" or \"note\"", + )); + } + }; + let mut rows = decode_candidate_observations(event, referent_kind)?; let selected = payload_uuid_array_opt(event, "selected")?.unwrap_or_default(); - push_selected_observations(event, selected, &mut rows)?; + push_selected_observations(event, selected, referent_kind, &mut rows)?; Ok(rows) } @@ -622,11 +657,11 @@ fn decode_recall_observations(event: &Event) -> Result, ru /// absent (`None`) — a present-but-malformed `final_scores` errors /// immediately instead of masking the problem by falling through. fn decode_rerank_observations(event: &Event) -> Result, rusqlite::Error> { - let mut rows = decode_candidate_observations(event)?; + let mut rows = decode_candidate_observations(event, ReferentKind::Note)?; let selected = payload_final_scores_uuid_array_opt(event, "final_scores")? .or(payload_reranked_uuid_array_opt(event, "reranked")?) .unwrap_or_default(); - push_selected_observations(event, selected, &mut rows)?; + push_selected_observations(event, selected, ReferentKind::Note, &mut rows)?; Ok(rows) } diff --git a/crates/khive-db/src/stores/event_tests.rs b/crates/khive-db/src/stores/event_tests.rs index 02bc5f833..d60cd4bdc 100644 --- a/crates/khive-db/src/stores/event_tests.rs +++ b/crates/khive-db/src/stores/event_tests.rs @@ -25,6 +25,7 @@ fn make_event(namespace: &str) -> Event { SubstrateKind::Note, "agent:test", ) + .with_payload(json!({ "result_kind": "note" })) } #[tokio::test] @@ -146,6 +147,7 @@ async fn append_event_writes_observations_atomically() { let mut event = make_event("default"); event.kind = EventKind::SearchExecuted; event.payload = json!({ + "result_kind": "note", "candidates": [candidate.to_string()], "selected": [selected.to_string()], "served_by_profile_id": "profile-a" @@ -187,6 +189,25 @@ async fn append_event_writes_observations_atomically() { assert_eq!(selected_count, 1, "expected one selected observation row"); } +#[tokio::test] +async fn search_executed_rejects_unknown_result_kind() { + let store = setup_memory_store(); + let mut event = make_event("default"); + event.payload = json!({ + "result_kind": "edge", + "candidates": [Uuid::new_v4().to_string()], + "selected": [] + }); + let event_id = event.id; + + let result = store.append_event(event).await; + assert!(result.is_err(), "unknown result_kind must be rejected"); + assert!( + store.get_event(event_id).await.unwrap().is_none(), + "invalid event and projection must roll back atomically" + ); +} + async fn selected_uuids_for(store: &SqlEventStore, event_id: Uuid) -> Vec { let pool = Arc::clone(&store.pool); let event_id_str = event_id.to_string(); @@ -647,6 +668,7 @@ async fn query_events_filters_by_observed() { let mut event = make_event("default"); event.kind = EventKind::SearchExecuted; event.payload = json!({ + "result_kind": "entity", "candidates": [entity_id.to_string()], "selected": [] }); @@ -677,6 +699,7 @@ async fn query_events_filters_by_selected() { let mut event = make_event("default"); event.kind = EventKind::SearchExecuted; event.payload = json!({ + "result_kind": "entity", "candidates": [], "selected": [entity_id.to_string()] }); diff --git a/crates/khive-pack-brain/src/tests.rs b/crates/khive-pack-brain/src/tests.rs index 67807ad01..5fb26afd1 100644 --- a/crates/khive-pack-brain/src/tests.rs +++ b/crates/khive-pack-brain/src/tests.rs @@ -5832,6 +5832,9 @@ mod event_counts_tests { ); event.created_at = created_at; event.payload = payload; + if kind == EventKind::SearchExecuted && event.payload.get("result_kind").is_none() { + event.payload["result_kind"] = json!("note"); + } rt.events(token) .expect("event store") .append_event(event) diff --git a/crates/khive-pack-kg/src/handlers/search.rs b/crates/khive-pack-kg/src/handlers/search.rs index ae3716024..182eb9389 100644 --- a/crates/khive-pack-kg/src/handlers/search.rs +++ b/crates/khive-pack-kg/src/handlers/search.rs @@ -356,6 +356,7 @@ async fn emit_search_executed_event( "query": query, "result_kind": result_kind, "result_count": result_count, + "candidates": selected, "selected": selected, "latency_us": latency_us, }); diff --git a/crates/khive-pack-kg/tests/integration.rs b/crates/khive-pack-kg/tests/integration.rs index 2a9fc410d..c5fbe03a6 100644 --- a/crates/khive-pack-kg/tests/integration.rs +++ b/crates/khive-pack-kg/tests/integration.rs @@ -11,7 +11,7 @@ use khive_runtime::{ EntityCreateSpec, KhiveRuntime, Namespace, NamespaceToken, ParamDef, RuntimeError, VerbCategory, VerbRegistry, VerbRegistryBuilder, Visibility, }; -use khive_storage::Note; +use khive_storage::{Note, SqlStatement, SqlValue}; use khive_types::Pack; use serde_json::{json, Value}; @@ -10987,6 +10987,109 @@ async fn poll_search_executed_events( Vec::new() } +async fn assert_search_projection( + rt: &KhiveRuntime, + store: &std::sync::Arc, + event: &khive_storage::Event, + hits: &[Value], + referent_kind: &str, +) { + let expected_ids: Vec = hits + .iter() + .map(|hit| { + hit["id"] + .as_str() + .expect("search hit id must be a UUID string") + .to_string() + }) + .collect(); + assert_eq!(event.payload["candidates"], json!(expected_ids)); + assert_eq!(event.payload["selected"], json!(expected_ids)); + + let mut reader = rt.sql().reader().await.expect("sql reader must open"); + let rows = reader + .query_all(SqlStatement { + sql: "SELECT entity_id, referent_kind, role, position \ + FROM event_observations WHERE event_id = ?1 \ + ORDER BY CASE role WHEN 'candidate' THEN 0 ELSE 1 END, position" + .into(), + params: vec![SqlValue::Text(event.id.to_string())], + label: Some("search_observation_projection".into()), + }) + .await + .expect("projection query must succeed"); + + let projected: Vec<(String, String, String, i64)> = rows + .iter() + .map(|row| { + let text = |column| match row.get(column) { + Some(SqlValue::Text(value)) => value.clone(), + other => panic!("{column} must be text, got {other:?}"), + }; + let position = match row.get("position") { + Some(SqlValue::Integer(value)) => *value, + other => panic!("position must be integer, got {other:?}"), + }; + ( + text("entity_id"), + text("referent_kind"), + text("role"), + position, + ) + }) + .collect(); + let expected_projection: Vec<(String, String, String, i64)> = ["candidate", "selected"] + .into_iter() + .flat_map(|role| { + expected_ids.iter().enumerate().map(move |(position, id)| { + ( + id.clone(), + referent_kind.to_string(), + role.to_string(), + position as i64, + ) + }) + }) + .collect(); + assert_eq!(projected, expected_projection); + + let first_id = uuid::Uuid::parse_str(&expected_ids[0]).expect("search hit id must parse"); + for (filter_name, filter) in [ + ( + "observed", + khive_storage::EventFilter { + kinds: vec![khive_types::EventKind::SearchExecuted], + observed: vec![first_id], + ..Default::default() + }, + ), + ( + "selected", + khive_storage::EventFilter { + kinds: vec![khive_types::EventKind::SearchExecuted], + selected: vec![first_id], + ..Default::default() + }, + ), + ] { + let page = store + .query_events( + filter, + khive_storage::types::PageRequest { + limit: 10, + offset: 0, + }, + ) + .await + .unwrap_or_else(|error| panic!("{filter_name} filter failed: {error}")); + assert_eq!( + page.items.iter().map(|item| item.id).collect::>(), + vec![event.id], + "{filter_name} must find the search event through its projection" + ); + } +} + #[tokio::test] async fn search_entity_emits_exactly_one_search_executed_event() { let rt = KhiveRuntime::memory().expect("in-memory runtime must succeed"); @@ -11060,6 +11163,7 @@ async fn search_entity_emits_exactly_one_search_executed_event() { hits.len(), "selected must carry the full served result-ID list" ); + assert_search_projection(&rt, &store, event, hits, "entity").await; } #[tokio::test] @@ -11105,4 +11209,5 @@ async fn search_note_emits_exactly_one_search_executed_event_with_note_result_ki let event = &search_events[0]; assert_eq!(event.payload["result_kind"], json!("note")); assert_eq!(event.payload["result_count"], json!(hits.len())); + assert_search_projection(&rt, &store, event, hits, "note").await; } diff --git a/crates/khive-runtime/tests/integration.rs b/crates/khive-runtime/tests/integration.rs index bc37c80f3..ca7026b30 100644 --- a/crates/khive-runtime/tests/integration.rs +++ b/crates/khive-runtime/tests/integration.rs @@ -1016,6 +1016,7 @@ async fn synthetic_edge_observed_as_selected_returns_memory_note() { "agent:test", ); event.payload = serde_json::json!({ + "result_kind": "note", "candidates": [], "selected": [memory_id.to_string()] }); diff --git a/implementation_notes.md b/implementation_notes.md new file mode 100644 index 000000000..e4883f36f --- /dev/null +++ b/implementation_notes.md @@ -0,0 +1,20 @@ +# Implementation notes + +- `crates/khive-pack-kg/src/handlers/search.rs` now records the ordered served UUIDs as both `candidates` and `selected` in every `SearchExecuted` payload. +- `crates/khive-db/src/stores/event.rs` now decodes `SearchExecuted` separately from recall, validates `result_kind`, and projects entity searches as entity referents and note searches as note referents. +- KG integration coverage now verifies payload completeness, Candidate/Selected roles, referent kinds, positions, and `observed`/`selected` event-filter queryability for both entity and note searches. +- Storage coverage rejects unknown search result kinds atomically; existing synthetic search events now declare their result substrate explicitly. + +## Verification + +- `cargo fmt --manifest-path crates/Cargo.toml --all -- --check` +- `cargo test --manifest-path crates/Cargo.toml -p khive-pack-kg --test integration search_entity_emits_exactly_one_search_executed_event -- --exact` +- `cargo test --manifest-path crates/Cargo.toml -p khive-pack-kg --test integration search_note_emits_exactly_one_search_executed_event_with_note_result_kind -- --exact` +- `cargo test --manifest-path crates/Cargo.toml -p khive-db search_executed_rejects_unknown_result_kind` +- `cargo test --manifest-path crates/Cargo.toml -p khive-db stores::event::tests` (32 passed) +- `cargo test --manifest-path crates/Cargo.toml -p khive-runtime synthetic_edge_observed_as_selected_returns_memory_note -- --exact` +- `cargo test --manifest-path crates/Cargo.toml -p khive-pack-brain event_counts_tests` (15 passed) + +## Domain utility + +`high` — ADR-041 directly specifies the Candidate/Selected role mapping, ordered positions, and typed referent contract implemented by this fix. From 774f50df528e031f958c52759ed74c2f13d8e63d Mon Sep 17 00:00:00 2001 From: Leo Date: Tue, 14 Jul 2026 10:38:37 -0400 Subject: [PATCH 6/6] chore: remove workflow scratch notes from tree Merge main and drop implementation_notes.md, a stray process-notes file that does not belong in the source tree. --- implementation_notes.md | 20 -------------------- 1 file changed, 20 deletions(-) delete mode 100644 implementation_notes.md diff --git a/implementation_notes.md b/implementation_notes.md deleted file mode 100644 index e4883f36f..000000000 --- a/implementation_notes.md +++ /dev/null @@ -1,20 +0,0 @@ -# Implementation notes - -- `crates/khive-pack-kg/src/handlers/search.rs` now records the ordered served UUIDs as both `candidates` and `selected` in every `SearchExecuted` payload. -- `crates/khive-db/src/stores/event.rs` now decodes `SearchExecuted` separately from recall, validates `result_kind`, and projects entity searches as entity referents and note searches as note referents. -- KG integration coverage now verifies payload completeness, Candidate/Selected roles, referent kinds, positions, and `observed`/`selected` event-filter queryability for both entity and note searches. -- Storage coverage rejects unknown search result kinds atomically; existing synthetic search events now declare their result substrate explicitly. - -## Verification - -- `cargo fmt --manifest-path crates/Cargo.toml --all -- --check` -- `cargo test --manifest-path crates/Cargo.toml -p khive-pack-kg --test integration search_entity_emits_exactly_one_search_executed_event -- --exact` -- `cargo test --manifest-path crates/Cargo.toml -p khive-pack-kg --test integration search_note_emits_exactly_one_search_executed_event_with_note_result_kind -- --exact` -- `cargo test --manifest-path crates/Cargo.toml -p khive-db search_executed_rejects_unknown_result_kind` -- `cargo test --manifest-path crates/Cargo.toml -p khive-db stores::event::tests` (32 passed) -- `cargo test --manifest-path crates/Cargo.toml -p khive-runtime synthetic_edge_observed_as_selected_returns_memory_note -- --exact` -- `cargo test --manifest-path crates/Cargo.toml -p khive-pack-brain event_counts_tests` (15 passed) - -## Domain utility - -`high` — ADR-041 directly specifies the Candidate/Selected role mapping, ordered positions, and typed referent contract implemented by this fix.