From cbf23fabfcef03ee1d7f127895cc785f7413a457 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 30 Jun 2026 14:55:32 +0800 Subject: [PATCH 1/7] {"schema":"decodex/commit/1","summary":"Split progressive search storage modules","authority":"manual"} --- .../src/progressive_search/storage.rs | 277 +----------------- .../src/progressive_search/storage/hash.rs | 12 + .../src/progressive_search/storage/hits.rs | 92 ++++++ .../src/progressive_search/storage/session.rs | 177 +++++++++++ 4 files changed, 287 insertions(+), 271 deletions(-) create mode 100644 packages/elf-service/src/progressive_search/storage/hash.rs create mode 100644 packages/elf-service/src/progressive_search/storage/hits.rs create mode 100644 packages/elf-service/src/progressive_search/storage/session.rs diff --git a/packages/elf-service/src/progressive_search/storage.rs b/packages/elf-service/src/progressive_search/storage.rs index ac5db23e..54459495 100644 --- a/packages/elf-service/src/progressive_search/storage.rs +++ b/packages/elf-service/src/progressive_search/storage.rs @@ -1,273 +1,8 @@ -use std::{ - collections::hash_map::DefaultHasher, - hash::{Hash, Hasher}, - str::FromStr, -}; - -use serde_json; -use sqlx::PgExecutor; -use time::{Duration, OffsetDateTime}; -use uuid::Uuid; +mod hash; +mod hits; +mod session; -use crate::{ - Error, Result, - progressive_search::types::{ - HitItem, NewSearchSession, SESSION_ABSOLUTE_TTL_HOURS, SESSION_SLIDING_TTL_HOURS, - SearchSession, SearchSessionItemRecord, SearchSessionMode, SearchSessionRow, - }, +pub(super) use self::{ + hits::record_detail_hits, + session::{load_search_session, store_search_session, touch_search_session}, }; -use elf_domain::english_gate; - -pub(super) async fn store_search_session<'e, E>( - executor: E, - session: NewSearchSession<'_>, -) -> Result<()> -where - E: PgExecutor<'e>, -{ - let items_json = serde_json::to_value(session.items).map_err(|err| Error::Storage { - message: format!("Failed to encode search session items: {err}"), - })?; - let query_plan_json = - session.query_plan.map(serde_json::to_value).transpose().map_err(|err| Error::Storage { - message: format!("Failed to encode search session query plan: {err}"), - })?; - let trajectory_summary_json = - session.trajectory_summary.map(serde_json::to_value).transpose().map_err(|err| { - Error::Storage { - message: format!("Failed to encode search session trajectory summary: {err}"), - } - })?; - - sqlx::query( - "\ -INSERT INTO search_sessions ( - search_session_id, - trace_id, - tenant_id, - project_id, - agent_id, - read_profile, - query, - mode, - trajectory_summary, - query_plan, - items, - created_at, - expires_at -) -VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)", - ) - .bind(session.search_session_id) - .bind(session.trace_id) - .bind(session.tenant_id.trim()) - .bind(session.project_id.trim()) - .bind(session.agent_id.trim()) - .bind(session.read_profile) - .bind(session.query) - .bind(session.mode.as_str()) - .bind(trajectory_summary_json) - .bind(query_plan_json) - .bind(items_json) - .bind(session.created_at) - .bind(session.expires_at) - .execute(executor) - .await?; - - Ok(()) -} - -pub(super) async fn load_search_session<'e, E>( - executor: E, - search_session_id: Uuid, - now: OffsetDateTime, -) -> Result -where - E: PgExecutor<'e>, -{ - let row = sqlx::query_as::<_, SearchSessionRow>( - "\ -SELECT - search_session_id, - trace_id, - tenant_id, - project_id, - agent_id, - read_profile, - query, - mode, - trajectory_summary, - query_plan, - items, - created_at, - expires_at -FROM search_sessions -WHERE search_session_id = $1", - ) - .bind(search_session_id) - .fetch_optional(executor) - .await?; - let Some(row) = row else { - return Err(Error::InvalidRequest { message: "Unknown search_session_id.".to_string() }); - }; - let expires_at: OffsetDateTime = row.expires_at; - - if expires_at <= now { - return Err(Error::InvalidRequest { message: "Search session expired.".to_string() }); - } - - let items: Vec = serde_json::from_value(row.items).map_err(|err| { - Error::Storage { message: format!("Failed to decode search session items: {err}") } - })?; - let mode = SearchSessionMode::from_str(row.mode.as_str())?; - let query_plan = match row.query_plan { - Some(value) => Some(serde_json::from_value(value).map_err(|err| Error::Storage { - message: format!("Failed to decode search session query_plan: {err}"), - })?), - None => None, - }; - let trajectory_summary = match row.trajectory_summary { - Some(value) => Some(serde_json::from_value(value).map_err(|err| Error::Storage { - message: format!("Failed to decode search session trajectory summary: {err}"), - })?), - None => None, - }; - - Ok(SearchSession { - search_session_id: row.search_session_id, - trace_id: row.trace_id, - tenant_id: row.tenant_id, - project_id: row.project_id, - agent_id: row.agent_id, - read_profile: row.read_profile, - query: row.query, - items, - mode, - trajectory_summary, - query_plan, - created_at: row.created_at, - expires_at, - }) -} - -pub(super) async fn touch_search_session<'e, E>( - executor: E, - session: &SearchSession, - now: OffsetDateTime, -) -> Result -where - E: PgExecutor<'e>, -{ - let absolute_expires_at = session.created_at + Duration::hours(SESSION_ABSOLUTE_TTL_HOURS); - let sliding_expires_at = now + Duration::hours(SESSION_SLIDING_TTL_HOURS); - let touched = if sliding_expires_at < absolute_expires_at { - sliding_expires_at - } else { - absolute_expires_at - }; - - if touched <= session.expires_at { - return Ok(session.expires_at); - } - - sqlx::query( - "UPDATE search_sessions SET expires_at = $1 WHERE search_session_id = $2 AND expires_at < $1", - ) - .bind(touched) - .bind(session.search_session_id) - .execute(executor) - .await?; - - Ok(touched) -} - -pub(super) async fn record_detail_hits<'e, E>( - executor: E, - query: &str, - items: &[HitItem], - now: OffsetDateTime, -) -> Result<()> -where - E: PgExecutor<'e>, -{ - if !english_gate::is_english_natural_language(query) { - return Err(Error::NonEnglishInput { field: "$.query".to_string() }); - } - - let query_hash = hash_query(query); - let mut hit_ids = Vec::with_capacity(items.len()); - let mut note_ids = Vec::with_capacity(items.len()); - let mut chunk_ids = Vec::with_capacity(items.len()); - let mut ranks = Vec::with_capacity(items.len()); - let mut final_scores = Vec::with_capacity(items.len()); - - for item in items { - let rank = i32::try_from(item.rank).map_err(|_| Error::InvalidRequest { - message: "Search session rank is out of range.".to_string(), - })?; - - hit_ids.push(Uuid::new_v4()); - note_ids.push(item.note_id); - chunk_ids.push(item.chunk_id); - ranks.push(rank); - final_scores.push(item.final_score); - } - - sqlx::query( - "\ -WITH hits AS ( - SELECT * - FROM unnest( - $1::uuid[], - $2::uuid[], - $3::uuid[], - $4::int4[], - $5::real[] -) AS t(hit_id, note_id, chunk_id, rank, final_score) -), -updated AS ( -UPDATE memory_notes -SET - hit_count = hit_count + 1, - last_hit_at = $6 -WHERE note_id = ANY($2) -) -INSERT INTO memory_hits ( - hit_id, - note_id, - chunk_id, - query_hash, - rank, - final_score, - ts -) -SELECT - hit_id, - note_id, - chunk_id, - $7, - rank, - final_score, - $6 -FROM hits", - ) - .bind(&hit_ids) - .bind(¬e_ids) - .bind(&chunk_ids) - .bind(&ranks) - .bind(&final_scores) - .bind(now) - .bind(query_hash.as_str()) - .execute(executor) - .await?; - - Ok(()) -} - -fn hash_query(query: &str) -> String { - let mut hasher = DefaultHasher::new(); - - Hash::hash(query, &mut hasher); - - format!("{:x}", hasher.finish()) -} diff --git a/packages/elf-service/src/progressive_search/storage/hash.rs b/packages/elf-service/src/progressive_search/storage/hash.rs new file mode 100644 index 00000000..8bf42577 --- /dev/null +++ b/packages/elf-service/src/progressive_search/storage/hash.rs @@ -0,0 +1,12 @@ +use std::{ + collections::hash_map::DefaultHasher, + hash::{Hash, Hasher}, +}; + +pub(super) fn hash_query(query: &str) -> String { + let mut hasher = DefaultHasher::new(); + + Hash::hash(query, &mut hasher); + + format!("{:x}", hasher.finish()) +} diff --git a/packages/elf-service/src/progressive_search/storage/hits.rs b/packages/elf-service/src/progressive_search/storage/hits.rs new file mode 100644 index 00000000..91f57acf --- /dev/null +++ b/packages/elf-service/src/progressive_search/storage/hits.rs @@ -0,0 +1,92 @@ +use sqlx::PgExecutor; +use time::OffsetDateTime; +use uuid::Uuid; + +use crate::{ + Error, Result, + progressive_search::{storage::hash, types::HitItem}, +}; +use elf_domain::english_gate; + +pub(crate) async fn record_detail_hits<'e, E>( + executor: E, + query: &str, + items: &[HitItem], + now: OffsetDateTime, +) -> Result<()> +where + E: PgExecutor<'e>, +{ + if !english_gate::is_english_natural_language(query) { + return Err(Error::NonEnglishInput { field: "$.query".to_string() }); + } + + let query_hash = hash::hash_query(query); + let mut hit_ids = Vec::with_capacity(items.len()); + let mut note_ids = Vec::with_capacity(items.len()); + let mut chunk_ids = Vec::with_capacity(items.len()); + let mut ranks = Vec::with_capacity(items.len()); + let mut final_scores = Vec::with_capacity(items.len()); + + for item in items { + let rank = i32::try_from(item.rank).map_err(|_| Error::InvalidRequest { + message: "Search session rank is out of range.".to_string(), + })?; + + hit_ids.push(Uuid::new_v4()); + note_ids.push(item.note_id); + chunk_ids.push(item.chunk_id); + ranks.push(rank); + final_scores.push(item.final_score); + } + + sqlx::query( + "\ +WITH hits AS ( + SELECT * + FROM unnest( + $1::uuid[], + $2::uuid[], + $3::uuid[], + $4::int4[], + $5::real[] +) AS t(hit_id, note_id, chunk_id, rank, final_score) +), +updated AS ( +UPDATE memory_notes +SET + hit_count = hit_count + 1, + last_hit_at = $6 +WHERE note_id = ANY($2) +) +INSERT INTO memory_hits ( + hit_id, + note_id, + chunk_id, + query_hash, + rank, + final_score, + ts +) +SELECT + hit_id, + note_id, + chunk_id, + $7, + rank, + final_score, + $6 +FROM hits", + ) + .bind(&hit_ids) + .bind(¬e_ids) + .bind(&chunk_ids) + .bind(&ranks) + .bind(&final_scores) + .bind(now) + .bind(query_hash.as_str()) + .execute(executor) + .await?; + + Ok(()) +} diff --git a/packages/elf-service/src/progressive_search/storage/session.rs b/packages/elf-service/src/progressive_search/storage/session.rs new file mode 100644 index 00000000..4e26f524 --- /dev/null +++ b/packages/elf-service/src/progressive_search/storage/session.rs @@ -0,0 +1,177 @@ +use std::str::FromStr; + +use serde_json; +use sqlx::PgExecutor; +use time::{Duration, OffsetDateTime}; +use uuid::Uuid; + +use crate::{ + Error, Result, + progressive_search::types::{ + NewSearchSession, SESSION_ABSOLUTE_TTL_HOURS, SESSION_SLIDING_TTL_HOURS, SearchSession, + SearchSessionItemRecord, SearchSessionMode, SearchSessionRow, + }, +}; + +pub(crate) async fn store_search_session<'e, E>( + executor: E, + session: NewSearchSession<'_>, +) -> Result<()> +where + E: PgExecutor<'e>, +{ + let items_json = serde_json::to_value(session.items).map_err(|err| Error::Storage { + message: format!("Failed to encode search session items: {err}"), + })?; + let query_plan_json = + session.query_plan.map(serde_json::to_value).transpose().map_err(|err| Error::Storage { + message: format!("Failed to encode search session query plan: {err}"), + })?; + let trajectory_summary_json = + session.trajectory_summary.map(serde_json::to_value).transpose().map_err(|err| { + Error::Storage { + message: format!("Failed to encode search session trajectory summary: {err}"), + } + })?; + + sqlx::query( + "\ +INSERT INTO search_sessions ( + search_session_id, + trace_id, + tenant_id, + project_id, + agent_id, + read_profile, + query, + mode, + trajectory_summary, + query_plan, + items, + created_at, + expires_at +) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)", + ) + .bind(session.search_session_id) + .bind(session.trace_id) + .bind(session.tenant_id.trim()) + .bind(session.project_id.trim()) + .bind(session.agent_id.trim()) + .bind(session.read_profile) + .bind(session.query) + .bind(session.mode.as_str()) + .bind(trajectory_summary_json) + .bind(query_plan_json) + .bind(items_json) + .bind(session.created_at) + .bind(session.expires_at) + .execute(executor) + .await?; + + Ok(()) +} + +pub(crate) async fn load_search_session<'e, E>( + executor: E, + search_session_id: Uuid, + now: OffsetDateTime, +) -> Result +where + E: PgExecutor<'e>, +{ + let row = sqlx::query_as::<_, SearchSessionRow>( + "\ +SELECT + search_session_id, + trace_id, + tenant_id, + project_id, + agent_id, + read_profile, + query, + mode, + trajectory_summary, + query_plan, + items, + created_at, + expires_at +FROM search_sessions +WHERE search_session_id = $1", + ) + .bind(search_session_id) + .fetch_optional(executor) + .await?; + let Some(row) = row else { + return Err(Error::InvalidRequest { message: "Unknown search_session_id.".to_string() }); + }; + let expires_at: OffsetDateTime = row.expires_at; + + if expires_at <= now { + return Err(Error::InvalidRequest { message: "Search session expired.".to_string() }); + } + + let items: Vec = serde_json::from_value(row.items).map_err(|err| { + Error::Storage { message: format!("Failed to decode search session items: {err}") } + })?; + let mode = SearchSessionMode::from_str(row.mode.as_str())?; + let query_plan = match row.query_plan { + Some(value) => Some(serde_json::from_value(value).map_err(|err| Error::Storage { + message: format!("Failed to decode search session query_plan: {err}"), + })?), + None => None, + }; + let trajectory_summary = match row.trajectory_summary { + Some(value) => Some(serde_json::from_value(value).map_err(|err| Error::Storage { + message: format!("Failed to decode search session trajectory summary: {err}"), + })?), + None => None, + }; + + Ok(SearchSession { + search_session_id: row.search_session_id, + trace_id: row.trace_id, + tenant_id: row.tenant_id, + project_id: row.project_id, + agent_id: row.agent_id, + read_profile: row.read_profile, + query: row.query, + items, + mode, + trajectory_summary, + query_plan, + created_at: row.created_at, + expires_at, + }) +} + +pub(crate) async fn touch_search_session<'e, E>( + executor: E, + session: &SearchSession, + now: OffsetDateTime, +) -> Result +where + E: PgExecutor<'e>, +{ + let absolute_expires_at = session.created_at + Duration::hours(SESSION_ABSOLUTE_TTL_HOURS); + let sliding_expires_at = now + Duration::hours(SESSION_SLIDING_TTL_HOURS); + let touched = if sliding_expires_at < absolute_expires_at { + sliding_expires_at + } else { + absolute_expires_at + }; + + if touched <= session.expires_at { + return Ok(session.expires_at); + } + + sqlx::query( + "UPDATE search_sessions SET expires_at = $1 WHERE search_session_id = $2 AND expires_at < $1", + ) + .bind(touched) + .bind(session.search_session_id) + .execute(executor) + .await?; + + Ok(touched) +} From d422bf2310ca42057512be96cc5dfa07dce549eb Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 30 Jun 2026 15:07:50 +0800 Subject: [PATCH 2/7] {"schema":"decodex/commit/1","summary":"Split progressive search details modules","authority":"manual"} --- .../src/progressive_search/details.rs | 309 +----------------- .../src/progressive_search/details/access.rs | 77 +++++ .../src/progressive_search/details/results.rs | 128 ++++++++ .../src/progressive_search/details/text.rs | 79 +++++ .../progressive_search/details/timeline.rs | 38 +++ 5 files changed, 332 insertions(+), 299 deletions(-) create mode 100644 packages/elf-service/src/progressive_search/details/access.rs create mode 100644 packages/elf-service/src/progressive_search/details/results.rs create mode 100644 packages/elf-service/src/progressive_search/details/text.rs create mode 100644 packages/elf-service/src/progressive_search/details/timeline.rs diff --git a/packages/elf-service/src/progressive_search/details.rs b/packages/elf-service/src/progressive_search/details.rs index e9479dbd..09a3ded3 100644 --- a/packages/elf-service/src/progressive_search/details.rs +++ b/packages/elf-service/src/progressive_search/details.rs @@ -1,300 +1,11 @@ -use std::{ - cmp::Ordering, - collections::{BTreeMap, HashMap, hash_set::HashSet}, +mod access; +mod results; +mod text; +mod timeline; + +pub(super) use self::{ + access::{resolve_read_scopes, validate_search_session_access}, + results::{SearchDetailsBuildArgs, build_search_details_results}, + text::build_summary, + timeline::build_timeline_by_day, }; - -use serde_json; -use time::OffsetDateTime; -use uuid::Uuid; - -use crate::{ - Error, NoteFetchResponse, PayloadLevel, Result, - access::{self, SharedSpaceGrantKey}, - progressive_search::types::{ - HitItem, SearchDetailsError, SearchDetailsResult, SearchIndexItem, SearchSession, - SearchSessionItemRecord, SearchTimelineGroup, SearchTimelineResponse, - }, - structured_fields::StructuredFields, -}; -use elf_config::Config; -use elf_storage::models::MemoryNote; - -pub(super) struct SearchDetailsBuildArgs<'a> { - pub(super) session_items_by_note_id: &'a HashMap, - pub(super) notes_by_id: &'a HashMap, - pub(super) structured_by_note: &'a HashMap, - pub(super) session: &'a SearchSession, - pub(super) shared_grants: &'a HashSet, - pub(super) allowed_scopes: &'a [String], - pub(super) now: OffsetDateTime, - pub(super) record_hits_enabled: bool, - pub(super) payload_level: PayloadLevel, - pub(super) max_note_chars: usize, -} - -pub(super) fn build_search_details_results( - requested_note_ids: Vec, - args: SearchDetailsBuildArgs<'_>, -) -> (Vec, Vec) { - let mut results = Vec::with_capacity(requested_note_ids.len()); - let mut hits = Vec::new(); - let mut hit_seen = HashSet::new(); - - for note_id in requested_note_ids { - let Some(session_item) = args.session_items_by_note_id.get(¬e_id) else { - results.push(SearchDetailsResult { - note_id, - note: None, - error: Some(SearchDetailsError { - code: "NOT_IN_SESSION".to_string(), - message: "Requested note_id is not present in the search session.".to_string(), - }), - }); - - continue; - }; - let Some(note) = args.notes_by_id.get(¬e_id) else { - results.push(SearchDetailsResult { - note_id, - note: None, - error: Some(SearchDetailsError { - code: "NOTE_NOT_FOUND".to_string(), - message: "Note not found.".to_string(), - }), - }); - - continue; - }; - let error = validate_note_access( - note, - args.session, - args.allowed_scopes, - args.shared_grants, - args.now, - ); - - if let Some(error) = error { - results.push(SearchDetailsResult { note_id, note: None, error: Some(error) }); - - continue; - } - - let structured = if args.payload_level == PayloadLevel::L0 { - None - } else { - args.structured_by_note.get(¬e.note_id).cloned() - }; - let note_text = apply_payload_level_to_search_details_text( - note.text.as_str(), - structured.as_ref(), - args.payload_level, - args.max_note_chars, - ); - let source_ref = if args.payload_level == PayloadLevel::L2 { - note.source_ref.clone() - } else { - serde_json::json!({}) - }; - let note_response = NoteFetchResponse { - note_id: note.note_id, - tenant_id: note.tenant_id.clone(), - project_id: note.project_id.clone(), - agent_id: note.agent_id.clone(), - scope: note.scope.clone(), - r#type: note.r#type.clone(), - key: note.key.clone(), - text: note_text, - importance: note.importance, - confidence: note.confidence, - status: note.status.clone(), - updated_at: note.updated_at, - expires_at: note.expires_at, - source_ref, - structured, - }; - - results.push(SearchDetailsResult { note_id, note: Some(note_response), error: None }); - - if args.record_hits_enabled && hit_seen.insert(note_id) { - hits.push(HitItem { - note_id, - chunk_id: session_item.chunk_id, - rank: session_item.rank, - final_score: session_item.final_score, - }); - } - } - - (results, hits) -} - -pub(super) fn build_timeline_by_day( - search_session_id: Uuid, - expires_at: OffsetDateTime, - items: &[SearchSessionItemRecord], -) -> Result { - let mut grouped: BTreeMap> = BTreeMap::new(); - - for item in items { - let date = item.updated_at.date().to_string(); - - grouped.entry(date).or_default().push(item.to_index_item()); - } - - let mut groups = Vec::with_capacity(grouped.len()); - - for (date, mut items) in grouped.into_iter().rev() { - items.sort_by(|a, b| { - b.updated_at - .cmp(&a.updated_at) - .then_with(|| b.final_score.partial_cmp(&a.final_score).unwrap_or(Ordering::Equal)) - }); - groups.push(SearchTimelineGroup { date, items }); - } - - Ok(SearchTimelineResponse { search_session_id, expires_at, groups }) -} - -pub(super) fn build_summary(raw: &str, max_chars: usize) -> String { - let normalized = normalize_whitespace(raw); - - truncate_chars(&normalized, max_chars) -} - -pub(super) fn resolve_read_scopes(cfg: &Config, profile: &str) -> Result> { - match profile { - "private_only" => Ok(cfg.scopes.read_profiles.private_only.clone()), - "private_plus_project" => Ok(cfg.scopes.read_profiles.private_plus_project.clone()), - "all_scopes" => Ok(cfg.scopes.read_profiles.all_scopes.clone()), - _ => Err(Error::InvalidRequest { message: "Unknown read_profile.".to_string() }), - } -} - -pub(super) fn validate_search_session_access( - session: &SearchSession, - tenant_id: &str, - project_id: &str, - agent_id: &str, -) -> Result<()> { - if session.tenant_id != tenant_id - || session.project_id != project_id - || session.agent_id != agent_id - { - return Err(Error::InvalidRequest { message: "Unknown search_session_id.".to_string() }); - } - - Ok(()) -} - -fn apply_payload_level_to_search_details_text( - raw_text: &str, - structured: Option<&StructuredFields>, - payload_level: PayloadLevel, - max_note_chars: usize, -) -> String { - match payload_level { - PayloadLevel::L0 => build_summary(raw_text, max_note_chars), - PayloadLevel::L1 => { - let candidate_text = structured - .and_then(|item| item.summary.as_deref()) - .filter(|summary| !summary.trim().is_empty()) - .unwrap_or(raw_text); - - build_summary(candidate_text, max_note_chars) - }, - PayloadLevel::L2 => raw_text.to_string(), - } -} - -fn normalize_whitespace(raw: &str) -> String { - let mut out = String::with_capacity(raw.len()); - let mut prev_space = false; - - for ch in raw.chars() { - if ch.is_whitespace() { - if !prev_space { - out.push(' '); - - prev_space = true; - } - - continue; - } - - out.push(ch); - - prev_space = false; - } - - out.trim().to_string() -} - -fn truncate_chars(raw: &str, max_chars: usize) -> String { - if raw.chars().count() <= max_chars { - return raw.to_string(); - } - - const TRUNCATION_MARKER: &str = "..."; - - let marker_chars = TRUNCATION_MARKER.chars().count(); - - if max_chars <= marker_chars { - return TRUNCATION_MARKER.chars().take(max_chars).collect(); - } - - let truncated_chars = max_chars - marker_chars; - let mut out = String::with_capacity(max_chars); - - for (idx, ch) in raw.chars().enumerate() { - if idx >= truncated_chars { - break; - } - - out.push(ch); - } - - out.push_str(TRUNCATION_MARKER); - - out -} - -fn validate_note_access( - note: &MemoryNote, - session: &SearchSession, - allowed_scopes: &[String], - shared_grants: &HashSet, - now: OffsetDateTime, -) -> Option { - if note.status != "active" { - return Some(SearchDetailsError { - code: "NOTE_INACTIVE".to_string(), - message: "Note is not active.".to_string(), - }); - } - if note.expires_at.map(|ts| ts <= now).unwrap_or(false) { - return Some(SearchDetailsError { - code: "NOTE_EXPIRED".to_string(), - message: "Note is expired.".to_string(), - }); - } - if !allowed_scopes.iter().any(|scope| scope == ¬e.scope) { - return Some(SearchDetailsError { - code: "SCOPE_DENIED".to_string(), - message: "Note scope is not allowed for this read_profile.".to_string(), - }); - } - if !access::note_read_allowed( - note, - session.agent_id.as_str(), - allowed_scopes, - shared_grants, - now, - ) { - return Some(SearchDetailsError { - code: "SCOPE_DENIED".to_string(), - message: "Note scope is not allowed for this read_profile.".to_string(), - }); - } - - None -} diff --git a/packages/elf-service/src/progressive_search/details/access.rs b/packages/elf-service/src/progressive_search/details/access.rs new file mode 100644 index 00000000..bdc77e6d --- /dev/null +++ b/packages/elf-service/src/progressive_search/details/access.rs @@ -0,0 +1,77 @@ +use std::collections::hash_set::HashSet; + +use time::OffsetDateTime; + +use crate::{ + Error, Result, + access::{self, SharedSpaceGrantKey}, + progressive_search::types::{SearchDetailsError, SearchSession}, +}; +use elf_config::Config; +use elf_storage::models::MemoryNote; + +pub(crate) fn resolve_read_scopes(cfg: &Config, profile: &str) -> Result> { + match profile { + "private_only" => Ok(cfg.scopes.read_profiles.private_only.clone()), + "private_plus_project" => Ok(cfg.scopes.read_profiles.private_plus_project.clone()), + "all_scopes" => Ok(cfg.scopes.read_profiles.all_scopes.clone()), + _ => Err(Error::InvalidRequest { message: "Unknown read_profile.".to_string() }), + } +} + +pub(crate) fn validate_search_session_access( + session: &SearchSession, + tenant_id: &str, + project_id: &str, + agent_id: &str, +) -> Result<()> { + if session.tenant_id != tenant_id + || session.project_id != project_id + || session.agent_id != agent_id + { + return Err(Error::InvalidRequest { message: "Unknown search_session_id.".to_string() }); + } + + Ok(()) +} + +pub(super) fn validate_note_access( + note: &MemoryNote, + session: &SearchSession, + allowed_scopes: &[String], + shared_grants: &HashSet, + now: OffsetDateTime, +) -> Option { + if note.status != "active" { + return Some(SearchDetailsError { + code: "NOTE_INACTIVE".to_string(), + message: "Note is not active.".to_string(), + }); + } + if note.expires_at.map(|ts| ts <= now).unwrap_or(false) { + return Some(SearchDetailsError { + code: "NOTE_EXPIRED".to_string(), + message: "Note is expired.".to_string(), + }); + } + if !allowed_scopes.iter().any(|scope| scope == ¬e.scope) { + return Some(SearchDetailsError { + code: "SCOPE_DENIED".to_string(), + message: "Note scope is not allowed for this read_profile.".to_string(), + }); + } + if !access::note_read_allowed( + note, + session.agent_id.as_str(), + allowed_scopes, + shared_grants, + now, + ) { + return Some(SearchDetailsError { + code: "SCOPE_DENIED".to_string(), + message: "Note scope is not allowed for this read_profile.".to_string(), + }); + } + + None +} diff --git a/packages/elf-service/src/progressive_search/details/results.rs b/packages/elf-service/src/progressive_search/details/results.rs new file mode 100644 index 00000000..192bc15b --- /dev/null +++ b/packages/elf-service/src/progressive_search/details/results.rs @@ -0,0 +1,128 @@ +use std::collections::{HashMap, hash_set::HashSet}; + +use serde_json; +use time::OffsetDateTime; +use uuid::Uuid; + +use crate::{ + NoteFetchResponse, PayloadLevel, + access::SharedSpaceGrantKey, + progressive_search::{ + details::{access, text}, + types::{ + HitItem, SearchDetailsError, SearchDetailsResult, SearchSession, + SearchSessionItemRecord, + }, + }, + structured_fields::StructuredFields, +}; +use elf_storage::models::MemoryNote; + +pub(crate) struct SearchDetailsBuildArgs<'a> { + pub(crate) session_items_by_note_id: &'a HashMap, + pub(crate) notes_by_id: &'a HashMap, + pub(crate) structured_by_note: &'a HashMap, + pub(crate) session: &'a SearchSession, + pub(crate) shared_grants: &'a HashSet, + pub(crate) allowed_scopes: &'a [String], + pub(crate) now: OffsetDateTime, + pub(crate) record_hits_enabled: bool, + pub(crate) payload_level: PayloadLevel, + pub(crate) max_note_chars: usize, +} + +pub(crate) fn build_search_details_results( + requested_note_ids: Vec, + args: SearchDetailsBuildArgs<'_>, +) -> (Vec, Vec) { + let mut results = Vec::with_capacity(requested_note_ids.len()); + let mut hits = Vec::new(); + let mut hit_seen = HashSet::new(); + + for note_id in requested_note_ids { + let Some(session_item) = args.session_items_by_note_id.get(¬e_id) else { + results.push(SearchDetailsResult { + note_id, + note: None, + error: Some(SearchDetailsError { + code: "NOT_IN_SESSION".to_string(), + message: "Requested note_id is not present in the search session.".to_string(), + }), + }); + + continue; + }; + let Some(note) = args.notes_by_id.get(¬e_id) else { + results.push(SearchDetailsResult { + note_id, + note: None, + error: Some(SearchDetailsError { + code: "NOTE_NOT_FOUND".to_string(), + message: "Note not found.".to_string(), + }), + }); + + continue; + }; + let error = access::validate_note_access( + note, + args.session, + args.allowed_scopes, + args.shared_grants, + args.now, + ); + + if let Some(error) = error { + results.push(SearchDetailsResult { note_id, note: None, error: Some(error) }); + + continue; + } + + let structured = if args.payload_level == PayloadLevel::L0 { + None + } else { + args.structured_by_note.get(¬e.note_id).cloned() + }; + let note_text = text::apply_payload_level_to_search_details_text( + note.text.as_str(), + structured.as_ref(), + args.payload_level, + args.max_note_chars, + ); + let source_ref = if args.payload_level == PayloadLevel::L2 { + note.source_ref.clone() + } else { + serde_json::json!({}) + }; + let note_response = NoteFetchResponse { + note_id: note.note_id, + tenant_id: note.tenant_id.clone(), + project_id: note.project_id.clone(), + agent_id: note.agent_id.clone(), + scope: note.scope.clone(), + r#type: note.r#type.clone(), + key: note.key.clone(), + text: note_text, + importance: note.importance, + confidence: note.confidence, + status: note.status.clone(), + updated_at: note.updated_at, + expires_at: note.expires_at, + source_ref, + structured, + }; + + results.push(SearchDetailsResult { note_id, note: Some(note_response), error: None }); + + if args.record_hits_enabled && hit_seen.insert(note_id) { + hits.push(HitItem { + note_id, + chunk_id: session_item.chunk_id, + rank: session_item.rank, + final_score: session_item.final_score, + }); + } + } + + (results, hits) +} diff --git a/packages/elf-service/src/progressive_search/details/text.rs b/packages/elf-service/src/progressive_search/details/text.rs new file mode 100644 index 00000000..1bed9c3b --- /dev/null +++ b/packages/elf-service/src/progressive_search/details/text.rs @@ -0,0 +1,79 @@ +use crate::{PayloadLevel, structured_fields::StructuredFields}; + +pub(crate) fn build_summary(raw: &str, max_chars: usize) -> String { + let normalized = normalize_whitespace(raw); + + truncate_chars(&normalized, max_chars) +} + +pub(super) fn apply_payload_level_to_search_details_text( + raw_text: &str, + structured: Option<&StructuredFields>, + payload_level: PayloadLevel, + max_note_chars: usize, +) -> String { + match payload_level { + PayloadLevel::L0 => build_summary(raw_text, max_note_chars), + PayloadLevel::L1 => { + let candidate_text = structured + .and_then(|item| item.summary.as_deref()) + .filter(|summary| !summary.trim().is_empty()) + .unwrap_or(raw_text); + + build_summary(candidate_text, max_note_chars) + }, + PayloadLevel::L2 => raw_text.to_string(), + } +} + +fn normalize_whitespace(raw: &str) -> String { + let mut out = String::with_capacity(raw.len()); + let mut prev_space = false; + + for ch in raw.chars() { + if ch.is_whitespace() { + if !prev_space { + out.push(' '); + + prev_space = true; + } + + continue; + } + + out.push(ch); + + prev_space = false; + } + + out.trim().to_string() +} + +fn truncate_chars(raw: &str, max_chars: usize) -> String { + if raw.chars().count() <= max_chars { + return raw.to_string(); + } + + const TRUNCATION_MARKER: &str = "..."; + + let marker_chars = TRUNCATION_MARKER.chars().count(); + + if max_chars <= marker_chars { + return TRUNCATION_MARKER.chars().take(max_chars).collect(); + } + + let truncated_chars = max_chars - marker_chars; + let mut out = String::with_capacity(max_chars); + + for (idx, ch) in raw.chars().enumerate() { + if idx >= truncated_chars { + break; + } + + out.push(ch); + } + + out.push_str(TRUNCATION_MARKER); + + out +} diff --git a/packages/elf-service/src/progressive_search/details/timeline.rs b/packages/elf-service/src/progressive_search/details/timeline.rs new file mode 100644 index 00000000..c4f4b07e --- /dev/null +++ b/packages/elf-service/src/progressive_search/details/timeline.rs @@ -0,0 +1,38 @@ +use std::{cmp::Ordering, collections::BTreeMap}; + +use time::OffsetDateTime; +use uuid::Uuid; + +use crate::{ + Result, + progressive_search::types::{ + SearchIndexItem, SearchSessionItemRecord, SearchTimelineGroup, SearchTimelineResponse, + }, +}; + +pub(crate) fn build_timeline_by_day( + search_session_id: Uuid, + expires_at: OffsetDateTime, + items: &[SearchSessionItemRecord], +) -> Result { + let mut grouped: BTreeMap> = BTreeMap::new(); + + for item in items { + let date = item.updated_at.date().to_string(); + + grouped.entry(date).or_default().push(item.to_index_item()); + } + + let mut groups = Vec::with_capacity(grouped.len()); + + for (date, mut items) in grouped.into_iter().rev() { + items.sort_by(|a, b| { + b.updated_at + .cmp(&a.updated_at) + .then_with(|| b.final_score.partial_cmp(&a.final_score).unwrap_or(Ordering::Equal)) + }); + groups.push(SearchTimelineGroup { date, items }); + } + + Ok(SearchTimelineResponse { search_session_id, expires_at, groups }) +} From 0e265461af9fd27ac4edbd8117ebf61b34d2ef76 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 30 Jun 2026 15:10:16 +0800 Subject: [PATCH 3/7] {"schema":"decodex/commit/1","summary":"Split progressive search followup endpoints","authority":"manual"} --- .../src/progressive_search/followup.rs | 231 +----------------- .../followup/details_endpoint.rs | 126 ++++++++++ .../followup/session_get.rs | 57 +++++ .../progressive_search/followup/timeline.rs | 65 +++++ 4 files changed, 251 insertions(+), 228 deletions(-) create mode 100644 packages/elf-service/src/progressive_search/followup/details_endpoint.rs create mode 100644 packages/elf-service/src/progressive_search/followup/session_get.rs create mode 100644 packages/elf-service/src/progressive_search/followup/timeline.rs diff --git a/packages/elf-service/src/progressive_search/followup.rs b/packages/elf-service/src/progressive_search/followup.rs index cd516793..d9e14459 100644 --- a/packages/elf-service/src/progressive_search/followup.rs +++ b/packages/elf-service/src/progressive_search/followup.rs @@ -1,228 +1,3 @@ -use std::collections::{HashMap, hash_set::HashSet}; - -use sqlx; -use time::OffsetDateTime; -use uuid::Uuid; - -use crate::{ - ElfService, Error, PayloadLevel, Result, - access::{self, ORG_PROJECT_ID}, - progressive_search::{ - details::{self, SearchDetailsBuildArgs}, - storage::{self}, - types::{ - SearchDetailsRequest, SearchDetailsResponse, SearchIndexItem, SearchSessionGetRequest, - SearchSessionGetResponse, SearchSessionItemRecord, SearchTimelineGroup, - SearchTimelineRequest, SearchTimelineResponse, - }, - }, - structured_fields, -}; -use elf_storage::models::MemoryNote; - -impl ElfService { - /// Reloads a stored search session and optionally extends its TTL. - pub async fn search_session_get( - &self, - req: SearchSessionGetRequest, - ) -> Result { - let tenant_id = req.tenant_id.trim(); - let project_id = req.project_id.trim(); - let agent_id = req.agent_id.trim(); - - if tenant_id.is_empty() || project_id.is_empty() || agent_id.is_empty() { - return Err(Error::InvalidRequest { - message: "tenant_id, project_id, and agent_id are required.".to_string(), - }); - } - - let now = OffsetDateTime::now_utc(); - let session = - storage::load_search_session(&self.db.pool, req.search_session_id, now).await?; - - details::validate_search_session_access(&session, tenant_id, project_id, agent_id)?; - - let touch = req.touch.unwrap_or(true); - let expires_at = if touch { - storage::touch_search_session(&self.db.pool, &session, now).await? - } else { - session.expires_at - }; - let top_k = req.top_k.unwrap_or(self.cfg.memory.top_k).max(1); - let items: Vec = session - .items - .into_iter() - .take(top_k as usize) - .map(|item| item.to_index_item()) - .collect(); - - Ok(SearchSessionGetResponse { - trace_id: session.trace_id, - search_session_id: session.search_session_id, - expires_at, - items, - mode: session.mode, - query_plan: session.query_plan, - trajectory_summary: session.trajectory_summary, - }) - } - - /// Reprojects a stored search session into timeline groups. - pub async fn search_timeline( - &self, - req: SearchTimelineRequest, - ) -> Result { - let tenant_id = req.tenant_id.trim(); - let project_id = req.project_id.trim(); - let agent_id = req.agent_id.trim(); - - if tenant_id.is_empty() || project_id.is_empty() || agent_id.is_empty() { - return Err(Error::InvalidRequest { - message: "tenant_id, project_id, and agent_id are required.".to_string(), - }); - } - - let now = OffsetDateTime::now_utc(); - let session = - storage::load_search_session(&self.db.pool, req.search_session_id, now).await?; - - details::validate_search_session_access(&session, tenant_id, project_id, agent_id)?; - - let expires_at = storage::touch_search_session(&self.db.pool, &session, now).await?; - let payload_level = req.payload_level; - let group_by = req.group_by.unwrap_or_else(|| { - if payload_level == PayloadLevel::L0 { "none".to_string() } else { "day".to_string() } - }); - - match group_by.as_str() { - "day" => details::build_timeline_by_day( - session.search_session_id, - expires_at, - &session.items, - ), - "none" => Ok(SearchTimelineResponse { - search_session_id: session.search_session_id, - expires_at, - groups: vec![SearchTimelineGroup { - date: "all".to_string(), - items: session - .items - .iter() - .map(SearchSessionItemRecord::to_index_item) - .collect(), - }], - }), - _ => Err(Error::InvalidRequest { - message: "group_by must be one of: day, none.".to_string(), - }), - } - } - - /// Materializes selected note details out of a stored search session. - pub async fn search_details(&self, req: SearchDetailsRequest) -> Result { - let tenant_id = req.tenant_id.trim(); - let project_id = req.project_id.trim(); - let agent_id = req.agent_id.trim(); - - if tenant_id.is_empty() || project_id.is_empty() || agent_id.is_empty() { - return Err(Error::InvalidRequest { - message: "tenant_id, project_id, and agent_id are required.".to_string(), - }); - } - - let now = OffsetDateTime::now_utc(); - let session = - storage::load_search_session(&self.db.pool, req.search_session_id, now).await?; - - details::validate_search_session_access(&session, tenant_id, project_id, agent_id)?; - - let expires_at = storage::touch_search_session(&self.db.pool, &session, now).await?; - let mut by_note_id: HashMap = HashMap::new(); - - for item in &session.items { - by_note_id.insert(item.note_id, item.clone()); - } - - let mut requested_in_session = Vec::new(); - let mut seen = HashSet::new(); - - for note_id in &req.note_ids { - if by_note_id.contains_key(note_id) && seen.insert(*note_id) { - requested_in_session.push(*note_id); - } - } - - let mut notes_by_id = HashMap::new(); - - if !requested_in_session.is_empty() { - let rows: Vec = sqlx::query_as::<_, MemoryNote>( - "\ -SELECT * -FROM memory_notes -WHERE note_id = ANY($1::uuid[]) - AND tenant_id = $2 - AND ( - project_id = $3 - OR (project_id = $4 AND scope = 'org_shared') - )", - ) - .bind(requested_in_session.as_slice()) - .bind(session.tenant_id.as_str()) - .bind(session.project_id.as_str()) - .bind(ORG_PROJECT_ID) - .fetch_all(&self.db.pool) - .await?; - - for note in rows { - notes_by_id.insert(note.note_id, note); - } - } - - let structured_by_note = if req.payload_level == PayloadLevel::L0 { - HashMap::new() - } else { - structured_fields::fetch_structured_fields( - &self.db.pool, - requested_in_session.as_slice(), - ) - .await? - }; - let allowed_scopes = details::resolve_read_scopes(&self.cfg, &session.read_profile)?; - let shared_grants = access::load_shared_read_grants_with_org_shared( - &self.db.pool, - session.tenant_id.as_str(), - session.project_id.as_str(), - agent_id, - allowed_scopes.iter().any(|scope| scope == "org_shared"), - ) - .await?; - let record_hits = req.record_hits.unwrap_or(true); - let details_args = SearchDetailsBuildArgs { - session_items_by_note_id: &by_note_id, - notes_by_id: ¬es_by_id, - structured_by_note: &structured_by_note, - session: &session, - shared_grants: &shared_grants, - allowed_scopes: &allowed_scopes, - now, - record_hits_enabled: record_hits, - payload_level: req.payload_level, - max_note_chars: self.cfg.memory.max_note_chars as usize, - }; - let (results, hits) = details::build_search_details_results(req.note_ids, details_args); - - if !hits.is_empty() { - let mut tx = self.db.pool.begin().await?; - - storage::record_detail_hits(&mut *tx, &session.query, &hits, now).await?; - - tx.commit().await?; - } - - Ok(SearchDetailsResponse { - search_session_id: session.search_session_id, - expires_at, - results, - }) - } -} +mod details_endpoint; +mod session_get; +mod timeline; diff --git a/packages/elf-service/src/progressive_search/followup/details_endpoint.rs b/packages/elf-service/src/progressive_search/followup/details_endpoint.rs new file mode 100644 index 00000000..6502248f --- /dev/null +++ b/packages/elf-service/src/progressive_search/followup/details_endpoint.rs @@ -0,0 +1,126 @@ +use std::collections::{HashMap, hash_set::HashSet}; + +use time::OffsetDateTime; +use uuid::Uuid; + +use crate::{ + ElfService, Error, PayloadLevel, Result, + access::{self, ORG_PROJECT_ID}, + progressive_search::{ + details::{self, SearchDetailsBuildArgs}, + storage, + types::{SearchDetailsRequest, SearchDetailsResponse, SearchSessionItemRecord}, + }, + structured_fields, +}; +use elf_storage::models::MemoryNote; + +impl ElfService { + /// Materializes selected note details out of a stored search session. + pub async fn search_details(&self, req: SearchDetailsRequest) -> Result { + let tenant_id = req.tenant_id.trim(); + let project_id = req.project_id.trim(); + let agent_id = req.agent_id.trim(); + + if tenant_id.is_empty() || project_id.is_empty() || agent_id.is_empty() { + return Err(Error::InvalidRequest { + message: "tenant_id, project_id, and agent_id are required.".to_string(), + }); + } + + let now = OffsetDateTime::now_utc(); + let session = + storage::load_search_session(&self.db.pool, req.search_session_id, now).await?; + + details::validate_search_session_access(&session, tenant_id, project_id, agent_id)?; + + let expires_at = storage::touch_search_session(&self.db.pool, &session, now).await?; + let mut by_note_id: HashMap = HashMap::new(); + + for item in &session.items { + by_note_id.insert(item.note_id, item.clone()); + } + + let mut requested_in_session = Vec::new(); + let mut seen = HashSet::new(); + + for note_id in &req.note_ids { + if by_note_id.contains_key(note_id) && seen.insert(*note_id) { + requested_in_session.push(*note_id); + } + } + + let mut notes_by_id = HashMap::new(); + + if !requested_in_session.is_empty() { + let rows: Vec = sqlx::query_as::<_, MemoryNote>( + "\ +SELECT * +FROM memory_notes +WHERE note_id = ANY($1::uuid[]) + AND tenant_id = $2 + AND ( + project_id = $3 + OR (project_id = $4 AND scope = 'org_shared') + )", + ) + .bind(requested_in_session.as_slice()) + .bind(session.tenant_id.as_str()) + .bind(session.project_id.as_str()) + .bind(ORG_PROJECT_ID) + .fetch_all(&self.db.pool) + .await?; + + for note in rows { + notes_by_id.insert(note.note_id, note); + } + } + + let structured_by_note = if req.payload_level == PayloadLevel::L0 { + HashMap::new() + } else { + structured_fields::fetch_structured_fields( + &self.db.pool, + requested_in_session.as_slice(), + ) + .await? + }; + let allowed_scopes = details::resolve_read_scopes(&self.cfg, &session.read_profile)?; + let shared_grants = access::load_shared_read_grants_with_org_shared( + &self.db.pool, + session.tenant_id.as_str(), + session.project_id.as_str(), + agent_id, + allowed_scopes.iter().any(|scope| scope == "org_shared"), + ) + .await?; + let record_hits = req.record_hits.unwrap_or(true); + let details_args = SearchDetailsBuildArgs { + session_items_by_note_id: &by_note_id, + notes_by_id: ¬es_by_id, + structured_by_note: &structured_by_note, + session: &session, + shared_grants: &shared_grants, + allowed_scopes: &allowed_scopes, + now, + record_hits_enabled: record_hits, + payload_level: req.payload_level, + max_note_chars: self.cfg.memory.max_note_chars as usize, + }; + let (results, hits) = details::build_search_details_results(req.note_ids, details_args); + + if !hits.is_empty() { + let mut tx = self.db.pool.begin().await?; + + storage::record_detail_hits(&mut *tx, &session.query, &hits, now).await?; + + tx.commit().await?; + } + + Ok(SearchDetailsResponse { + search_session_id: session.search_session_id, + expires_at, + results, + }) + } +} diff --git a/packages/elf-service/src/progressive_search/followup/session_get.rs b/packages/elf-service/src/progressive_search/followup/session_get.rs new file mode 100644 index 00000000..1b502d60 --- /dev/null +++ b/packages/elf-service/src/progressive_search/followup/session_get.rs @@ -0,0 +1,57 @@ +use time::OffsetDateTime; + +use crate::{ + ElfService, Error, Result, + progressive_search::{ + details, storage, + types::{SearchIndexItem, SearchSessionGetRequest, SearchSessionGetResponse}, + }, +}; + +impl ElfService { + /// Reloads a stored search session and optionally extends its TTL. + pub async fn search_session_get( + &self, + req: SearchSessionGetRequest, + ) -> Result { + let tenant_id = req.tenant_id.trim(); + let project_id = req.project_id.trim(); + let agent_id = req.agent_id.trim(); + + if tenant_id.is_empty() || project_id.is_empty() || agent_id.is_empty() { + return Err(Error::InvalidRequest { + message: "tenant_id, project_id, and agent_id are required.".to_string(), + }); + } + + let now = OffsetDateTime::now_utc(); + let session = + storage::load_search_session(&self.db.pool, req.search_session_id, now).await?; + + details::validate_search_session_access(&session, tenant_id, project_id, agent_id)?; + + let touch = req.touch.unwrap_or(true); + let expires_at = if touch { + storage::touch_search_session(&self.db.pool, &session, now).await? + } else { + session.expires_at + }; + let top_k = req.top_k.unwrap_or(self.cfg.memory.top_k).max(1); + let items: Vec = session + .items + .into_iter() + .take(top_k as usize) + .map(|item| item.to_index_item()) + .collect(); + + Ok(SearchSessionGetResponse { + trace_id: session.trace_id, + search_session_id: session.search_session_id, + expires_at, + items, + mode: session.mode, + query_plan: session.query_plan, + trajectory_summary: session.trajectory_summary, + }) + } +} diff --git a/packages/elf-service/src/progressive_search/followup/timeline.rs b/packages/elf-service/src/progressive_search/followup/timeline.rs new file mode 100644 index 00000000..b27d32b5 --- /dev/null +++ b/packages/elf-service/src/progressive_search/followup/timeline.rs @@ -0,0 +1,65 @@ +use time::OffsetDateTime; + +use crate::{ + ElfService, Error, PayloadLevel, Result, + progressive_search::{ + details, storage, + types::{ + SearchSessionItemRecord, SearchTimelineGroup, SearchTimelineRequest, + SearchTimelineResponse, + }, + }, +}; + +impl ElfService { + /// Reprojects a stored search session into timeline groups. + pub async fn search_timeline( + &self, + req: SearchTimelineRequest, + ) -> Result { + let tenant_id = req.tenant_id.trim(); + let project_id = req.project_id.trim(); + let agent_id = req.agent_id.trim(); + + if tenant_id.is_empty() || project_id.is_empty() || agent_id.is_empty() { + return Err(Error::InvalidRequest { + message: "tenant_id, project_id, and agent_id are required.".to_string(), + }); + } + + let now = OffsetDateTime::now_utc(); + let session = + storage::load_search_session(&self.db.pool, req.search_session_id, now).await?; + + details::validate_search_session_access(&session, tenant_id, project_id, agent_id)?; + + let expires_at = storage::touch_search_session(&self.db.pool, &session, now).await?; + let payload_level = req.payload_level; + let group_by = req.group_by.unwrap_or_else(|| { + if payload_level == PayloadLevel::L0 { "none".to_string() } else { "day".to_string() } + }); + + match group_by.as_str() { + "day" => details::build_timeline_by_day( + session.search_session_id, + expires_at, + &session.items, + ), + "none" => Ok(SearchTimelineResponse { + search_session_id: session.search_session_id, + expires_at, + groups: vec![SearchTimelineGroup { + date: "all".to_string(), + items: session + .items + .iter() + .map(SearchSessionItemRecord::to_index_item) + .collect(), + }], + }), + _ => Err(Error::InvalidRequest { + message: "group_by must be one of: day, none.".to_string(), + }), + } + } +} From f056beb7a1ae159070177fb1e438a762775268c3 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 30 Jun 2026 15:12:10 +0800 Subject: [PATCH 4/7] {"schema":"decodex/commit/1","summary":"Split progressive search service modules","authority":"manual"} --- .../src/progressive_search/service.rs | 146 +----------------- .../progressive_search/service/entrypoints.rs | 43 ++++++ .../progressive_search/service/sessionized.rs | 108 +++++++++++++ 3 files changed, 153 insertions(+), 144 deletions(-) create mode 100644 packages/elf-service/src/progressive_search/service/entrypoints.rs create mode 100644 packages/elf-service/src/progressive_search/service/sessionized.rs diff --git a/packages/elf-service/src/progressive_search/service.rs b/packages/elf-service/src/progressive_search/service.rs index 04817342..062cb3b6 100644 --- a/packages/elf-service/src/progressive_search/service.rs +++ b/packages/elf-service/src/progressive_search/service.rs @@ -1,144 +1,2 @@ -use time::{Duration, OffsetDateTime}; -use uuid::Uuid; - -use crate::{ - ElfService, Error, Result, SearchRequest, - progressive_search::{ - details, - storage::{self}, - types::{ - NewSearchSession, SESSION_SLIDING_TTL_HOURS, SearchIndexItem, - SearchIndexPlannedResponse, SearchIndexResponse, SearchSessionItemRecord, - SearchSessionMode, SearchSessionizePath, SearchSessionizedOutput, - }, - }, - structured_fields, -}; - -impl ElfService { - /// Runs the default progressive-search path and returns indexed results. - pub async fn search(&self, req: SearchRequest) -> Result { - let response = self.search_planned(req).await?; - - Ok(SearchIndexResponse { - trace_id: response.trace_id, - search_session_id: response.search_session_id, - expires_at: response.expires_at, - items: response.items, - trajectory_summary: response.trajectory_summary, - }) - } - - /// Runs quick-find search and stores a quick session without a query plan. - pub async fn search_quick(&self, req: SearchRequest) -> Result { - self.search_sessionized(req, SearchSessionizePath::Quick).await.map(|output| output.index) - } - - /// Runs planned search and stores a session with a query plan. - pub async fn search_planned(&self, req: SearchRequest) -> Result { - let output = self.search_sessionized(req, SearchSessionizePath::Planned).await?; - let query_plan = output.query_plan.ok_or_else(|| Error::Storage { - message: "Planned search response is missing query_plan.".to_string(), - })?; - - Ok(SearchIndexPlannedResponse { - trace_id: output.index.trace_id, - search_session_id: output.index.search_session_id, - expires_at: output.index.expires_at, - items: output.index.items, - trajectory_summary: output.index.trajectory_summary, - query_plan, - }) - } - - async fn search_sessionized( - &self, - req: SearchRequest, - path: SearchSessionizePath, - ) -> Result { - let top_k = req.top_k.unwrap_or(self.cfg.memory.top_k).max(1); - let candidate_k = req.candidate_k.unwrap_or(self.cfg.memory.candidate_k).max(top_k); - let mut raw_req = req.clone(); - - raw_req.top_k = Some(candidate_k); - raw_req.record_hits = Some(false); - - let (trace_id, raw_items, trajectory_summary, query_plan) = match path { - SearchSessionizePath::Quick => { - let raw = self.search_raw_quick(raw_req).await?; - - (raw.trace_id, raw.items, raw.trajectory_summary, None) - }, - SearchSessionizePath::Planned => { - let raw = self.search_raw_planned(raw_req).await?; - - (raw.trace_id, raw.items, raw.trajectory_summary, Some(raw.query_plan)) - }, - }; - let now = OffsetDateTime::now_utc(); - let expires_at = now + Duration::hours(SESSION_SLIDING_TTL_HOURS); - let search_session_id = Uuid::new_v4(); - let note_ids: Vec = raw_items.iter().map(|item| item.note_id).collect(); - let structured_by_note = - structured_fields::fetch_structured_fields(&self.db.pool, ¬e_ids).await?; - let mut items = Vec::with_capacity(raw_items.len()); - - for (idx, item) in raw_items.iter().enumerate() { - let summary = structured_by_note - .get(&item.note_id) - .and_then(|value| value.summary.clone()) - .unwrap_or_else(|| { - details::build_summary(&item.snippet, self.cfg.memory.max_note_chars as usize) - }); - - items.push(SearchSessionItemRecord { - rank: idx as u32 + 1, - note_id: item.note_id, - chunk_id: item.chunk_id, - final_score: item.final_score, - updated_at: item.updated_at, - expires_at: item.expires_at, - r#type: item.r#type.clone(), - key: item.key.clone(), - scope: item.scope.clone(), - importance: item.importance, - confidence: item.confidence, - summary, - }); - } - - storage::store_search_session( - &self.db.pool, - NewSearchSession { - search_session_id, - trace_id, - tenant_id: &req.tenant_id, - project_id: &req.project_id, - agent_id: &req.agent_id, - read_profile: &req.read_profile, - query: &req.query, - mode: SearchSessionMode::from(path), - query_plan: query_plan.as_ref(), - trajectory_summary: trajectory_summary.as_ref(), - items: &items, - created_at: now, - expires_at, - }, - ) - .await?; - - let response_items: Vec = - items.into_iter().take(top_k as usize).map(|item| item.to_index_item()).collect(); - - Ok(SearchSessionizedOutput { - index: SearchIndexResponse { - trace_id, - search_session_id, - expires_at, - items: response_items, - trajectory_summary, - }, - query_plan, - }) - } -} +mod entrypoints; +mod sessionized; diff --git a/packages/elf-service/src/progressive_search/service/entrypoints.rs b/packages/elf-service/src/progressive_search/service/entrypoints.rs new file mode 100644 index 00000000..752cc745 --- /dev/null +++ b/packages/elf-service/src/progressive_search/service/entrypoints.rs @@ -0,0 +1,43 @@ +use crate::{ + ElfService, Error, Result, SearchRequest, + progressive_search::types::{ + SearchIndexPlannedResponse, SearchIndexResponse, SearchSessionizePath, + }, +}; + +impl ElfService { + /// Runs the default progressive-search path and returns indexed results. + pub async fn search(&self, req: SearchRequest) -> Result { + let response = self.search_planned(req).await?; + + Ok(SearchIndexResponse { + trace_id: response.trace_id, + search_session_id: response.search_session_id, + expires_at: response.expires_at, + items: response.items, + trajectory_summary: response.trajectory_summary, + }) + } + + /// Runs quick-find search and stores a quick session without a query plan. + pub async fn search_quick(&self, req: SearchRequest) -> Result { + self.search_sessionized(req, SearchSessionizePath::Quick).await.map(|output| output.index) + } + + /// Runs planned search and stores a session with a query plan. + pub async fn search_planned(&self, req: SearchRequest) -> Result { + let output = self.search_sessionized(req, SearchSessionizePath::Planned).await?; + let query_plan = output.query_plan.ok_or_else(|| Error::Storage { + message: "Planned search response is missing query_plan.".to_string(), + })?; + + Ok(SearchIndexPlannedResponse { + trace_id: output.index.trace_id, + search_session_id: output.index.search_session_id, + expires_at: output.index.expires_at, + items: output.index.items, + trajectory_summary: output.index.trajectory_summary, + query_plan, + }) + } +} diff --git a/packages/elf-service/src/progressive_search/service/sessionized.rs b/packages/elf-service/src/progressive_search/service/sessionized.rs new file mode 100644 index 00000000..440ef07a --- /dev/null +++ b/packages/elf-service/src/progressive_search/service/sessionized.rs @@ -0,0 +1,108 @@ +use time::{Duration, OffsetDateTime}; +use uuid::Uuid; + +use crate::{ + ElfService, Result, SearchRequest, + progressive_search::{ + details, storage, + types::{ + NewSearchSession, SESSION_SLIDING_TTL_HOURS, SearchIndexItem, SearchIndexResponse, + SearchSessionItemRecord, SearchSessionMode, SearchSessionizePath, + SearchSessionizedOutput, + }, + }, + structured_fields, +}; + +impl ElfService { + pub(super) async fn search_sessionized( + &self, + req: SearchRequest, + path: SearchSessionizePath, + ) -> Result { + let top_k = req.top_k.unwrap_or(self.cfg.memory.top_k).max(1); + let candidate_k = req.candidate_k.unwrap_or(self.cfg.memory.candidate_k).max(top_k); + let mut raw_req = req.clone(); + + raw_req.top_k = Some(candidate_k); + raw_req.record_hits = Some(false); + + let (trace_id, raw_items, trajectory_summary, query_plan) = match path { + SearchSessionizePath::Quick => { + let raw = self.search_raw_quick(raw_req).await?; + + (raw.trace_id, raw.items, raw.trajectory_summary, None) + }, + SearchSessionizePath::Planned => { + let raw = self.search_raw_planned(raw_req).await?; + + (raw.trace_id, raw.items, raw.trajectory_summary, Some(raw.query_plan)) + }, + }; + let now = OffsetDateTime::now_utc(); + let expires_at = now + Duration::hours(SESSION_SLIDING_TTL_HOURS); + let search_session_id = Uuid::new_v4(); + let note_ids: Vec = raw_items.iter().map(|item| item.note_id).collect(); + let structured_by_note = + structured_fields::fetch_structured_fields(&self.db.pool, ¬e_ids).await?; + let mut items = Vec::with_capacity(raw_items.len()); + + for (idx, item) in raw_items.iter().enumerate() { + let summary = structured_by_note + .get(&item.note_id) + .and_then(|value| value.summary.clone()) + .unwrap_or_else(|| { + details::build_summary(&item.snippet, self.cfg.memory.max_note_chars as usize) + }); + + items.push(SearchSessionItemRecord { + rank: idx as u32 + 1, + note_id: item.note_id, + chunk_id: item.chunk_id, + final_score: item.final_score, + updated_at: item.updated_at, + expires_at: item.expires_at, + r#type: item.r#type.clone(), + key: item.key.clone(), + scope: item.scope.clone(), + importance: item.importance, + confidence: item.confidence, + summary, + }); + } + + storage::store_search_session( + &self.db.pool, + NewSearchSession { + search_session_id, + trace_id, + tenant_id: &req.tenant_id, + project_id: &req.project_id, + agent_id: &req.agent_id, + read_profile: &req.read_profile, + query: &req.query, + mode: SearchSessionMode::from(path), + query_plan: query_plan.as_ref(), + trajectory_summary: trajectory_summary.as_ref(), + items: &items, + created_at: now, + expires_at, + }, + ) + .await?; + + let response_items: Vec = + items.into_iter().take(top_k as usize).map(|item| item.to_index_item()).collect(); + + Ok(SearchSessionizedOutput { + index: SearchIndexResponse { + trace_id, + search_session_id, + expires_at, + items: response_items, + trajectory_summary, + }, + query_plan, + }) + } +} From 43335428fe3dc749542365ec8b98378c0a5d53a3 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 30 Jun 2026 15:19:32 +0800 Subject: [PATCH 5/7] {"schema":"decodex/commit/1","summary":"Split recall debug type modules","authority":"manual"} --- packages/elf-service/src/recall_debug.rs | 13 +- .../elf-service/src/recall_debug/types.rs | 248 +----------------- .../src/recall_debug/types/constants.rs | 8 + .../src/recall_debug/types/panel.rs | 79 ++++++ .../src/recall_debug/types/request.rs | 53 ++++ .../elf-service/src/recall_debug/types/row.rs | 33 +++ .../src/recall_debug/types/source_record.rs | 9 + .../src/recall_debug/types/trace.rs | 65 +++++ 8 files changed, 267 insertions(+), 241 deletions(-) create mode 100644 packages/elf-service/src/recall_debug/types/constants.rs create mode 100644 packages/elf-service/src/recall_debug/types/panel.rs create mode 100644 packages/elf-service/src/recall_debug/types/request.rs create mode 100644 packages/elf-service/src/recall_debug/types/row.rs create mode 100644 packages/elf-service/src/recall_debug/types/source_record.rs create mode 100644 packages/elf-service/src/recall_debug/types/trace.rs diff --git a/packages/elf-service/src/recall_debug.rs b/packages/elf-service/src/recall_debug.rs index 5733c40b..e486864f 100644 --- a/packages/elf-service/src/recall_debug.rs +++ b/packages/elf-service/src/recall_debug.rs @@ -15,16 +15,15 @@ pub use types::{ use std::collections::{BTreeMap, BTreeSet, HashSet}; -use serde::{Deserialize, Serialize}; +use serde::Serialize; use serde_json::Value; use time::OffsetDateTime; use uuid::Uuid; use crate::{ - DocsSearchL0Request, DreamingReviewQueueRequest, ElfService, Error, GraphQueryEntityRef, - GraphQueryPredicateRef, GraphReportRequest, KnowledgePageSearchItem, - KnowledgePageSearchRequest, Result, SearchExplainItem, SearchTrace, SearchTrajectoryStage, - TraceBundleGetRequest, + DocsSearchL0Request, DreamingReviewQueueRequest, ElfService, Error, GraphQueryPredicateRef, + GraphReportRequest, KnowledgePageSearchItem, KnowledgePageSearchRequest, Result, + SearchExplainItem, SearchTrace, SearchTrajectoryStage, TraceBundleGetRequest, access::{ORG_PROJECT_ID, SharedSpaceGrantKey}, search::{TraceBundleMode, TraceReplayCandidate}, }; @@ -41,8 +40,8 @@ use trace::{ not_requested_layer, summarize_layers, }; use types::{ - DEFAULT_RECALL_DEBUG_LIMIT, MAX_RECALL_DEBUG_DOCS_LIMIT, MAX_RECALL_DEBUG_LIMIT, - NoteDebugSourceRow, + constants::{DEFAULT_RECALL_DEBUG_LIMIT, MAX_RECALL_DEBUG_DOCS_LIMIT, MAX_RECALL_DEBUG_LIMIT}, + source_record::NoteDebugSourceRow, }; #[cfg(test)] #[path = "recall_debug/tests.rs"] diff --git a/packages/elf-service/src/recall_debug/types.rs b/packages/elf-service/src/recall_debug/types.rs index c5fdec4d..d8ed1a14 100644 --- a/packages/elf-service/src/recall_debug/types.rs +++ b/packages/elf-service/src/recall_debug/types.rs @@ -1,235 +1,15 @@ -use crate::recall_debug::{ - BTreeMap, Deserialize, GraphQueryEntityRef, GraphQueryPredicateRef, OffsetDateTime, Serialize, - Uuid, Value, +pub(in crate::recall_debug) mod constants; +pub(in crate::recall_debug) mod source_record; + +mod panel; +mod request; +mod row; +mod trace; + +pub use self::{ + constants::{ELF_RECALL_DEBUG_PANEL_SCHEMA_V1, ELF_RECALL_TRACE_SCHEMA_V1}, + panel::{RecallDebugLayer, RecallDebugPanelResponse, RecallDebugPanelSummary}, + request::{RecallDebugPanelRequest, RecallDebugPanelRequestEcho}, + row::RecallDebugRow, + trace::{RecallTrace, RecallTraceEntry, RecallTraceSummary}, }; - -/// Schema identifier for recall/debug panel responses. -pub const ELF_RECALL_DEBUG_PANEL_SCHEMA_V1: &str = "elf.recall_debug_panel/v1"; -/// Schema identifier for deterministic recall trace projections. -pub const ELF_RECALL_TRACE_SCHEMA_V1: &str = "elf.recall_trace/v1"; - -pub(super) const DEFAULT_RECALL_DEBUG_LIMIT: u32 = 25; -pub(super) const MAX_RECALL_DEBUG_LIMIT: u32 = 100; -pub(super) const MAX_RECALL_DEBUG_DOCS_LIMIT: u32 = 32; - -/// Request payload for the cross-layer recall/debug panel. -#[derive(Clone, Debug, Deserialize, Serialize)] -pub struct RecallDebugPanelRequest { - /// Tenant that owns the readback. - pub tenant_id: String, - /// Project that owns the readback. - pub project_id: String, - /// Agent requesting the readback. - pub agent_id: String, - /// Read profile used for memory, document, and graph visibility. - pub read_profile: String, - /// Optional search trace anchor for memory selected/dropped rows. - pub trace_id: Option, - /// Shared query used when docs_query or knowledge_query are omitted. - pub query: Option, - /// Optional Source Library query. - pub docs_query: Option, - /// Optional Knowledge Workspace page query. - pub knowledge_query: Option, - /// Optional graph subject selector. - pub graph_subject: Option, - /// Optional graph predicate selector. - pub graph_predicate: Option, - /// Whether to include Dreaming review queue proposals. Omitted means not requested. - pub include_dreaming: Option, - /// Maximum rows per layer. - pub limit: Option, - #[serde(skip)] - /// Whether project-scoped trace anchors are allowed for an admin mirror request. - pub allow_project_trace_debug: bool, -} - -/// Cross-layer recall/debug panel response. -#[derive(Clone, Debug, Serialize)] -pub struct RecallDebugPanelResponse { - /// Response schema identifier. - pub schema: String, - #[serde(with = "crate::time_serde")] - /// Panel generation timestamp. - pub generated_at: OffsetDateTime, - /// Echo of the effective anchors used for this response. - pub request: RecallDebugPanelRequestEcho, - /// Aggregate panel summary. - pub summary: RecallDebugPanelSummary, - /// Deterministic flat trace projection for agents and fixture assertions. - pub recall_trace: RecallTrace, - /// Cross-layer rows grouped by source layer. - pub layers: Vec, -} - -/// Deterministic flat recall trace over all requested layers. -#[derive(Clone, Debug, Serialize)] -pub struct RecallTrace { - /// Trace schema identifier. - pub schema: String, - /// Aggregate trace counters. - pub summary: RecallTraceSummary, - /// Stable trace entries in layer and row order. - pub entries: Vec, -} - -/// Aggregate counters for a recall trace. -#[derive(Clone, Debug, Default, Serialize)] -pub struct RecallTraceSummary { - /// Number of trace entries. - pub entry_count: usize, - /// Entries whose row selection state is selected. - pub selected_count: usize, - /// Entries whose row selection state is dropped. - pub dropped_count: usize, - /// Entries whose freshness state indicates stale or non-current evidence. - pub stale_count: usize, - /// Entries representing blocked layers. - pub blocked_count: usize, - /// Entries representing layers that were not requested. - pub not_requested_count: usize, - /// Entries that require raw SQL for diagnosis. - pub raw_sql_needed_count: usize, - /// Entries with a replay command or deterministic artifact path. - pub replay_command_count: usize, -} - -/// One compact recall trace entry. -#[derive(Clone, Debug, Serialize)] -pub struct RecallTraceEntry { - /// Layer identifier. - pub layer: String, - /// Primary trace state for compact assertions. - pub context_state: String, - /// Original row selection state or layer evidence class. - pub selection_state: String, - /// Authority layer that owns the context. - pub authority_layer: String, - /// Freshness or temporal state. - pub freshness_state: String, - /// Stable identifiers for replay or hydration. - pub item_ref: Value, - /// Source refs or source snapshots supporting the context. - pub source_refs: Value, - /// Optional score. - pub score: Option, - /// Optional rank. - pub rank: Option, - /// Compact policy or stage reason for the state. - pub policy_reason: Option, - /// Replay command or deterministic artifact path. - pub replay_command: Option, - /// Layer or row evidence class. - pub evidence_class: String, - /// Whether raw SQL is required to diagnose this entry. - pub raw_sql_needed: bool, -} - -/// Stable request echo for panel responses. -#[derive(Clone, Debug, Serialize)] -pub struct RecallDebugPanelRequestEcho { - /// Search trace anchor used for memory rows. - pub trace_id: Option, - /// Effective Source Library query. - pub docs_query: Option, - /// Effective Knowledge Workspace query. - pub knowledge_query: Option, - /// Whether a graph subject was supplied. - pub graph_subject_supplied: bool, - /// Whether Dreaming proposals were included. - pub include_dreaming: bool, - /// Effective row cap per layer. - pub limit: u32, -} - -/// Aggregate panel counters. -#[derive(Clone, Debug, Default, Serialize)] -pub struct RecallDebugPanelSummary { - /// Number of returned layers. - pub layer_count: usize, - /// Total returned row count. - pub row_count: usize, - /// Rows selected by a retrieval or review stage. - pub selected_count: usize, - /// Rows dropped by a retrieval or review stage. - pub dropped_count: usize, - /// Rows available for inspection but not selected/dropped. - pub available_count: usize, - /// Layers skipped because no anchor was supplied. - pub not_requested_layer_count: usize, - /// Layers that require follow-up before they can prove a debug claim. - pub incomplete_layer_count: usize, - /// Rows or layers that require raw SQL to inspect. - pub raw_sql_needed_count: usize, - /// Rows with a replay command or deterministic artifact path. - pub replay_command_count: usize, - /// Evidence-class counts across layers. - pub evidence_class_counts: BTreeMap, -} - -/// One recall/debug source layer. -#[derive(Clone, Debug, Serialize)] -pub struct RecallDebugLayer { - /// Layer identifier. - pub layer: String, - /// Evidence class for this layer. - pub evidence_class: String, - /// Human-readable layer summary. - pub summary: String, - /// Query or object anchor used by the layer. - pub anchor: Option, - /// Number of returned rows. - pub row_count: usize, - /// Selected rows in this layer. - pub selected_count: usize, - /// Dropped rows in this layer. - pub dropped_count: usize, - /// Available review/inspection rows in this layer. - pub available_count: usize, - /// Whether raw SQL is needed to inspect this layer. - pub raw_sql_needed: bool, - /// Whether the layer includes replay commands or deterministic artifact paths. - pub replayable: bool, - /// Compact layer-level debug artifacts. - pub debug_artifacts: Value, - /// Returned layer rows. - pub rows: Vec, -} - -/// One item in the recall/debug panel. -#[derive(Clone, Debug, Serialize)] -pub struct RecallDebugRow { - /// Layer identifier. - pub layer: String, - /// Stable item reference. - pub item_ref: Value, - /// Selection state such as selected, dropped, available, or reviewable. - pub selection_state: String, - /// Authority layer that owns the row. - pub authority_layer: String, - /// Freshness or temporal state. - pub freshness_state: String, - /// Source refs or source snapshots backing the row. - pub source_refs: Value, - /// Optional final score. - pub score: Option, - /// Optional rank within the layer. - pub rank: Option, - /// Short selection rationale. - pub rationale: Option, - /// Stage reason for selected/dropped status. - pub stage_reason: Option, - /// Replay command or deterministic artifact path when available. - pub replay_command: Option, - /// Row-level evidence class. - pub evidence_class: String, - /// Layer-specific debug artifacts. - pub debug_artifacts: Value, -} - -#[derive(Clone, Debug)] -pub(super) struct NoteDebugSourceRow { - pub(super) status: String, - pub(super) source_ref: Value, - pub(super) updated_at: OffsetDateTime, -} diff --git a/packages/elf-service/src/recall_debug/types/constants.rs b/packages/elf-service/src/recall_debug/types/constants.rs new file mode 100644 index 00000000..224636ee --- /dev/null +++ b/packages/elf-service/src/recall_debug/types/constants.rs @@ -0,0 +1,8 @@ +/// Schema identifier for recall/debug panel responses. +pub const ELF_RECALL_DEBUG_PANEL_SCHEMA_V1: &str = "elf.recall_debug_panel/v1"; +/// Schema identifier for deterministic recall trace projections. +pub const ELF_RECALL_TRACE_SCHEMA_V1: &str = "elf.recall_trace/v1"; + +pub(in crate::recall_debug) const DEFAULT_RECALL_DEBUG_LIMIT: u32 = 25; +pub(in crate::recall_debug) const MAX_RECALL_DEBUG_LIMIT: u32 = 100; +pub(in crate::recall_debug) const MAX_RECALL_DEBUG_DOCS_LIMIT: u32 = 32; diff --git a/packages/elf-service/src/recall_debug/types/panel.rs b/packages/elf-service/src/recall_debug/types/panel.rs new file mode 100644 index 00000000..625c7b29 --- /dev/null +++ b/packages/elf-service/src/recall_debug/types/panel.rs @@ -0,0 +1,79 @@ +use std::collections::BTreeMap; + +use serde::Serialize; +use serde_json::Value; +use time::OffsetDateTime; + +use crate::recall_debug::{RecallDebugPanelRequestEcho, RecallDebugRow, RecallTrace}; + +/// Cross-layer recall/debug panel response. +#[derive(Clone, Debug, Serialize)] +pub struct RecallDebugPanelResponse { + /// Response schema identifier. + pub schema: String, + #[serde(with = "crate::time_serde")] + /// Panel generation timestamp. + pub generated_at: OffsetDateTime, + /// Echo of the effective anchors used for this response. + pub request: RecallDebugPanelRequestEcho, + /// Aggregate panel summary. + pub summary: RecallDebugPanelSummary, + /// Deterministic flat trace projection for agents and fixture assertions. + pub recall_trace: RecallTrace, + /// Cross-layer rows grouped by source layer. + pub layers: Vec, +} + +/// Aggregate panel counters. +#[derive(Clone, Debug, Default, Serialize)] +pub struct RecallDebugPanelSummary { + /// Number of returned layers. + pub layer_count: usize, + /// Total returned row count. + pub row_count: usize, + /// Rows selected by a retrieval or review stage. + pub selected_count: usize, + /// Rows dropped by a retrieval or review stage. + pub dropped_count: usize, + /// Rows available for inspection but not selected/dropped. + pub available_count: usize, + /// Layers skipped because no anchor was supplied. + pub not_requested_layer_count: usize, + /// Layers that require follow-up before they can prove a debug claim. + pub incomplete_layer_count: usize, + /// Rows or layers that require raw SQL to inspect. + pub raw_sql_needed_count: usize, + /// Rows with a replay command or deterministic artifact path. + pub replay_command_count: usize, + /// Evidence-class counts across layers. + pub evidence_class_counts: BTreeMap, +} + +/// One recall/debug source layer. +#[derive(Clone, Debug, Serialize)] +pub struct RecallDebugLayer { + /// Layer identifier. + pub layer: String, + /// Evidence class for this layer. + pub evidence_class: String, + /// Human-readable layer summary. + pub summary: String, + /// Query or object anchor used by the layer. + pub anchor: Option, + /// Number of returned rows. + pub row_count: usize, + /// Selected rows in this layer. + pub selected_count: usize, + /// Dropped rows in this layer. + pub dropped_count: usize, + /// Available review/inspection rows in this layer. + pub available_count: usize, + /// Whether raw SQL is needed to inspect this layer. + pub raw_sql_needed: bool, + /// Whether the layer includes replay commands or deterministic artifact paths. + pub replayable: bool, + /// Compact layer-level debug artifacts. + pub debug_artifacts: Value, + /// Returned layer rows. + pub rows: Vec, +} diff --git a/packages/elf-service/src/recall_debug/types/request.rs b/packages/elf-service/src/recall_debug/types/request.rs new file mode 100644 index 00000000..30b32837 --- /dev/null +++ b/packages/elf-service/src/recall_debug/types/request.rs @@ -0,0 +1,53 @@ +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::{GraphQueryEntityRef, GraphQueryPredicateRef}; + +/// Request payload for the cross-layer recall/debug panel. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct RecallDebugPanelRequest { + /// Tenant that owns the readback. + pub tenant_id: String, + /// Project that owns the readback. + pub project_id: String, + /// Agent requesting the readback. + pub agent_id: String, + /// Read profile used for memory, document, and graph visibility. + pub read_profile: String, + /// Optional search trace anchor for memory selected/dropped rows. + pub trace_id: Option, + /// Shared query used when docs_query or knowledge_query are omitted. + pub query: Option, + /// Optional Source Library query. + pub docs_query: Option, + /// Optional Knowledge Workspace page query. + pub knowledge_query: Option, + /// Optional graph subject selector. + pub graph_subject: Option, + /// Optional graph predicate selector. + pub graph_predicate: Option, + /// Whether to include Dreaming review queue proposals. Omitted means not requested. + pub include_dreaming: Option, + /// Maximum rows per layer. + pub limit: Option, + #[serde(skip)] + /// Whether project-scoped trace anchors are allowed for an admin mirror request. + pub allow_project_trace_debug: bool, +} + +/// Stable request echo for panel responses. +#[derive(Clone, Debug, Serialize)] +pub struct RecallDebugPanelRequestEcho { + /// Search trace anchor used for memory rows. + pub trace_id: Option, + /// Effective Source Library query. + pub docs_query: Option, + /// Effective Knowledge Workspace query. + pub knowledge_query: Option, + /// Whether a graph subject was supplied. + pub graph_subject_supplied: bool, + /// Whether Dreaming proposals were included. + pub include_dreaming: bool, + /// Effective row cap per layer. + pub limit: u32, +} diff --git a/packages/elf-service/src/recall_debug/types/row.rs b/packages/elf-service/src/recall_debug/types/row.rs new file mode 100644 index 00000000..40fb8a0c --- /dev/null +++ b/packages/elf-service/src/recall_debug/types/row.rs @@ -0,0 +1,33 @@ +use serde::Serialize; +use serde_json::Value; + +/// One item in the recall/debug panel. +#[derive(Clone, Debug, Serialize)] +pub struct RecallDebugRow { + /// Layer identifier. + pub layer: String, + /// Stable item reference. + pub item_ref: Value, + /// Selection state such as selected, dropped, available, or reviewable. + pub selection_state: String, + /// Authority layer that owns the row. + pub authority_layer: String, + /// Freshness or temporal state. + pub freshness_state: String, + /// Source refs or source snapshots backing the row. + pub source_refs: Value, + /// Optional final score. + pub score: Option, + /// Optional rank within the layer. + pub rank: Option, + /// Short selection rationale. + pub rationale: Option, + /// Stage reason for selected/dropped status. + pub stage_reason: Option, + /// Replay command or deterministic artifact path when available. + pub replay_command: Option, + /// Row-level evidence class. + pub evidence_class: String, + /// Layer-specific debug artifacts. + pub debug_artifacts: Value, +} diff --git a/packages/elf-service/src/recall_debug/types/source_record.rs b/packages/elf-service/src/recall_debug/types/source_record.rs new file mode 100644 index 00000000..c0047da9 --- /dev/null +++ b/packages/elf-service/src/recall_debug/types/source_record.rs @@ -0,0 +1,9 @@ +use serde_json::Value; +use time::OffsetDateTime; + +#[derive(Clone, Debug)] +pub(in crate::recall_debug) struct NoteDebugSourceRow { + pub(in crate::recall_debug) status: String, + pub(in crate::recall_debug) source_ref: Value, + pub(in crate::recall_debug) updated_at: OffsetDateTime, +} diff --git a/packages/elf-service/src/recall_debug/types/trace.rs b/packages/elf-service/src/recall_debug/types/trace.rs new file mode 100644 index 00000000..c6e3c944 --- /dev/null +++ b/packages/elf-service/src/recall_debug/types/trace.rs @@ -0,0 +1,65 @@ +use serde::Serialize; +use serde_json::Value; + +/// Deterministic flat recall trace over all requested layers. +#[derive(Clone, Debug, Serialize)] +pub struct RecallTrace { + /// Trace schema identifier. + pub schema: String, + /// Aggregate trace counters. + pub summary: RecallTraceSummary, + /// Stable trace entries in layer and row order. + pub entries: Vec, +} + +/// Aggregate counters for a recall trace. +#[derive(Clone, Debug, Default, Serialize)] +pub struct RecallTraceSummary { + /// Number of trace entries. + pub entry_count: usize, + /// Entries whose row selection state is selected. + pub selected_count: usize, + /// Entries whose row selection state is dropped. + pub dropped_count: usize, + /// Entries whose freshness state indicates stale or non-current evidence. + pub stale_count: usize, + /// Entries representing blocked layers. + pub blocked_count: usize, + /// Entries representing layers that were not requested. + pub not_requested_count: usize, + /// Entries that require raw SQL for diagnosis. + pub raw_sql_needed_count: usize, + /// Entries with a replay command or deterministic artifact path. + pub replay_command_count: usize, +} + +/// One compact recall trace entry. +#[derive(Clone, Debug, Serialize)] +pub struct RecallTraceEntry { + /// Layer identifier. + pub layer: String, + /// Primary trace state for compact assertions. + pub context_state: String, + /// Original row selection state or layer evidence class. + pub selection_state: String, + /// Authority layer that owns the context. + pub authority_layer: String, + /// Freshness or temporal state. + pub freshness_state: String, + /// Stable identifiers for replay or hydration. + pub item_ref: Value, + /// Source refs or source snapshots supporting the context. + pub source_refs: Value, + /// Optional score. + pub score: Option, + /// Optional rank. + pub rank: Option, + /// Compact policy or stage reason for the state. + pub policy_reason: Option, + /// Replay command or deterministic artifact path. + pub replay_command: Option, + /// Layer or row evidence class. + pub evidence_class: String, + /// Whether raw SQL is required to diagnose this entry. + pub raw_sql_needed: bool, +} From e3c9484cba63e7f3ad99683ede3a5844989fb5e9 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 30 Jun 2026 15:22:39 +0800 Subject: [PATCH 6/7] {"schema":"decodex/commit/1","summary":"Split recall debug trace modules","authority":"manual"} --- .../elf-service/src/recall_debug/trace.rs | 248 +----------------- .../src/recall_debug/trace/entries.rs | 80 ++++++ .../src/recall_debug/trace/layer_builders.rs | 89 +++++++ .../src/recall_debug/trace/summary.rs | 78 ++++++ 4 files changed, 257 insertions(+), 238 deletions(-) create mode 100644 packages/elf-service/src/recall_debug/trace/entries.rs create mode 100644 packages/elf-service/src/recall_debug/trace/layer_builders.rs create mode 100644 packages/elf-service/src/recall_debug/trace/summary.rs diff --git a/packages/elf-service/src/recall_debug/trace.rs b/packages/elf-service/src/recall_debug/trace.rs index e8e1eeff..ff0bc8ab 100644 --- a/packages/elf-service/src/recall_debug/trace.rs +++ b/packages/elf-service/src/recall_debug/trace.rs @@ -1,239 +1,11 @@ -use crate::recall_debug::{ - self, ELF_RECALL_TRACE_SCHEMA_V1, Error, RecallDebugLayer, RecallDebugPanelSummary, - RecallDebugRow, RecallTrace, RecallTraceEntry, RecallTraceSummary, Value, +mod entries; +mod layer_builders; +mod summary; + +pub(super) use self::{ + entries::build_recall_trace, + layer_builders::{ + blocked_layer, layer_from_rows, layer_from_rows_with_artifacts, not_requested_layer, + }, + summary::summarize_layers, }; - -pub(super) fn summarize_layers(layers: &[RecallDebugLayer]) -> RecallDebugPanelSummary { - let mut summary = RecallDebugPanelSummary { layer_count: layers.len(), ..Default::default() }; - - for layer in layers { - summary.row_count += layer.row_count; - summary.selected_count += layer.selected_count; - summary.dropped_count += layer.dropped_count; - summary.available_count += layer.available_count; - - if layer.evidence_class == "not_requested" { - summary.not_requested_layer_count += 1; - } - if matches!(layer.evidence_class.as_str(), "incomplete" | "blocked" | "wrong_result") { - summary.incomplete_layer_count += 1; - } - if layer.raw_sql_needed { - summary.raw_sql_needed_count += 1; - } - - summary.replay_command_count += layer - .rows - .iter() - .filter(|row| row.replay_command.as_ref().is_some_and(|value| !value.is_empty())) - .count(); - *summary.evidence_class_counts.entry(layer.evidence_class.clone()).or_default() += 1; - } - - summary -} - -pub(super) fn build_recall_trace(layers: &[RecallDebugLayer]) -> RecallTrace { - let mut entries = Vec::new(); - - for layer in layers { - if layer.rows.is_empty() { - if matches!( - layer.evidence_class.as_str(), - "blocked" | "not_requested" | "incomplete" | "wrong_result" - ) { - entries.push(layer_trace_entry(layer)); - } - - continue; - } - - entries.extend(layer.rows.iter().map(row_trace_entry)); - } - - let summary = summarize_trace_entries(&entries); - - RecallTrace { schema: ELF_RECALL_TRACE_SCHEMA_V1.to_string(), summary, entries } -} - -pub(super) fn summarize_trace_entries(entries: &[RecallTraceEntry]) -> RecallTraceSummary { - let mut summary = RecallTraceSummary { entry_count: entries.len(), ..Default::default() }; - - for entry in entries { - match entry.selection_state.as_str() { - "selected" => summary.selected_count += 1, - "dropped" => summary.dropped_count += 1, - "blocked" => summary.blocked_count += 1, - "not_requested" => summary.not_requested_count += 1, - _ => {}, - } - - if entry.context_state == "stale" || stale_freshness_state(&entry.freshness_state) { - summary.stale_count += 1; - } - if entry.raw_sql_needed { - summary.raw_sql_needed_count += 1; - } - if entry.replay_command.as_ref().is_some_and(|value| !value.is_empty()) { - summary.replay_command_count += 1; - } - } - - summary -} - -pub(super) fn layer_trace_entry(layer: &RecallDebugLayer) -> RecallTraceEntry { - let context_state = match layer.evidence_class.as_str() { - "not_requested" => "not_requested", - "blocked" => "blocked", - "incomplete" => "incomplete", - "wrong_result" => "wrong_result", - _ => "available", - }; - - RecallTraceEntry { - layer: layer.layer.clone(), - context_state: context_state.to_string(), - selection_state: layer.evidence_class.clone(), - authority_layer: layer.layer.clone(), - freshness_state: layer.evidence_class.clone(), - item_ref: serde_json::json!({ - "layer": layer.layer.clone(), - "anchor": layer.anchor.clone(), - }), - source_refs: serde_json::json!([]), - score: None, - rank: None, - policy_reason: Some(layer.summary.clone()), - replay_command: None, - evidence_class: layer.evidence_class.clone(), - raw_sql_needed: layer.raw_sql_needed, - } -} - -pub(super) fn row_trace_entry(row: &RecallDebugRow) -> RecallTraceEntry { - let context_state = if stale_freshness_state(&row.freshness_state) { - "stale" - } else { - row.selection_state.as_str() - }; - - RecallTraceEntry { - layer: row.layer.clone(), - context_state: context_state.to_string(), - selection_state: row.selection_state.clone(), - authority_layer: row.authority_layer.clone(), - freshness_state: row.freshness_state.clone(), - item_ref: row.item_ref.clone(), - source_refs: row.source_refs.clone(), - score: row.score, - rank: row.rank, - policy_reason: row.stage_reason.clone().or_else(|| row.rationale.clone()), - replay_command: row.replay_command.clone(), - evidence_class: row.evidence_class.clone(), - raw_sql_needed: false, - } -} - -pub(super) fn stale_freshness_state(freshness_state: &str) -> bool { - matches!( - freshness_state, - "stale" - | "deprecated" - | "deleted" - | "superseded" - | "tombstoned" - | "historical" - | "archived" - | "lint_warning" - | "lint_error" - ) -} - -pub(super) fn layer_from_rows( - layer: &str, - evidence_class: &str, - anchor: Option, - summary: &str, - rows: Vec, -) -> RecallDebugLayer { - layer_from_rows_with_artifacts( - layer, - evidence_class, - anchor, - summary, - rows, - serde_json::json!({}), - ) -} - -pub(super) fn layer_from_rows_with_artifacts( - layer: &str, - evidence_class: &str, - anchor: Option, - summary: &str, - rows: Vec, - debug_artifacts: Value, -) -> RecallDebugLayer { - let selected_count = rows.iter().filter(|row| row.selection_state == "selected").count(); - let dropped_count = rows.iter().filter(|row| row.selection_state == "dropped").count(); - let available_count = rows - .iter() - .filter(|row| matches!(row.selection_state.as_str(), "available" | "reviewable")) - .count(); - let replayable = rows.iter().any(|row| row.replay_command.is_some()); - - RecallDebugLayer { - layer: layer.to_string(), - evidence_class: evidence_class.to_string(), - summary: summary.to_string(), - anchor, - row_count: rows.len(), - selected_count, - dropped_count, - available_count, - raw_sql_needed: false, - replayable, - debug_artifacts, - rows, - } -} - -pub(super) fn not_requested_layer(layer: &str, summary: &str) -> RecallDebugLayer { - RecallDebugLayer { - layer: layer.to_string(), - evidence_class: "not_requested".to_string(), - summary: summary.to_string(), - anchor: None, - row_count: 0, - selected_count: 0, - dropped_count: 0, - available_count: 0, - raw_sql_needed: false, - replayable: false, - debug_artifacts: serde_json::json!({}), - rows: Vec::new(), - } -} - -pub(super) fn blocked_layer( - layer: &str, - anchor: Option, - summary: &str, - err: &Error, -) -> RecallDebugLayer { - RecallDebugLayer { - layer: layer.to_string(), - evidence_class: "blocked".to_string(), - summary: format!("{summary} error_class={}", recall_debug::public_error_class(err)), - anchor, - row_count: 0, - selected_count: 0, - dropped_count: 0, - available_count: 0, - raw_sql_needed: false, - replayable: false, - debug_artifacts: serde_json::json!({}), - rows: Vec::new(), - } -} diff --git a/packages/elf-service/src/recall_debug/trace/entries.rs b/packages/elf-service/src/recall_debug/trace/entries.rs new file mode 100644 index 00000000..2af438d1 --- /dev/null +++ b/packages/elf-service/src/recall_debug/trace/entries.rs @@ -0,0 +1,80 @@ +use crate::recall_debug::{ + ELF_RECALL_TRACE_SCHEMA_V1, RecallDebugLayer, RecallDebugRow, RecallTrace, RecallTraceEntry, + trace::summary, +}; + +pub(in crate::recall_debug) fn build_recall_trace(layers: &[RecallDebugLayer]) -> RecallTrace { + let mut entries = Vec::new(); + + for layer in layers { + if layer.rows.is_empty() { + if matches!( + layer.evidence_class.as_str(), + "blocked" | "not_requested" | "incomplete" | "wrong_result" + ) { + entries.push(layer_trace_entry(layer)); + } + + continue; + } + + entries.extend(layer.rows.iter().map(row_trace_entry)); + } + + let summary = summary::summarize_trace_entries(&entries); + + RecallTrace { schema: ELF_RECALL_TRACE_SCHEMA_V1.to_string(), summary, entries } +} + +fn layer_trace_entry(layer: &RecallDebugLayer) -> RecallTraceEntry { + let context_state = match layer.evidence_class.as_str() { + "not_requested" => "not_requested", + "blocked" => "blocked", + "incomplete" => "incomplete", + "wrong_result" => "wrong_result", + _ => "available", + }; + + RecallTraceEntry { + layer: layer.layer.clone(), + context_state: context_state.to_string(), + selection_state: layer.evidence_class.clone(), + authority_layer: layer.layer.clone(), + freshness_state: layer.evidence_class.clone(), + item_ref: serde_json::json!({ + "layer": layer.layer.clone(), + "anchor": layer.anchor.clone(), + }), + source_refs: serde_json::json!([]), + score: None, + rank: None, + policy_reason: Some(layer.summary.clone()), + replay_command: None, + evidence_class: layer.evidence_class.clone(), + raw_sql_needed: layer.raw_sql_needed, + } +} + +fn row_trace_entry(row: &RecallDebugRow) -> RecallTraceEntry { + let context_state = if summary::stale_freshness_state(&row.freshness_state) { + "stale" + } else { + row.selection_state.as_str() + }; + + RecallTraceEntry { + layer: row.layer.clone(), + context_state: context_state.to_string(), + selection_state: row.selection_state.clone(), + authority_layer: row.authority_layer.clone(), + freshness_state: row.freshness_state.clone(), + item_ref: row.item_ref.clone(), + source_refs: row.source_refs.clone(), + score: row.score, + rank: row.rank, + policy_reason: row.stage_reason.clone().or_else(|| row.rationale.clone()), + replay_command: row.replay_command.clone(), + evidence_class: row.evidence_class.clone(), + raw_sql_needed: false, + } +} diff --git a/packages/elf-service/src/recall_debug/trace/layer_builders.rs b/packages/elf-service/src/recall_debug/trace/layer_builders.rs new file mode 100644 index 00000000..03ffa8f8 --- /dev/null +++ b/packages/elf-service/src/recall_debug/trace/layer_builders.rs @@ -0,0 +1,89 @@ +use crate::recall_debug::{self, Error, RecallDebugLayer, RecallDebugRow, Value}; + +pub(in crate::recall_debug) fn layer_from_rows( + layer: &str, + evidence_class: &str, + anchor: Option, + summary: &str, + rows: Vec, +) -> RecallDebugLayer { + layer_from_rows_with_artifacts( + layer, + evidence_class, + anchor, + summary, + rows, + serde_json::json!({}), + ) +} + +pub(in crate::recall_debug) fn layer_from_rows_with_artifacts( + layer: &str, + evidence_class: &str, + anchor: Option, + summary: &str, + rows: Vec, + debug_artifacts: Value, +) -> RecallDebugLayer { + let selected_count = rows.iter().filter(|row| row.selection_state == "selected").count(); + let dropped_count = rows.iter().filter(|row| row.selection_state == "dropped").count(); + let available_count = rows + .iter() + .filter(|row| matches!(row.selection_state.as_str(), "available" | "reviewable")) + .count(); + let replayable = rows.iter().any(|row| row.replay_command.is_some()); + + RecallDebugLayer { + layer: layer.to_string(), + evidence_class: evidence_class.to_string(), + summary: summary.to_string(), + anchor, + row_count: rows.len(), + selected_count, + dropped_count, + available_count, + raw_sql_needed: false, + replayable, + debug_artifacts, + rows, + } +} + +pub(in crate::recall_debug) fn not_requested_layer(layer: &str, summary: &str) -> RecallDebugLayer { + RecallDebugLayer { + layer: layer.to_string(), + evidence_class: "not_requested".to_string(), + summary: summary.to_string(), + anchor: None, + row_count: 0, + selected_count: 0, + dropped_count: 0, + available_count: 0, + raw_sql_needed: false, + replayable: false, + debug_artifacts: serde_json::json!({}), + rows: Vec::new(), + } +} + +pub(in crate::recall_debug) fn blocked_layer( + layer: &str, + anchor: Option, + summary: &str, + err: &Error, +) -> RecallDebugLayer { + RecallDebugLayer { + layer: layer.to_string(), + evidence_class: "blocked".to_string(), + summary: format!("{summary} error_class={}", recall_debug::public_error_class(err)), + anchor, + row_count: 0, + selected_count: 0, + dropped_count: 0, + available_count: 0, + raw_sql_needed: false, + replayable: false, + debug_artifacts: serde_json::json!({}), + rows: Vec::new(), + } +} diff --git a/packages/elf-service/src/recall_debug/trace/summary.rs b/packages/elf-service/src/recall_debug/trace/summary.rs new file mode 100644 index 00000000..0d600256 --- /dev/null +++ b/packages/elf-service/src/recall_debug/trace/summary.rs @@ -0,0 +1,78 @@ +use crate::recall_debug::{ + RecallDebugLayer, RecallDebugPanelSummary, RecallTraceEntry, RecallTraceSummary, +}; + +pub(in crate::recall_debug) fn summarize_layers( + layers: &[RecallDebugLayer], +) -> RecallDebugPanelSummary { + let mut summary = RecallDebugPanelSummary { layer_count: layers.len(), ..Default::default() }; + + for layer in layers { + summary.row_count += layer.row_count; + summary.selected_count += layer.selected_count; + summary.dropped_count += layer.dropped_count; + summary.available_count += layer.available_count; + + if layer.evidence_class == "not_requested" { + summary.not_requested_layer_count += 1; + } + if matches!(layer.evidence_class.as_str(), "incomplete" | "blocked" | "wrong_result") { + summary.incomplete_layer_count += 1; + } + if layer.raw_sql_needed { + summary.raw_sql_needed_count += 1; + } + + summary.replay_command_count += layer + .rows + .iter() + .filter(|row| row.replay_command.as_ref().is_some_and(|value| !value.is_empty())) + .count(); + *summary.evidence_class_counts.entry(layer.evidence_class.clone()).or_default() += 1; + } + + summary +} + +pub(in crate::recall_debug::trace) fn summarize_trace_entries( + entries: &[RecallTraceEntry], +) -> RecallTraceSummary { + let mut summary = RecallTraceSummary { entry_count: entries.len(), ..Default::default() }; + + for entry in entries { + match entry.selection_state.as_str() { + "selected" => summary.selected_count += 1, + "dropped" => summary.dropped_count += 1, + "blocked" => summary.blocked_count += 1, + "not_requested" => summary.not_requested_count += 1, + _ => {}, + } + + if entry.context_state == "stale" || stale_freshness_state(&entry.freshness_state) { + summary.stale_count += 1; + } + if entry.raw_sql_needed { + summary.raw_sql_needed_count += 1; + } + if entry.replay_command.as_ref().is_some_and(|value| !value.is_empty()) { + summary.replay_command_count += 1; + } + } + + summary +} + +pub(in crate::recall_debug::trace) fn stale_freshness_state(freshness_state: &str) -> bool { + matches!( + freshness_state, + "stale" + | "deprecated" + | "deleted" + | "superseded" + | "tombstoned" + | "historical" + | "archived" + | "lint_warning" + | "lint_error" + ) +} From ea50acc933dd5a99487b6f5675f9c212f1134c21 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 30 Jun 2026 15:25:34 +0800 Subject: [PATCH 7/7] {"schema":"decodex/commit/1","summary":"Split recall debug replay modules","authority":"manual"} --- .../elf-service/src/recall_debug/replay.rs | 254 +----------------- .../src/recall_debug/replay/artifact.rs | 31 +++ .../recall_debug/replay/candidate_replay.rs | 86 ++++++ .../src/recall_debug/replay/candidate_row.rs | 50 ++++ .../src/recall_debug/replay/controls.rs | 42 +++ .../recall_debug/replay/selected_context.rs | 42 +++ 6 files changed, 259 insertions(+), 246 deletions(-) create mode 100644 packages/elf-service/src/recall_debug/replay/artifact.rs create mode 100644 packages/elf-service/src/recall_debug/replay/candidate_replay.rs create mode 100644 packages/elf-service/src/recall_debug/replay/candidate_row.rs create mode 100644 packages/elf-service/src/recall_debug/replay/controls.rs create mode 100644 packages/elf-service/src/recall_debug/replay/selected_context.rs diff --git a/packages/elf-service/src/recall_debug/replay.rs b/packages/elf-service/src/recall_debug/replay.rs index c483d09f..23fb486c 100644 --- a/packages/elf-service/src/recall_debug/replay.rs +++ b/packages/elf-service/src/recall_debug/replay.rs @@ -1,247 +1,9 @@ -use crate::recall_debug::{ - self, BTreeMap, BTreeSet, NoteDebugSourceRow, RecallDebugRow, SearchExplainItem, SearchTrace, - SearchTrajectoryStage, TraceReplayCandidate, Uuid, Value, +mod artifact; +mod candidate_replay; +mod candidate_row; +mod controls; +mod selected_context; + +pub(super) use self::{ + artifact::memory_compact_replay_artifact, candidate_row::candidate_debug_row, }; - -pub(super) fn candidate_debug_row( - trace_id: Uuid, - candidate: &TraceReplayCandidate, - source: Option<&NoteDebugSourceRow>, - replay_command: &str, -) -> RecallDebugRow { - let selected_by_diversity = candidate.diversity_selected.unwrap_or(false); - let skipped_reason = candidate.diversity_skipped_reason.clone().or_else(|| { - if selected_by_diversity { - candidate.diversity_selected_reason.clone() - } else { - Some("not_in_final_top_k".to_string()) - } - }); - - RecallDebugRow { - layer: "memory_notes".to_string(), - item_ref: serde_json::json!({ - "trace_id": trace_id, - "note_id": candidate.note_id, - "chunk_id": candidate.chunk_id, - "chunk_index": candidate.chunk_index, - }), - selection_state: "dropped".to_string(), - authority_layer: "memory_note".to_string(), - freshness_state: recall_debug::freshness_from_note_source(source), - source_refs: recall_debug::source_ref_from_note_source(source), - score: candidate.retrieval_score, - rank: Some(candidate.retrieval_rank), - rationale: Some( - "candidate captured for replay but not selected in final result set".to_string(), - ), - stage_reason: skipped_reason, - replay_command: Some(replay_command.to_string()), - evidence_class: "pass".to_string(), - debug_artifacts: serde_json::json!({ - "snippet": candidate.snippet, - "rerank_score": candidate.rerank_score, - "note_scope": candidate.note_scope, - "diversity_selected": candidate.diversity_selected, - "diversity_selected_rank": candidate.diversity_selected_rank, - "diversity_nearest_selected_note_id": candidate.diversity_nearest_selected_note_id, - "diversity_similarity": candidate.diversity_similarity, - "diversity_mmr_score": candidate.diversity_mmr_score, - "diversity_missing_embedding": candidate.diversity_missing_embedding, - }), - } -} - -pub(super) fn memory_compact_replay_artifact( - trace: &SearchTrace, - stages: &[SearchTrajectoryStage], - candidates: &[TraceReplayCandidate], - selected_items: &[&SearchExplainItem], - selected_candidate_keys: &BTreeSet<(Uuid, Uuid)>, - source_refs: &BTreeMap, - replay_command: &str, -) -> Value { - serde_json::json!({ - "schema": "elf.recall_debug.compact_replay/v1", - "trace_id": trace.trace_id, - "query": trace.query, - "replay_command": replay_command, - "controls": compact_replay_controls(trace), - "stage_movement": compact_stage_movement(stages), - "candidate_replay": compact_candidate_replay(candidates, selected_candidate_keys, source_refs), - "selected_context": compact_selected_context(selected_items, source_refs), - "authority": { - "source_refs_visible": true, - "policy_reasons_visible": true, - "raw_sql_needed": false, - }, - }) -} - -pub(super) fn compact_replay_controls(trace: &SearchTrace) -> Value { - serde_json::json!({ - "top_k": trace.top_k, - "candidate_count": trace.candidate_count, - "expansion_mode": trace.expansion_mode, - "expanded_query_count": trace.expanded_queries.len(), - "expanded_queries": trace.expanded_queries, - "allowed_scopes": trace.allowed_scopes, - "search": compact_pointer(&trace.config_snapshot, "/search"), - "ranking": { - "policy_id": compact_pointer(&trace.config_snapshot, "/ranking/policy_id"), - "blend": compact_pointer(&trace.config_snapshot, "/ranking/blend"), - "diversity": compact_pointer(&trace.config_snapshot, "/ranking/diversity"), - "retrieval_sources": compact_pointer(&trace.config_snapshot, "/ranking/retrieval_sources"), - "override": compact_pointer(&trace.config_snapshot, "/ranking/override"), - }, - }) -} - -pub(super) fn compact_stage_movement(stages: &[SearchTrajectoryStage]) -> Vec { - stages - .iter() - .map(|stage| { - serde_json::json!({ - "stage_order": stage.stage_order, - "stage_name": stage.stage_name, - "item_count": stage.items.len(), - "stats": compact_pointer(&stage.stage_payload, "/stats"), - "decisions": compact_pointer(&stage.stage_payload, "/decisions"), - "filter_impact": compact_pointer(&stage.stage_payload, "/filter_impact"), - }) - }) - .collect() -} - -pub(super) fn compact_candidate_replay( - candidates: &[TraceReplayCandidate], - selected_candidate_keys: &BTreeSet<(Uuid, Uuid)>, - source_refs: &BTreeMap, -) -> Value { - let rerank_ranks = candidate_rerank_ranks(candidates); - let rows = candidates - .iter() - .map(|candidate| { - let key = recall_debug::candidate_identity(candidate.note_id, candidate.chunk_id); - let rerank_rank = rerank_ranks.get(&key).copied(); - let selection_state = - if selected_candidate_keys.contains(&key) { "selected" } else { "dropped" }; - let stage_reason = candidate_stage_reason(candidate, selection_state); - let source_ref = - source_refs.get(&candidate.note_id).map(|source| source.source_ref.clone()); - - serde_json::json!({ - "note_id": candidate.note_id, - "chunk_id": candidate.chunk_id, - "source_ref": source_ref, - "source_ref_available": source_ref.is_some(), - "retrieval_rank": candidate.retrieval_rank, - "rerank_rank": rerank_rank, - "rerank_delta": rerank_rank.map(|rank| candidate.retrieval_rank as i64 - i64::from(rank)), - "rerank_score": candidate.rerank_score, - "retrieval_score": candidate.retrieval_score, - "selection_state": selection_state, - "stage_reason": stage_reason, - "policy_reason": stage_reason, - "note_scope": candidate.note_scope, - "diversity_selected": candidate.diversity_selected, - "diversity_skipped_reason": candidate.diversity_skipped_reason, - }) - }) - .collect::>(); - let selected_count = rows - .iter() - .filter(|row| row.get("selection_state").and_then(Value::as_str) == Some("selected")) - .count(); - - serde_json::json!({ - "candidate_count": candidates.len(), - "selected_count": selected_count, - "dropped_count": rows.len().saturating_sub(selected_count), - "rows": rows, - }) -} - -pub(super) fn candidate_rerank_ranks( - candidates: &[TraceReplayCandidate], -) -> BTreeMap<(Uuid, Uuid), u32> { - let mut ordered = candidates.iter().collect::>(); - - ordered.sort_by(|a, b| { - b.rerank_score - .total_cmp(&a.rerank_score) - .then_with(|| a.retrieval_rank.cmp(&b.retrieval_rank)) - .then_with(|| a.note_id.cmp(&b.note_id)) - .then_with(|| a.chunk_id.cmp(&b.chunk_id)) - }); - - ordered - .into_iter() - .enumerate() - .map(|(index, candidate)| { - ( - recall_debug::candidate_identity(candidate.note_id, candidate.chunk_id), - index as u32 + 1, - ) - }) - .collect() -} - -pub(super) fn candidate_stage_reason( - candidate: &TraceReplayCandidate, - selection_state: &str, -) -> String { - if selection_state == "selected" { - candidate.diversity_selected_reason.clone().unwrap_or_else(|| "selection.final".to_string()) - } else { - candidate - .diversity_skipped_reason - .clone() - .unwrap_or_else(|| "not_in_final_top_k".to_string()) - } -} - -pub(super) fn compact_selected_context( - selected_items: &[&SearchExplainItem], - source_refs: &BTreeMap, -) -> Vec { - selected_items - .iter() - .map(|item| { - let source = source_refs.get(&item.note_id); - - serde_json::json!({ - "result_handle": item.result_handle, - "note_id": item.note_id, - "chunk_id": item.chunk_id, - "source_ref": source.map(|row| row.source_ref.clone()), - "source_ref_available": source.is_some(), - "freshness_state": recall_debug::freshness_from_note_source(source), - "final_rank": item.rank, - "final_score": item.explain.ranking.final_score, - "policy_id": item.explain.ranking.policy_id, - "policy_reason": "final ranked search result", - "ranking_terms": item - .explain - .ranking - .terms - .iter() - .map(|term| serde_json::json!({ - "name": term.name, - "value": term.value, - })) - .collect::>(), - "relation_context_count": item - .explain - .relation_context - .as_ref() - .map(Vec::len) - .unwrap_or_default(), - }) - }) - .collect() -} - -pub(super) fn compact_pointer(value: &Value, pointer: &str) -> Value { - value.pointer(pointer).cloned().unwrap_or(Value::Null) -} diff --git a/packages/elf-service/src/recall_debug/replay/artifact.rs b/packages/elf-service/src/recall_debug/replay/artifact.rs new file mode 100644 index 00000000..16d533a9 --- /dev/null +++ b/packages/elf-service/src/recall_debug/replay/artifact.rs @@ -0,0 +1,31 @@ +use crate::recall_debug::{ + BTreeMap, BTreeSet, NoteDebugSourceRow, SearchExplainItem, SearchTrace, SearchTrajectoryStage, + TraceReplayCandidate, Uuid, Value, + replay::{candidate_replay, controls, selected_context}, +}; + +pub(in crate::recall_debug) fn memory_compact_replay_artifact( + trace: &SearchTrace, + stages: &[SearchTrajectoryStage], + candidates: &[TraceReplayCandidate], + selected_items: &[&SearchExplainItem], + selected_candidate_keys: &BTreeSet<(Uuid, Uuid)>, + source_refs: &BTreeMap, + replay_command: &str, +) -> Value { + serde_json::json!({ + "schema": "elf.recall_debug.compact_replay/v1", + "trace_id": trace.trace_id, + "query": trace.query, + "replay_command": replay_command, + "controls": controls::compact_replay_controls(trace), + "stage_movement": controls::compact_stage_movement(stages), + "candidate_replay": candidate_replay::compact_candidate_replay(candidates, selected_candidate_keys, source_refs), + "selected_context": selected_context::compact_selected_context(selected_items, source_refs), + "authority": { + "source_refs_visible": true, + "policy_reasons_visible": true, + "raw_sql_needed": false, + }, + }) +} diff --git a/packages/elf-service/src/recall_debug/replay/candidate_replay.rs b/packages/elf-service/src/recall_debug/replay/candidate_replay.rs new file mode 100644 index 00000000..fd9a699e --- /dev/null +++ b/packages/elf-service/src/recall_debug/replay/candidate_replay.rs @@ -0,0 +1,86 @@ +use crate::recall_debug::{ + self, BTreeMap, BTreeSet, NoteDebugSourceRow, TraceReplayCandidate, Uuid, Value, +}; + +pub(in crate::recall_debug::replay) fn compact_candidate_replay( + candidates: &[TraceReplayCandidate], + selected_candidate_keys: &BTreeSet<(Uuid, Uuid)>, + source_refs: &BTreeMap, +) -> Value { + let rerank_ranks = candidate_rerank_ranks(candidates); + let rows = candidates + .iter() + .map(|candidate| { + let key = recall_debug::candidate_identity(candidate.note_id, candidate.chunk_id); + let rerank_rank = rerank_ranks.get(&key).copied(); + let selection_state = + if selected_candidate_keys.contains(&key) { "selected" } else { "dropped" }; + let stage_reason = candidate_stage_reason(candidate, selection_state); + let source_ref = + source_refs.get(&candidate.note_id).map(|source| source.source_ref.clone()); + + serde_json::json!({ + "note_id": candidate.note_id, + "chunk_id": candidate.chunk_id, + "source_ref": source_ref, + "source_ref_available": source_ref.is_some(), + "retrieval_rank": candidate.retrieval_rank, + "rerank_rank": rerank_rank, + "rerank_delta": rerank_rank.map(|rank| candidate.retrieval_rank as i64 - i64::from(rank)), + "rerank_score": candidate.rerank_score, + "retrieval_score": candidate.retrieval_score, + "selection_state": selection_state, + "stage_reason": stage_reason, + "policy_reason": stage_reason, + "note_scope": candidate.note_scope, + "diversity_selected": candidate.diversity_selected, + "diversity_skipped_reason": candidate.diversity_skipped_reason, + }) + }) + .collect::>(); + let selected_count = rows + .iter() + .filter(|row| row.get("selection_state").and_then(Value::as_str) == Some("selected")) + .count(); + + serde_json::json!({ + "candidate_count": candidates.len(), + "selected_count": selected_count, + "dropped_count": rows.len().saturating_sub(selected_count), + "rows": rows, + }) +} + +fn candidate_rerank_ranks(candidates: &[TraceReplayCandidate]) -> BTreeMap<(Uuid, Uuid), u32> { + let mut ordered = candidates.iter().collect::>(); + + ordered.sort_by(|a, b| { + b.rerank_score + .total_cmp(&a.rerank_score) + .then_with(|| a.retrieval_rank.cmp(&b.retrieval_rank)) + .then_with(|| a.note_id.cmp(&b.note_id)) + .then_with(|| a.chunk_id.cmp(&b.chunk_id)) + }); + + ordered + .into_iter() + .enumerate() + .map(|(index, candidate)| { + ( + recall_debug::candidate_identity(candidate.note_id, candidate.chunk_id), + index as u32 + 1, + ) + }) + .collect() +} + +fn candidate_stage_reason(candidate: &TraceReplayCandidate, selection_state: &str) -> String { + if selection_state == "selected" { + candidate.diversity_selected_reason.clone().unwrap_or_else(|| "selection.final".to_string()) + } else { + candidate + .diversity_skipped_reason + .clone() + .unwrap_or_else(|| "not_in_final_top_k".to_string()) + } +} diff --git a/packages/elf-service/src/recall_debug/replay/candidate_row.rs b/packages/elf-service/src/recall_debug/replay/candidate_row.rs new file mode 100644 index 00000000..0c2b3423 --- /dev/null +++ b/packages/elf-service/src/recall_debug/replay/candidate_row.rs @@ -0,0 +1,50 @@ +use crate::recall_debug::{self, NoteDebugSourceRow, RecallDebugRow, TraceReplayCandidate, Uuid}; + +pub(in crate::recall_debug) fn candidate_debug_row( + trace_id: Uuid, + candidate: &TraceReplayCandidate, + source: Option<&NoteDebugSourceRow>, + replay_command: &str, +) -> RecallDebugRow { + let selected_by_diversity = candidate.diversity_selected.unwrap_or(false); + let skipped_reason = candidate.diversity_skipped_reason.clone().or_else(|| { + if selected_by_diversity { + candidate.diversity_selected_reason.clone() + } else { + Some("not_in_final_top_k".to_string()) + } + }); + + RecallDebugRow { + layer: "memory_notes".to_string(), + item_ref: serde_json::json!({ + "trace_id": trace_id, + "note_id": candidate.note_id, + "chunk_id": candidate.chunk_id, + "chunk_index": candidate.chunk_index, + }), + selection_state: "dropped".to_string(), + authority_layer: "memory_note".to_string(), + freshness_state: recall_debug::freshness_from_note_source(source), + source_refs: recall_debug::source_ref_from_note_source(source), + score: candidate.retrieval_score, + rank: Some(candidate.retrieval_rank), + rationale: Some( + "candidate captured for replay but not selected in final result set".to_string(), + ), + stage_reason: skipped_reason, + replay_command: Some(replay_command.to_string()), + evidence_class: "pass".to_string(), + debug_artifacts: serde_json::json!({ + "snippet": candidate.snippet, + "rerank_score": candidate.rerank_score, + "note_scope": candidate.note_scope, + "diversity_selected": candidate.diversity_selected, + "diversity_selected_rank": candidate.diversity_selected_rank, + "diversity_nearest_selected_note_id": candidate.diversity_nearest_selected_note_id, + "diversity_similarity": candidate.diversity_similarity, + "diversity_mmr_score": candidate.diversity_mmr_score, + "diversity_missing_embedding": candidate.diversity_missing_embedding, + }), + } +} diff --git a/packages/elf-service/src/recall_debug/replay/controls.rs b/packages/elf-service/src/recall_debug/replay/controls.rs new file mode 100644 index 00000000..cce1b26e --- /dev/null +++ b/packages/elf-service/src/recall_debug/replay/controls.rs @@ -0,0 +1,42 @@ +use crate::recall_debug::{SearchTrace, SearchTrajectoryStage, Value}; + +pub(in crate::recall_debug::replay) fn compact_replay_controls(trace: &SearchTrace) -> Value { + serde_json::json!({ + "top_k": trace.top_k, + "candidate_count": trace.candidate_count, + "expansion_mode": trace.expansion_mode, + "expanded_query_count": trace.expanded_queries.len(), + "expanded_queries": trace.expanded_queries, + "allowed_scopes": trace.allowed_scopes, + "search": compact_pointer(&trace.config_snapshot, "/search"), + "ranking": { + "policy_id": compact_pointer(&trace.config_snapshot, "/ranking/policy_id"), + "blend": compact_pointer(&trace.config_snapshot, "/ranking/blend"), + "diversity": compact_pointer(&trace.config_snapshot, "/ranking/diversity"), + "retrieval_sources": compact_pointer(&trace.config_snapshot, "/ranking/retrieval_sources"), + "override": compact_pointer(&trace.config_snapshot, "/ranking/override"), + }, + }) +} + +pub(in crate::recall_debug::replay) fn compact_stage_movement( + stages: &[SearchTrajectoryStage], +) -> Vec { + stages + .iter() + .map(|stage| { + serde_json::json!({ + "stage_order": stage.stage_order, + "stage_name": stage.stage_name, + "item_count": stage.items.len(), + "stats": compact_pointer(&stage.stage_payload, "/stats"), + "decisions": compact_pointer(&stage.stage_payload, "/decisions"), + "filter_impact": compact_pointer(&stage.stage_payload, "/filter_impact"), + }) + }) + .collect() +} + +fn compact_pointer(value: &Value, pointer: &str) -> Value { + value.pointer(pointer).cloned().unwrap_or(Value::Null) +} diff --git a/packages/elf-service/src/recall_debug/replay/selected_context.rs b/packages/elf-service/src/recall_debug/replay/selected_context.rs new file mode 100644 index 00000000..1bc015f5 --- /dev/null +++ b/packages/elf-service/src/recall_debug/replay/selected_context.rs @@ -0,0 +1,42 @@ +use crate::recall_debug::{self, BTreeMap, NoteDebugSourceRow, SearchExplainItem, Uuid, Value}; + +pub(in crate::recall_debug::replay) fn compact_selected_context( + selected_items: &[&SearchExplainItem], + source_refs: &BTreeMap, +) -> Vec { + selected_items + .iter() + .map(|item| { + let source = source_refs.get(&item.note_id); + + serde_json::json!({ + "result_handle": item.result_handle, + "note_id": item.note_id, + "chunk_id": item.chunk_id, + "source_ref": source.map(|row| row.source_ref.clone()), + "source_ref_available": source.is_some(), + "freshness_state": recall_debug::freshness_from_note_source(source), + "final_rank": item.rank, + "final_score": item.explain.ranking.final_score, + "policy_id": item.explain.ranking.policy_id, + "policy_reason": "final ranked search result", + "ranking_terms": item + .explain + .ranking + .terms + .iter() + .map(|term| serde_json::json!({ + "name": term.name, + "value": term.value, + })) + .collect::>(), + "relation_context_count": item + .explain + .relation_context + .as_ref() + .map(Vec::len) + .unwrap_or_default(), + }) + }) + .collect() +}