feat(pack-kg): served-result telemetry on search

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<Vec<EventObservation>, 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<Option<Uuid>, rusq
         .map_err(|e| invalid_payload(event.kind, field, e))
 }
 
-fn decode_candidate_observations(event: &Event) -> Result<Vec<EventObservation>, rusqlite::Error> {
+fn decode_candidate_observations(
+    event: &Event,
+    referent_kind: ReferentKind,
+) -> Result<Vec<EventObservation>, 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<Vec<EventObservation>,
         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<Vec<EventObservation>,
 fn push_selected_observations(
     event: &Event,
     selected: Vec<Uuid>,
+    referent_kind: ReferentKind,
     rows: &mut Vec<EventObservation>,
 ) -> Result<(), rusqlite::Error> {
     for (position, entity_id) in selected.into_iter().enumerate() {
@@ -594,21 +599,51 @@ 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,
         });
     }
     Ok(())
 }
 
-/// `RecallExecuted`/`SearchExecuted` payloads carry a flat `selected: Vec<Uuid>`
-/// 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<Vec<EventObservation>, 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<Vec<EventObservation>, 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<Vec<EventObservation>, 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<Vec<EventObservation>, 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)
 }

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,43 @@ 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"
+    );
+}
+
+#[tokio::test]
+async fn search_executed_rejects_absent_result_kind() {
+    let store = setup_memory_store();
+    let mut event = make_event("default");
+    event.payload = json!({
+        "candidates": [Uuid::new_v4().to_string()],
+        "selected": []
+    });
+    let event_id = event.id;
+
+    let result = store.append_event(event).await;
+    assert!(result.is_err(), "absent 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<String> {
     let pool = Arc::clone(&store.pool);
     let event_id_str = event_id.to_string();
@@ -647,6 +686,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 +717,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()]
     });

crates/khive-pack-brain/src/tests.rs

@@ -5896,6 +5896,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)

crates/khive-pack-kg/src/handlers/search.rs

@@ -6,10 +6,12 @@ use std::collections::HashMap;
 /// See `docs/api/scan-cliff.md`.
 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;
 
@@ -26,6 +28,7 @@ impl KgPack {
         params: Value,
         registry: &VerbRegistry,
     ) -> Result<Value, RuntimeError> {
+        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)?;
@@ -141,6 +144,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 } => {
@@ -242,6 +252,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(
@@ -255,4 +272,92 @@ 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<String> = 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<String>,
+    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,
+        "candidates": selected,
+        "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"
+        );
+    }
 }