use crate::{AppState, DaemonReadiness};
use anyhow::{anyhow, Context, Result};
use serde_json::{json, Value};
use trusty_common::memory_core::palace::RoomType;
use trusty_common::memory_core::retrieval::{
recall_across_palaces, recall_deep_scoped, recall_scoped, scope_admits, PalaceHandle,
RecallScope,
};
use crate::service::recall_stream::recall_streamed;
use super::bm25::{bm25_hits_to_recall_results, bm25_search_optional, fuse_bm25_into_recall};
use super::helpers::open_palace_handle;
use super::palace_index::{resolve_palace_or_index, PalaceScope};
use super::recall_projection::{
apply_score_floor, candidate_window, include_creator_tags_arg, min_score_arg, project_tags,
serialize_recall, RecallProjection,
};
use super::wing_ops::resolve_wing_arg;
fn vector_lane_available(state: &AppState) -> bool {
state.readiness() != DaemonReadiness::Warming
|| trusty_common::memory_core::retrieval::shared_embedder_initialized()
}
async fn recall_without_embedder(
state: &AppState,
handle: &trusty_common::memory_core::retrieval::PalaceHandle,
query: &str,
scope: &RecallScope,
top_k: usize,
) -> Vec<trusty_common::memory_core::retrieval::RecallResult> {
let palace = handle.id.as_str();
let mut results = trusty_common::memory_core::retrieval::retrieve_l0_l1(handle);
let allowed = scope.allowed_room_ids(&handle.kg);
if let Some(bm25_hits) = bm25_search_optional(state, palace, query, top_k).await {
for hydrated in bm25_hits_to_recall_results(handle, &bm25_hits) {
if !scope_admits(&allowed, hydrated.drawer.room_id) {
continue;
}
if !results.iter().any(|r| r.drawer.id == hydrated.drawer.id) {
results.push(hydrated);
}
}
}
results.sort_by(|a, b| {
b.score
.partial_cmp(&a.score)
.unwrap_or(std::cmp::Ordering::Equal)
});
results.truncate(top_k);
results
}
fn recall_room_filter(args: &Value) -> Option<RoomType> {
args.get("room")
.and_then(|v| v.as_str())
.filter(|s| !s.trim().is_empty())
.map(RoomType::parse)
}
pub(crate) fn recall_scope(handle: &PalaceHandle, args: &Value, tool: &str) -> Result<RecallScope> {
let room = recall_room_filter(args);
match resolve_wing_arg(handle, args, tool)? {
Some(wing_id) => {
if room.is_some() {
return Err(anyhow!(
"{tool}: 'wing' and 'room' together are not supported yet — \
pass one or the other"
));
}
Ok(RecallScope::Wing(wing_id))
}
None => Ok(RecallScope::from_room_filter(room)),
}
}
pub(crate) async fn handle_memory_recall(state: &AppState, args: Value) -> Result<Value> {
let palace = match resolve_palace_or_index(state, &args, "memory_recall").await? {
PalaceScope::Palace(p) => p,
PalaceScope::Index(index) => return Ok(index),
};
let query = args
.get("query")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow!("memory_recall: missing 'query'"))?;
let top_k = args.get("top_k").and_then(|v| v.as_u64()).unwrap_or(10) as usize;
let include_creator_tags = include_creator_tags_arg(&args);
let min_score = min_score_arg(&args, "memory_recall")?;
let fetch_k = candidate_window(top_k, min_score);
let handle = open_palace_handle(state, &palace)?;
let scope = recall_scope(&handle, &args, "memory_recall")?;
if !vector_lane_available(state) {
let mut results = recall_without_embedder(state, &handle, query, &scope, fetch_k).await;
let dropped_below_floor = apply_score_floor(&mut results, min_score, top_k);
return Ok(serialize_recall(
&palace,
query,
results,
&RecallProjection {
include_creator_tags,
dropped_below_floor,
},
));
}
let embedder = state.embedder().await?;
let vector_fut = recall_scoped(&handle, embedder.as_ref(), query, &scope, fetch_k);
let bm25_fut = bm25_search_optional(state, handle.id.as_str(), query, fetch_k);
let (vector_res, bm25_res) = tokio::join!(vector_fut, bm25_fut);
let mut results = vector_res.context("recall")?;
if let Some(bm25_hits) = bm25_res {
fuse_bm25_into_recall(&mut results, &bm25_hits, fetch_k);
}
let dropped_below_floor = apply_score_floor(&mut results, min_score, top_k);
Ok(serialize_recall(
&palace,
query,
results,
&RecallProjection {
include_creator_tags,
dropped_below_floor,
},
))
}
pub(crate) async fn handle_memory_recall_deep(state: &AppState, args: Value) -> Result<Value> {
let palace = match resolve_palace_or_index(state, &args, "memory_recall_deep").await? {
PalaceScope::Palace(p) => p,
PalaceScope::Index(index) => return Ok(index),
};
let query = args
.get("query")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow!("memory_recall_deep: missing 'query'"))?;
let top_k = args.get("top_k").and_then(|v| v.as_u64()).unwrap_or(10) as usize;
let include_creator_tags = include_creator_tags_arg(&args);
let min_score = min_score_arg(&args, "memory_recall_deep")?;
let fetch_k = candidate_window(top_k, min_score);
let handle = open_palace_handle(state, &palace)?;
let scope = recall_scope(&handle, &args, "memory_recall_deep")?;
if !vector_lane_available(state) {
let mut results = recall_without_embedder(state, &handle, query, &scope, fetch_k).await;
let dropped_below_floor = apply_score_floor(&mut results, min_score, top_k);
return Ok(serialize_recall(
&palace,
query,
results,
&RecallProjection {
include_creator_tags,
dropped_below_floor,
},
));
}
let embedder = state.embedder().await?;
let vector_fut = recall_deep_scoped(&handle, embedder.as_ref(), query, &scope, fetch_k);
let bm25_fut = bm25_search_optional(state, handle.id.as_str(), query, fetch_k);
let (vector_res, bm25_res) = tokio::join!(vector_fut, bm25_fut);
let mut results = vector_res.context("recall_deep")?;
if let Some(bm25_hits) = bm25_res {
fuse_bm25_into_recall(&mut results, &bm25_hits, fetch_k);
}
let dropped_below_floor = apply_score_floor(&mut results, min_score, top_k);
Ok(serialize_recall(
&palace,
query,
results,
&RecallProjection {
include_creator_tags,
dropped_below_floor,
},
))
}
async fn recall_all_without_embedder(
state: &AppState,
handles: &[std::sync::Arc<trusty_common::memory_core::retrieval::PalaceHandle>],
query: &str,
top_k: usize,
) -> Vec<trusty_common::memory_core::retrieval::CrossPalaceResult> {
let mut merged = Vec::new();
for handle in handles {
let palace_id = handle.id.as_str().to_string();
let hits = recall_without_embedder(state, handle, query, &RecallScope::All, top_k).await;
merged.extend(hits.into_iter().map(|result| {
trusty_common::memory_core::retrieval::CrossPalaceResult {
palace_id: palace_id.clone(),
result,
}
}));
}
merged.sort_by(|a, b| {
b.result
.score
.partial_cmp(&a.result.score)
.unwrap_or(std::cmp::Ordering::Equal)
});
merged.truncate(top_k);
merged
}
pub(crate) async fn handle_memory_recall_all(state: &AppState, args: Value) -> Result<Value> {
let query = args
.get("q")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow!("memory_recall_all: missing 'q'"))?;
let top_k = args.get("top_k").and_then(|v| v.as_u64()).unwrap_or(10) as usize;
let deep = args.get("deep").and_then(|v| v.as_bool()).unwrap_or(false);
let include_creator_tags = include_creator_tags_arg(&args);
let palaces = crate::service::helpers::list_palaces_blocking(state).await?;
let results = if !vector_lane_available(state) {
recall_streamed(
state,
&palaces,
"memory_recall_all",
top_k,
|handles| async move {
Ok(recall_all_without_embedder(state, &handles, query, top_k).await)
},
)
.await?
} else {
let embedder = state.embedder().await?;
recall_streamed(state, &palaces, "memory_recall_all", top_k, |handles| {
let embedder = embedder.clone();
async move { recall_across_palaces(&handles, &embedder, query, top_k, deep).await }
})
.await
.context("recall_across_palaces")?
};
let payload: Vec<Value> = results
.iter()
.map(|r| {
json!({
"palace_id": r.palace_id,
"drawer_id": r.result.drawer.id.to_string(),
"content": r.result.drawer.content(),
"importance": r.result.drawer.importance,
"tags": project_tags(&r.result.drawer.tags, include_creator_tags),
"score": r.result.score,
"layer": r.result.layer,
"drawer_type": r.result.drawer.drawer_type.as_str(),
})
})
.collect();
Ok(json!({ "query": query, "results": payload }))
}