#![forbid(unsafe_code)]
use async_trait::async_trait;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::fmt::Write;
use std::sync::Arc;
use wm_core::{Context, EffectRow, Galaxy, Gana, Resource, Tool, ToolStats};
use wm_memory::{
AssociationStore, MemoryStore, RecallEngine, SearchEngine, episodic::detect_conflicts,
};
use super::common::{
bool_prop, galaxy_name, galaxy_search_arg, int_prop, num_prop, parse_galaxy, parse_galaxy_or,
schema, str_prop,
};
fn fused_mode_label(vector_weight: f32) -> &'static str {
if vector_weight > 0.0 {
"hybrid"
} else {
"bm25"
}
}
fn with_navigation_disclosure(mut result: serde_json::Value, original: &str) -> serde_json::Value {
let limit = wm_memory::search::MAX_INDEX_CONTENT_LEN;
let scrubbed = result
.get("content")
.and_then(serde_json::Value::as_str)
.is_none_or(|navigation| navigation != original);
if let Some(obj) = result.as_object_mut() {
obj.insert(
"content_representation".into(),
json!("scrubbed_navigation"),
);
obj.insert("content_character_limit".into(), json!(limit));
obj.insert(
"content_truncated".into(),
json!(original.chars().nth(limit).is_some()),
);
obj.insert("content_scrubbed".into(), json!(scrubbed));
obj.insert("exact_read_available".into(), json!(true));
if let Some(pattern) = wm_memory::detect_injection(original) {
obj.insert("instruction_shaped".into(), json!(true));
obj.insert("instruction_pattern".into(), json!(pattern));
}
}
result
}
fn abstention_floor() -> f32 {
std::env::var("WM_RECALL_ABSTENTION_FLOOR")
.ok()
.and_then(|value| value.trim().parse::<f32>().ok())
.filter(|value| *value > 0.0)
.unwrap_or(0.0)
}
fn abstention_coverage() -> f32 {
std::env::var("WM_RECALL_ABSTENTION_COVERAGE")
.ok()
.and_then(|value| value.trim().parse::<f32>().ok())
.filter(|value| (0.0..=1.0).contains(value) && *value > 0.0)
.unwrap_or(0.0)
}
fn weak_evidence_abstention(
results: &[serde_json::Value],
query: &str,
) -> Option<serde_json::Value> {
weak_evidence_abstention_with(results, query, abstention_floor(), abstention_coverage())
}
fn weak_evidence_abstention_with(
results: &[serde_json::Value],
query: &str,
floor: f32,
coverage_floor: f32,
) -> Option<serde_json::Value> {
if floor <= 0.0 && coverage_floor <= 0.0 {
return None;
}
let top = results.first()?;
let source = top
.get("source")
.and_then(serde_json::Value::as_str)
.unwrap_or("");
match source {
"hybrid" | "bm25" | "fts" => {
let raw = top
.get("raw_score")
.and_then(serde_json::Value::as_f64)
.unwrap_or(0.0);
if floor > 0.0 && raw > 0.0 && (raw as f32) < floor {
return Some(json!({
"status": "insufficient_evidence",
"reason": "top_below_floor",
"scope": "retrieval",
"signal": "bm25",
"top_score": raw,
"floor": floor,
}));
}
}
"episodic" => {
let matched = top
.get("matched_terms")
.and_then(serde_json::Value::as_u64)
.unwrap_or(0);
let query_terms = wm_memory::strip_stopwords(query).split_whitespace().count() as u64;
if coverage_floor > 0.0 && query_terms > 0 {
let coverage = matched as f64 / query_terms as f64;
if coverage < f64::from(coverage_floor) {
return Some(json!({
"status": "insufficient_evidence",
"reason": "coverage_below_floor",
"scope": "retrieval",
"signal": "coverage",
"coverage": coverage,
"matched_terms": matched,
"query_terms": query_terms,
"floor": coverage_floor,
}));
}
}
}
_ => {}
}
None
}
fn build_evidence_bundle(store: &MemoryStore, results: &[serde_json::Value]) -> serde_json::Value {
let entries: Vec<serde_json::Value> = results
.iter()
.map(|r| {
let id = r
.get("id")
.and_then(serde_json::Value::as_str)
.unwrap_or_default()
.to_string();
let galaxy = r
.get("galaxy")
.and_then(serde_json::Value::as_str)
.unwrap_or_default()
.to_string();
let resolved = uuid::Uuid::parse_str(&id)
.ok()
.and_then(|key| resolve_memory_across_galaxies(store, key));
let (source_time, visibility, history) = match &resolved {
Some((_, mem)) => {
let revisions = store
.revisions(mem.metadata.galaxy, mem.metadata.id)
.unwrap_or_default();
let chain_valid =
wm_memory::revision::verify_chain(&revisions, &mem.metadata.content_hash)
.valid;
(
json!({
"created_at": mem.metadata.created_at,
"basis": "recorded_at",
"event_time": serde_json::Value::Null,
"event_time_basis": "not_tracked",
}),
json!({
"private": mem.metadata.is_private,
"model_exclude": mem.metadata.model_exclude,
}),
json!({
"revision_count": revisions.len(),
"superseded": !revisions.is_empty(),
"chain_valid": chain_valid,
"current": true,
}),
)
}
None => (
json!({
"created_at": serde_json::Value::Null,
"basis": "unavailable_cold_record",
"event_time": serde_json::Value::Null,
"event_time_basis": "not_tracked",
}),
json!({
"private": false,
"model_exclude": !r
.get("model_visible")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false),
}),
json!({
"revision_count": 0,
"superseded": false,
"chain_valid": serde_json::Value::Null,
"current": true,
"basis": "unavailable_cold_record",
}),
),
};
let mut retrieval = json!({
"route": r.get("source").cloned().unwrap_or(serde_json::Value::Null),
"score": r.get("score").cloned().unwrap_or(serde_json::Value::Null),
});
if let Some(terms) = r.get("matched_terms") {
retrieval["matched_terms"] = terms.clone();
}
if let Some(via) = r.get("via") {
retrieval["via"] = via.clone();
}
json!({
"id": id,
"galaxy": galaxy,
"retrieval": retrieval,
"source_time": source_time,
"history": history,
"integrity": r
.get("integrity")
.cloned()
.unwrap_or_else(|| json!("source_read")),
"visibility": visibility,
"coverage": {
"representation": r
.get("content_representation")
.cloned()
.unwrap_or(serde_json::Value::Null),
"truncated": r
.get("content_truncated")
.cloned()
.unwrap_or(serde_json::Value::Null),
"exact_read_available": r
.get("exact_read_available")
.cloned()
.unwrap_or(json!(false)),
},
})
})
.collect();
let shims: Vec<wm_memory::episodic::EpisodicSearchResult> = results
.iter()
.filter_map(|r| {
let id = r.get("id").and_then(serde_json::Value::as_str)?;
let key = uuid::Uuid::parse_str(id).ok()?;
let record = store.episodic().get(key).ok().flatten()?;
Some(wm_memory::episodic::EpisodicSearchResult {
record,
score: 0.0,
matched_terms: 0,
})
})
.collect();
let conflicts = detect_conflicts(&shims);
let pairs: Vec<serde_json::Value> = conflicts
.iter()
.map(|c| {
json!({
"later": c.later_record.to_string(),
"earlier": c.earlier_record.to_string(),
"marker": c.marker,
"shared_terms": c.shared_terms,
})
})
.collect();
json!({
"version": "v0",
"count": entries.len(),
"entries": entries,
"conflicts": {"count": pairs.len(), "pairs": pairs},
})
}
fn resolve_memory_across_galaxies(
store: &MemoryStore,
id: uuid::Uuid,
) -> Option<(wm_core::Galaxy, wm_memory::Memory)> {
for galaxy in wm_core::Galaxy::memory_galaxies() {
if let Ok(Some(mem)) = store.get(galaxy, id) {
return Some((galaxy, mem));
}
}
None
}
pub struct MemoryConsolidateTool {
store: Arc<MemoryStore>,
search: Option<Arc<SearchEngine>>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryConsolidateTool {
pub fn new(store: Arc<MemoryStore>, search: Option<Arc<SearchEngine>>) -> Self {
Self {
store,
search,
stats: ToolStats::default(),
effects: EffectRow {
writes: super::common::memory_galaxy_writes(),
reads: super::common::memory_galaxy_reads(),
destructive: true,
..Default::default()
},
}
}
}
#[async_trait]
impl Tool for MemoryConsolidateTool {
fn input_schema(&self) -> Value {
schema(
&json!({
"galaxy": super::common::str_prop("Galaxy to consolidate (optional; default codex)"),
}),
&[],
)
}
fn name(&self) -> &str {
"memory.consolidate"
}
fn gana(&self) -> Gana {
Gana::Encampment
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Deduplicate memories by content_hash within a galaxy"
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = args
.get("galaxy")
.and_then(|v| v.as_str())
.unwrap_or("codex");
let galaxy = parse_galaxy(galaxy)?;
let memories = self.store.scan_all(galaxy)?;
let mut seen_hashes: HashMap<String, uuid::Uuid> = HashMap::new();
let mut duplicates = 0u32;
for mem in &memories {
let hash = &mem.metadata.content_hash;
if let Some(existing_id) = seen_hashes.get(hash) {
if *existing_id != mem.metadata.id {
self.store.delete(galaxy, mem.metadata.id)?;
super::common::deindex(self.search.as_deref(), &mem.metadata.id.to_string());
duplicates += 1;
}
} else {
seen_hashes.insert(hash.clone(), mem.metadata.id);
}
}
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"scanned": memories.len(),
"duplicates_removed": duplicates,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryDecayTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryDecayTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow {
writes: super::common::memory_galaxy_writes(),
reads: super::common::memory_galaxy_reads(),
..Default::default()
},
}
}
}
#[async_trait]
impl Tool for MemoryDecayTool {
fn input_schema(&self) -> Value {
schema(
&json!({
"galaxy": super::common::str_prop("Galaxy to decay (optional; default codex)"),
"importance_threshold": super::common::num_prop("Decay memories below this importance (0-1)"),
"decay_factor": super::common::num_prop("Multiplier applied to importance (0-1)"),
}),
&[],
)
}
fn name(&self) -> &str {
"memory.decay"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Lower importance of old, low-access memories (never deletes)"
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = args
.get("galaxy")
.and_then(|v| v.as_str())
.unwrap_or("codex");
let galaxy = parse_galaxy(galaxy)?;
let threshold = args
.get("importance_threshold")
.and_then(serde_json::Value::as_f64)
.unwrap_or(0.3) as f32;
let decay_factor = args
.get("decay_factor")
.and_then(serde_json::Value::as_f64)
.unwrap_or(0.9) as f32;
let memories = self.store.scan(galaxy, 10_000)?;
let mut decayed = 0u32;
for mem in &memories {
if mem.metadata.importance < threshold {
let mut updated = mem.clone();
updated.metadata.importance =
(updated.metadata.importance * decay_factor).clamp(0.0, 1.0);
if (updated.metadata.importance - mem.metadata.importance).abs() > 0.001 {
self.store.put(galaxy, &updated)?;
decayed += 1;
}
}
}
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"scanned": memories.len(),
"decayed": decayed,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryBatchReadTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryBatchReadTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow::read_only(vec![Resource::Galaxy("codex".into())]),
}
}
}
#[async_trait]
impl Tool for MemoryBatchReadTool {
fn name(&self) -> &str {
"memory.batch_read"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Read multiple memories by ID from a galaxy"
}
fn input_schema(&self) -> Value {
super::common::schema(
&json!({
"ids": super::common::str_array_prop("Memory UUIDs to read"),
"galaxy": super::common::str_prop("Galaxy (default: codex)"),
}),
&["ids"],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = args
.get("galaxy")
.and_then(|v| v.as_str())
.unwrap_or("codex");
let galaxy = parse_galaxy(galaxy)?;
let ids = args
.get("ids")
.and_then(|v| v.as_array())
.ok_or_else(|| wm_core::CoreError::InvalidArgs("Missing 'ids' array".into()))?;
let mut results = Vec::new();
let mut misses = 0u32;
for id_val in ids {
if let Some(id_str) = id_val.as_str() {
if let Ok(id) = uuid::Uuid::parse_str(id_str) {
match self.store.get(galaxy, id)? {
Some(mem)
if crate::expansion::common::mcp_visible(&mem)
&& crate::expansion::common::validity_visible(&mem) =>
{
results.push(json!({
"id": mem.metadata.id,
"content": mem.content,
"tags": mem.metadata.tags,
"importance": mem.metadata.importance,
}));
}
Some(_) => {
misses += 1;
}
None => {
misses += 1;
}
}
}
}
}
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"found": results.len(),
"misses": misses,
"memories": results,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryUpdateTool {
store: Arc<MemoryStore>,
search: Option<Arc<SearchEngine>>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryUpdateTool {
pub fn new(store: Arc<MemoryStore>, search: Option<Arc<SearchEngine>>) -> Self {
Self {
store,
search,
stats: ToolStats::default(),
effects: EffectRow {
writes: super::common::memory_galaxy_writes(),
reads: super::common::memory_galaxy_reads(),
sandbox: wm_core::Sandbox::StoreScoped,
..Default::default()
},
}
}
}
#[async_trait]
impl Tool for MemoryUpdateTool {
fn name(&self) -> &str {
"memory.update"
}
fn gana(&self) -> Gana {
Gana::Encampment
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Update tags, importance, title/topic, or content of an existing memory"
}
fn input_schema(&self) -> Value {
super::common::schema(
&json!({
"id": super::common::str_prop("Memory UUID to update"),
"content": super::common::str_prop("New content (optional)"),
"tags": super::common::str_array_prop("Replacement tags (optional)"),
"importance": super::common::bounded_num_prop("New importance 0.0-1.0 (optional)", 0.0, 1.0),
"title": super::common::str_prop("New title (optional)"),
"topic": super::common::str_prop("New topic label (optional)"),
"galaxy": super::common::str_prop("Galaxy (default: codex)"),
}),
&["id"],
)
}
async fn call(&self, ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = parse_galaxy_or(args.get("galaxy").and_then(|v| v.as_str()), Galaxy::Codex)?;
let id_str = args
.get("id")
.and_then(|v| v.as_str())
.ok_or_else(|| wm_core::CoreError::InvalidArgs("Missing 'id'".into()))?;
let id = uuid::Uuid::parse_str(id_str)
.map_err(|e| wm_core::CoreError::InvalidArgs(format!("Invalid UUID: {e}")))?;
if let Some(search) = &self.search {
if search.is_readonly() {
return Err(wm_core::CoreError::InvalidArgs(
"read-only mode: memory.update disabled (another process owns the index)"
.into(),
));
}
}
let mut mem = self.store.get(galaxy, id)?.ok_or_else(|| {
wm_core::CoreError::NotFound(format!(
"Memory {id} not found in {}",
galaxy_name(galaxy)
))
})?;
let previous_hash = mem.metadata.content_hash.clone();
let content_changed = args.get("content").and_then(|v| v.as_str()).is_some();
if let Some(tags) = args.get("tags").and_then(|v| v.as_array()) {
mem.metadata.tags = tags
.iter()
.filter_map(|t| t.as_str().map(String::from))
.collect();
}
if args.get("importance").is_some() && !args["importance"].is_null() {
let importance =
wm_dispatch::write_gate::parse_importance_value(args.get("importance"))
.map_err(wm_core::CoreError::InvalidArgs)?;
if let Some(importance) = importance {
mem.metadata.importance = importance;
}
}
if let Some(title) = args.get("title") {
mem.metadata.title = title
.as_str()
.map(str::trim)
.filter(|s| !s.is_empty())
.map(String::from);
}
if let Some(topic) = args.get("topic") {
mem.metadata.topic = topic
.as_str()
.map(str::trim)
.filter(|s| !s.is_empty())
.map(String::from);
}
if let Some(content) = args.get("content").and_then(|v| v.as_str()) {
mem.content = content.to_string();
mem.metadata.content_hash = wm_memory::content_hash(content);
mem.metadata.revision_count = mem.metadata.revision_count.saturating_add(1);
}
self.store.put(galaxy, &mem)?;
let mut revision_disclosure = None;
if content_changed {
let actor = wm_memory::RevisionActor {
session: ctx.session_id.map(|sid| sid.to_string()),
user: ctx.user_id.clone(),
compartment: ctx.compartment.clone(),
};
match self.store.record_revision(
galaxy,
id,
&previous_hash,
&mem.metadata.content_hash,
actor,
) {
Ok(rev) => {
revision_disclosure = Some(serde_json::json!({
"seq": rev.seq,
"old_hash": rev.old_hash,
"new_hash": rev.new_hash,
}));
}
Err(e) => {
tracing::warn!(error = %e, "revision chain record failed for {id_str}");
revision_disclosure = Some(serde_json::json!({"record_failed": e.to_string()}));
}
}
}
let cred_kinds = args
.get("content")
.and_then(serde_json::Value::as_str)
.map(wm_memory::credential_shaped_content)
.unwrap_or_default();
let instruction_pattern = wm_memory::detect_injection(&mem.content);
super::common::replace_memory_index(&self.store, self.search.as_deref(), &mem);
let mut response = json!({
"status": "success",
"id": mem.metadata.id,
"galaxy": galaxy_name(galaxy),
"tags": mem.metadata.tags,
"importance": mem.metadata.importance,
"content_hash": mem.metadata.content_hash,
});
if content_changed {
response["prev_content_hash"] = json!(previous_hash);
}
if let Some(rev) = revision_disclosure {
response["revision"] = rev;
}
if !cred_kinds.is_empty() || instruction_pattern.is_some() {
let mut warnings: Vec<String> = cred_kinds
.iter()
.map(|k| {
format!(
"content looks like a credential ({k}) — {}",
wm_memory::CREDENTIAL_ADVICE
)
})
.collect();
if let Some(pattern) = instruction_pattern {
warnings.push(format!(
"content contains an instruction-shaped pattern ({pattern}) — stored as data; \
review it before trusting it as context"
));
}
response["warnings"] = json!(warnings);
}
Ok(response)
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryRevisionsTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryRevisionsTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow {
reads: super::common::memory_galaxy_reads(),
..Default::default()
},
}
}
}
#[async_trait]
impl Tool for MemoryRevisionsTool {
fn name(&self) -> &str {
"memory.revisions"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"List or verify a memory's content revision chain (tamper evidence)"
}
fn input_schema(&self) -> Value {
super::common::schema(
&json!({
"id": super::common::str_prop("Memory UUID to inspect"),
"action": super::common::str_prop("Action: list (default) | verify"),
"galaxy": super::common::str_prop("Galaxy (default: codex)"),
}),
&["id"],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = parse_galaxy_or(args.get("galaxy").and_then(|v| v.as_str()), Galaxy::Codex)?;
let id_str = args
.get("id")
.and_then(|v| v.as_str())
.ok_or_else(|| wm_core::CoreError::InvalidArgs("Missing 'id'".into()))?;
let id = uuid::Uuid::parse_str(id_str)
.map_err(|e| wm_core::CoreError::InvalidArgs(format!("Invalid UUID: {e}")))?;
let action = args
.get("action")
.and_then(|v| v.as_str())
.unwrap_or("list");
let revisions = self.store.revisions(galaxy, id)?;
match action {
"verify" => {
let mem = self.store.get(galaxy, id)?.ok_or_else(|| {
wm_core::CoreError::NotFound(format!(
"Memory {id} not found in {}",
galaxy_name(galaxy)
))
})?;
let report =
self.store
.verify_revision_chain(galaxy, id, &mem.metadata.content_hash)?;
Ok(json!({
"status": "success",
"id": id,
"galaxy": galaxy_name(galaxy),
"action": "verify",
"valid": report.valid,
"entries": report.entries,
"matches_head": report.matches_head,
"breaks": report.breaks,
"note": if report.entries == 0 {
"no revisions recorded (never content-updated or pre-S11c)"
} else { "" },
}))
}
"list" => Ok(json!({
"status": "success",
"id": id,
"galaxy": galaxy_name(galaxy),
"action": "list",
"count": revisions.len(),
"revisions": revisions,
})),
other => Err(wm_core::CoreError::InvalidArgs(format!(
"Unknown action '{other}' (expected 'list' or 'verify')"
))),
}
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryTagTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryTagTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow {
writes: super::common::memory_galaxy_writes(),
reads: super::common::memory_galaxy_reads(),
..Default::default()
},
}
}
}
#[async_trait]
impl Tool for MemoryTagTool {
fn name(&self) -> &str {
"memory.tag"
}
fn gana(&self) -> Gana {
Gana::Net
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Add or remove tags from a memory"
}
fn input_schema(&self) -> Value {
super::common::schema(
&json!({
"id": super::common::str_prop("Memory UUID to tag"),
"tags": super::common::str_array_prop("Tags to apply"),
"action": super::common::str_prop("Action: add (default) | remove"),
"galaxy": super::common::str_prop("Galaxy (default: codex)"),
}),
&["id", "tags"],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = parse_galaxy_or(args.get("galaxy").and_then(|v| v.as_str()), Galaxy::Codex)?;
let id_str = args
.get("id")
.and_then(|v| v.as_str())
.ok_or_else(|| wm_core::CoreError::InvalidArgs("Missing 'id'".into()))?;
let id = uuid::Uuid::parse_str(id_str)
.map_err(|e| wm_core::CoreError::InvalidArgs(format!("Invalid UUID: {e}")))?;
let mut mem = self
.store
.get(galaxy, id)?
.ok_or_else(|| wm_core::CoreError::NotFound(format!("Memory {id} not found")))?;
let action = args.get("action").and_then(|v| v.as_str()).unwrap_or("add");
let tags = args
.get("tags")
.and_then(|v| v.as_array())
.ok_or_else(|| wm_core::CoreError::InvalidArgs("Missing 'tags' array".into()))?;
let tag_list: Vec<String> = tags
.iter()
.filter_map(|t| t.as_str().map(String::from))
.collect();
match action {
"remove" => {
mem.metadata.tags.retain(|t| !tag_list.contains(t));
}
_ => {
for t in &tag_list {
if !mem.metadata.tags.contains(t) {
mem.metadata.tags.push(t.clone());
}
}
}
}
self.store.put(galaxy, &mem)?;
Ok(json!({
"status": "success",
"id": mem.metadata.id,
"action": action,
"tags": mem.metadata.tags,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryStatsTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryStatsTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow::read_only(vec![Resource::Galaxy("codex".into())]),
}
}
}
#[async_trait]
impl Tool for MemoryStatsTool {
fn input_schema(&self) -> Value {
schema(
&json!({
"galaxy": super::common::str_prop("Galaxy to summarize (optional; default codex)"),
}),
&[],
)
}
fn name(&self) -> &str {
"memory.stats"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Statistics for a galaxy (count, avg importance, tag frequency)"
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = parse_galaxy_or(args.get("galaxy").and_then(|v| v.as_str()), Galaxy::Codex)?;
let memories = self.store.scan(galaxy, 10_000)?;
let total = memories.len();
let avg_importance = if total > 0 {
memories.iter().map(|m| m.metadata.importance).sum::<f32>() / total as f32
} else {
0.0
};
let mut tag_freq: HashMap<String, u32> = HashMap::new();
for mem in &memories {
for tag in &mem.metadata.tags {
*tag_freq.entry(tag.clone()).or_insert(0) += 1;
}
}
let top_tags: Vec<(String, u32)> = tag_freq.into_iter().filter(|(_, c)| *c >= 2).collect();
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"count": total,
"avg_importance": (avg_importance * 100.0).round() / 100.0,
"tag_clusters": top_tags.len(),
"top_tags": top_tags.into_iter().take(10).collect::<Vec<_>>(),
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryHybridRecallTool {
store: Arc<MemoryStore>,
search: Option<Arc<SearchEngine>>,
recall: Option<Arc<RecallEngine>>,
associations: Option<Arc<AssociationStore>>,
stats: ToolStats,
effects: EffectRow,
route_name: &'static str,
}
impl MemoryHybridRecallTool {
pub fn new(
store: Arc<MemoryStore>,
search: Option<Arc<SearchEngine>>,
recall: Option<Arc<RecallEngine>>,
) -> Self {
Self::named("memory.hybrid_recall", store, search, recall)
}
pub fn as_search(
store: Arc<MemoryStore>,
search: Option<Arc<SearchEngine>>,
recall: Option<Arc<RecallEngine>>,
) -> Self {
Self::named("memory.search", store, search, recall)
}
#[must_use]
pub fn with_associations(mut self, associations: Option<Arc<AssociationStore>>) -> Self {
self.associations = associations;
self
}
fn named(
route_name: &'static str,
store: Arc<MemoryStore>,
search: Option<Arc<SearchEngine>>,
recall: Option<Arc<RecallEngine>>,
) -> Self {
Self {
store,
search,
recall,
associations: None,
stats: ToolStats::default(),
effects: EffectRow::read_only(vec![Resource::Galaxy("codex".into())]),
route_name,
}
}
}
pub struct MemoryReembedTool {
recall: Option<Arc<RecallEngine>>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryReembedTool {
#[must_use]
pub fn new(recall: Option<Arc<RecallEngine>>) -> Self {
Self {
recall,
stats: ToolStats::default(),
effects: EffectRow {
writes: {
let mut writes = super::common::memory_galaxy_writes();
writes.push(Resource::Galaxy("embeddings".into()));
writes
},
reads: super::common::memory_galaxy_reads(),
destructive: false,
..Default::default()
},
}
}
}
#[async_trait]
impl Tool for MemoryReembedTool {
fn name(&self) -> &str {
"memory.reembed"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn stats(&self) -> &ToolStats {
&self.stats
}
fn description(&self) -> &str {
"Backfill per-memory embedding vectors for memories that have none (dry-run by default; requires a real embedder). Bounded by limit; re-run to continue. Populates the persistent vector index used by hybrid recall."
}
fn input_schema(&self) -> Value {
schema(
&json!({
"galaxy": str_prop("Only this galaxy (optional; default: all memory galaxies)"),
"limit": int_prop("Maximum vectors to embed this pass (default 200; 0 = no cap)"),
"dry_run": bool_prop("Plan only, no writes (default true)"),
}),
&[],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = match args.get("galaxy").and_then(|v| v.as_str()) {
Some(name) => Some(parse_galaxy(name)?),
None => None,
};
let limit = args
.get("limit")
.and_then(serde_json::Value::as_u64)
.map_or(200usize, |v| v as usize);
let dry_run = args
.get("dry_run")
.and_then(serde_json::Value::as_bool)
.unwrap_or(true);
let Some(recall) = self.recall.as_ref() else {
return Ok(json!({
"status": "error",
"error": "no real embedder wired in this server — set WM_EMBEDDER_ENDPOINT (or the onnx backend) and restart; memory.reembed will not store stub noise",
}));
};
let report = recall.backfill_embeddings(galaxy, limit, dry_run)?;
let mut out = serde_json::to_value(&report).unwrap_or_else(|_| json!({}));
if let Some(obj) = out.as_object_mut() {
obj.insert("status".into(), json!("success"));
obj.insert(
"hint".into(),
json!(if dry_run {
"dry-run only — call again with dry_run: false to persist vectors"
} else {
"vectors persisted; the shared vector index is updated in this process, and restarted processes rehydrate it on first hybrid search"
}),
);
}
Ok(out)
}
}
fn recall_visible(galaxy: Galaxy, galaxy_explicit: bool) -> bool {
galaxy_explicit || galaxy != Galaxy::Telemetry
}
fn empty_result_hint(store: &MemoryStore, galaxy: Galaxy) -> String {
let mut populated: Vec<String> = Vec::new();
let mut requested_total = 0usize;
for g in Galaxy::memory_galaxies() {
if g == galaxy {
requested_total = store.count(g).unwrap_or(0);
continue;
}
let n = store.count(g).unwrap_or(0);
if n > 0 {
populated.push(format!("{} ({})", g.db_name(), n));
}
}
let location = if requested_total == 0 {
format!("galaxy '{}' contains no memories", galaxy_name(galaxy))
} else {
format!(
"no matches for this query in '{}' ({} memories)",
galaxy_name(galaxy),
requested_total
)
};
if populated.is_empty() {
format!("{location}; the store is empty")
} else {
format!(
"{}; other galaxies with content: {}. Pass an explicit \"galaxy\" to search there.",
location,
populated.join(", ")
)
}
}
fn empty_result_hint_all(store: &MemoryStore) -> String {
let mut populated: Vec<String> = Vec::new();
let mut total = 0usize;
for g in Galaxy::memory_galaxies() {
let n = store.count(g).unwrap_or(0);
total += n;
if n > 0 {
populated.push(format!("{} ({})", g.db_name(), n));
}
}
if populated.is_empty() {
"no matches for this query; the store is empty".to_string()
} else {
format!(
"no matches for this query across all memory galaxies ({} total); populated: {}",
total,
populated.join(", ")
)
}
}
#[async_trait]
impl Tool for MemoryHybridRecallTool {
fn name(&self) -> &str {
self.route_name
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Search memories: hybrid BM25+vector fusion with a real embedder; otherwise the episodic deterministic route, falling back to BM25 full-text. Every result discloses recall_mode (hybrid|bm25|episodic|fts|importance|cold|none) — bm25 means the fusion ranked with the vector weight configured to zero (the query embed is skipped when the vector half cannot contribute). memory.hybrid_recall is a compatibility alias."
}
fn input_schema(&self) -> Value {
schema(
&json!({
"query": str_prop("Full-text query"),
"galaxy": str_prop("Galaxy filter (optional; default: search all memory galaxies, results labeled; \"all\" is accepted as an alias for the unfiltered default)"),
"limit": super::common::positive_int_prop("Maximum results (default 10; must be >= 1)"),
"min_importance": num_prop("Minimum memory importance (0-1)"),
"min_score": num_prop("Absolute BM25 score floor"),
"min_score_ratio": num_prop("Relative floor: reject hits below this fraction of the top score"),
"min_trust": num_prop("Minimum source_trust (0-1): drop results below this trust floor"),
"include_cold": bool_prop("Opt-in unranked cold recovery (no thaw). Trust/importance floors apply; BM25 floors do not apply to unscored recovery. Search content is scrubbed navigation capped at 8192 characters; read by id/galaxy for the exact original."),
"cold_scan_limit": int_prop("Maximum cold records to scan when include_cold is set (default 2048)"),
}),
&["query"],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy_arg = galaxy_search_arg(args.get("galaxy").and_then(|v| v.as_str()));
let galaxy_explicit = galaxy_arg.is_some();
let galaxy = parse_galaxy_or(galaxy_arg, Galaxy::Codex)?;
let query = args.get("query").and_then(|v| v.as_str()).unwrap_or("");
let limit = super::common::positive_usize_arg(&args, "limit", 10)
.map_err(wm_core::CoreError::InvalidArgs)?;
let include_cold = args
.get("include_cold")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let cold_scan_limit = args
.get("cold_scan_limit")
.and_then(serde_json::Value::as_u64)
.map_or(2048, |v| v.clamp(1, 100_000) as usize);
let min_importance = super::common::bounded_f64_arg(&args, "min_importance", 0.0, Some(1.0))
.map_err(wm_core::CoreError::InvalidArgs)?
.unwrap_or(0.0) as f32;
let min_score = super::common::bounded_f64_arg(&args, "min_score", 0.0, None)
.map_err(wm_core::CoreError::InvalidArgs)?
.map(|v| v as f32)
.filter(|v| *v > 0.0);
let min_score_ratio =
super::common::bounded_f64_arg(&args, "min_score_ratio", 0.0, Some(1.0))
.map_err(wm_core::CoreError::InvalidArgs)?
.map_or(Some(0.05), |v| Some(v as f32));
let mut results = Vec::new();
let trust_weight = std::env::var("WM_TRUST_WEIGHT")
.ok()
.and_then(|v| v.parse::<f32>().ok())
.unwrap_or(0.0)
.clamp(0.0, 1.0);
let mut result_extra: Option<serde_json::Value> = None;
let mut trust_disclosure: Option<serde_json::Value> = None;
let mut recall_mode = "none";
let mut hybrid_available = self
.recall
.as_ref()
.is_some_and(|recall| recall.embedder_is_real());
let fused_mode = fused_mode_label(
self.recall
.as_ref()
.map_or(1.0, |recall| recall.config().vector_weight),
);
if hybrid_available {
let recall = self.recall.as_ref().expect("hybrid_available checked");
if !query.is_empty() {
let (hybrid_results, conformal) = recall.hybrid_search_with_disclosure(
query,
limit * 2,
galaxy_explicit.then_some(galaxy),
);
for hr in hybrid_results {
if !recall_visible(hr.galaxy, galaxy_explicit) {
continue;
}
if let Ok(Some(mem)) = self.store.get(hr.galaxy, hr.memory_id) {
if mem.metadata.importance >= min_importance
&& crate::expansion::common::mcp_visible(&mem)
&& crate::expansion::common::validity_visible(&mem)
{
let navigation = wm_memory::scrub_text(&mem.content);
results.push(with_navigation_disclosure(
json!({
"id": mem.metadata.id,
"galaxy": mem.metadata.galaxy.db_name(),
"content": navigation,
"importance": mem.metadata.importance,
"score": hr.score,
"raw_score": hr.raw_bm25_score,
"trust_factor": hr.trust_factor,
"corroboration": hr.corroboration,
"in_conformal_set": hr.in_conformal_set,
"bm25_score": hr.bm25_score,
"vector_score": hr.vector_score,
"trust": mem.metadata.source_trust,
"source": fused_mode,
}),
&mem.content,
));
}
}
}
if let Some(info) = conformal {
result_extra = serde_json::to_value(&info).ok();
}
if trust_weight > 0.0 {
trust_disclosure = Some(json!({
"wm_trust_weight": trust_weight,
"applied_in": "fuse_results",
}));
}
if !results.is_empty() {
recall_mode = fused_mode;
}
}
}
let mut hybrid_degraded: Option<String> = None;
if hybrid_available && results.is_empty() && !query.is_empty() {
if let Some(recall) = self.recall.as_ref() {
if let Err(error) = recall.embedder_probe() {
hybrid_degraded = Some(error.to_string());
hybrid_available = false;
}
}
}
if results.is_empty() && !query.is_empty() && !hybrid_available {
const EPISODIC_RECALL_POOL: usize = 100;
let pool = limit.max(EPISODIC_RECALL_POOL);
let episodic_hits = match self
.store
.episodic()
.search_with_limits(query, pool, pool, false)
{
Ok(hits) => hits,
Err(error) => {
tracing::warn!("episodic default-route search failed: {error}");
Vec::new()
}
};
for er in episodic_hits {
let Some((hit_galaxy, mem)) =
resolve_memory_across_galaxies(&self.store, er.record.id)
else {
continue;
};
if galaxy_explicit && hit_galaxy != galaxy {
continue;
}
if !recall_visible(hit_galaxy, galaxy_explicit) {
continue;
}
if mem.metadata.importance < min_importance
|| !crate::expansion::common::mcp_visible(&mem)
|| !crate::expansion::common::validity_visible(&mem)
{
continue;
}
let navigation = wm_memory::scrub_text(&mem.content);
results.push(with_navigation_disclosure(
json!({
"id": mem.metadata.id,
"galaxy": hit_galaxy.db_name(),
"content": navigation,
"importance": mem.metadata.importance,
"score": er.score,
"matched_terms": er.matched_terms,
"trust": mem.metadata.source_trust,
"source": "episodic",
}),
&mem.content,
));
}
if !results.is_empty() {
recall_mode = "episodic";
}
}
if results.is_empty() {
if let Some(ref search) = self.search {
if !query.is_empty() {
let opts = wm_memory::SearchOptions {
limit: limit * 2,
min_score,
relative_floor: min_score_ratio,
galaxy: galaxy_explicit.then_some(galaxy),
..wm_memory::SearchOptions::default()
};
let hits = search.search_opt(query, &opts)?;
for hit in hits {
if let Ok(id) = uuid::Uuid::parse_str(&hit.memory_id) {
let hit_galaxy = if galaxy_explicit {
Some(galaxy)
} else {
wm_core::Galaxy::all()
.into_iter()
.find(|g| g.db_name() == hit.galaxy)
};
let Some(hit_galaxy) = hit_galaxy else {
continue;
};
if !recall_visible(hit_galaxy, galaxy_explicit) {
continue;
}
if let Ok(Some(mem)) = self.store.get(hit_galaxy, id) {
if mem.metadata.importance >= min_importance
&& crate::expansion::common::mcp_visible(&mem)
&& crate::expansion::common::validity_visible(&mem)
{
let navigation = wm_memory::scrub_text(&mem.content);
results.push(with_navigation_disclosure(
json!({
"id": mem.metadata.id,
"galaxy": hit_galaxy.db_name(),
"content": navigation,
"importance": mem.metadata.importance,
"score": wm_memory::trust_weighted_score(
hit.score,
mem.metadata.source_trust,
trust_weight,
),
"raw_score": hit.score,
"normalized_score": hit.normalized_score,
"trust": mem.metadata.source_trust,
"source": "fts",
}),
&mem.content,
));
if recall_mode == "none" {
recall_mode = "fts";
}
}
}
}
}
}
}
}
if results.is_empty() && query.is_empty() {
let mut memories = self.store.scan(galaxy, 100)?;
memories.sort_by(|a, b| {
b.metadata
.importance
.partial_cmp(&a.metadata.importance)
.unwrap_or(std::cmp::Ordering::Equal)
});
for mem in memories
.iter()
.filter(|m| {
m.metadata.importance >= min_importance
&& crate::expansion::common::mcp_visible(m)
&& crate::expansion::common::validity_visible(m)
})
.take(limit)
{
results.push(json!({
"id": mem.metadata.id,
"content": &mem.content,
"importance": mem.metadata.importance,
"trust": mem.metadata.source_trust,
"score": mem.metadata.importance,
"source": "importance",
"content_representation": "verbatim",
"content_truncated": false,
"content_scrubbed": false,
"exact_read_available": true,
}));
if recall_mode == "none" {
recall_mode = "importance";
}
}
}
if !results.is_empty() {
if let Some(assoc_store) = &self.associations {
let mut anchors: Vec<(uuid::Uuid, f32)> = results
.iter()
.filter_map(|r| {
let id = r.get("id")?.as_str()?;
uuid::Uuid::parse_str(id).ok().map(|u| {
(
u,
r.get("score")
.and_then(serde_json::Value::as_f64)
.unwrap_or(0.0) as f32,
)
})
})
.collect();
anchors.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
anchors.dedup_by(|a, b| a.0 == b.0);
anchors.truncate(5);
let mut expansions: Vec<(uuid::Uuid, f32, f32, String, uuid::Uuid)> = Vec::new();
for (seed_id, seed_score) in &anchors {
let mut links = Vec::new();
if let Ok(outgoing) = assoc_store.find_from(self.store.env(), *seed_id) {
links.extend(outgoing);
}
if let Ok(incoming) = assoc_store.find_to(self.store.env(), *seed_id) {
links.extend(incoming);
}
for assoc in links {
if assoc.weight < 0.05 {
continue;
}
let neighbor = if assoc.target == *seed_id {
assoc.source
} else {
assoc.target
};
let score = seed_score * assoc.weight * 0.5;
if score <= 0.0 {
continue;
}
expansions.push((
neighbor,
score,
assoc.weight,
assoc.link_type.as_str().to_string(),
*seed_id,
));
}
}
expansions.sort_by(|a, b| {
b.1.partial_cmp(&a.1)
.unwrap_or(std::cmp::Ordering::Equal)
.then_with(|| a.0.cmp(&b.0))
});
expansions.dedup_by(|a, b| a.0 == b.0);
expansions.truncate(5);
let direct_ids: Vec<String> = results
.iter()
.filter_map(|r| {
r.get("id")
.and_then(serde_json::Value::as_str)
.map(String::from)
})
.collect();
for (neighbor_id, score, weight, link_type, seed_id) in expansions {
if direct_ids.iter().any(|id| id == &neighbor_id.to_string()) {
continue;
}
let Some((_, mem)) = resolve_memory_across_galaxies(&self.store, neighbor_id)
else {
continue;
};
if !recall_visible(mem.metadata.galaxy, galaxy_explicit) {
continue;
}
if mem.metadata.importance < min_importance
|| !crate::expansion::common::mcp_visible(&mem)
|| !crate::expansion::common::validity_visible(&mem)
{
continue;
}
let navigation = wm_memory::scrub_text(&mem.content);
results.push(with_navigation_disclosure(
json!({
"id": mem.metadata.id,
"content": navigation,
"importance": mem.metadata.importance,
"trust": mem.metadata.source_trust,
"score": score,
"weight": weight,
"link_type": link_type,
"via": seed_id.to_string(),
"source": "association",
}),
&mem.content,
));
}
}
}
if trust_weight > 0.0 {
results.sort_by(|a, b| {
b.get("score")
.and_then(serde_json::Value::as_f64)
.partial_cmp(&a.get("score").and_then(serde_json::Value::as_f64))
.unwrap_or(std::cmp::Ordering::Equal)
});
}
let min_trust = super::common::bounded_f64_arg(&args, "min_trust", 0.0, Some(1.0))
.map_err(wm_core::CoreError::InvalidArgs)?;
let pre_filter = results.len();
if let Some(min) = min_trust {
results.retain(
|r| match r.get("trust").and_then(serde_json::Value::as_f64) {
Some(t) => (t as f32) >= min as f32,
None => false,
},
);
}
let min_trust_filtered = pre_filter - results.len();
results.truncate(limit);
let cold_discovery: Option<serde_json::Value> = if include_cold && !query.is_empty() {
let terms: Vec<String> = query.split_whitespace().map(str::to_lowercase).collect();
let remaining = limit.saturating_sub(results.len());
if remaining == 0 {
Some(
json!({"enabled":true,"scanned":0,"candidates":0,"matched":0,"appended":0,"integrity_rejected":0,"private_skipped":0,"non_current_skipped":0,"eligibility_skipped":0,"stop_reason":"no_headroom","exhausted":false,"no_thaw":true,"ranked":false,"scan_order":"uuid_key","score_floors":"not_applicable_unscored_recovery"}),
)
} else {
let existing: std::collections::HashSet<String> = results
.iter()
.filter_map(|r| {
r.get("id")
.and_then(serde_json::Value::as_str)
.map(str::to_string)
})
.collect();
let outcome = self.store.find_cold_matching_eligible(
&terms,
if galaxy_explicit { Some(galaxy) } else { None },
remaining,
cold_scan_limit,
|mem| {
recall_visible(mem.metadata.galaxy, galaxy_explicit)
&& mem.metadata.importance >= min_importance
&& min_trust
.is_none_or(|floor| f64::from(mem.metadata.source_trust) >= floor)
&& !existing.contains(&mem.metadata.id.to_string())
},
)?;
let mut appended = 0usize;
for record in &outcome.records {
if appended >= remaining {
break;
}
let id = record.id.to_string();
if existing.contains(&id) {
continue;
}
let mem = record.decompress()?;
let navigation = wm_memory::search::scrub_text(&mem.content);
results.push(with_navigation_disclosure(
json!({
"id": id,
"galaxy": mem.metadata.galaxy.db_name(),
"content": navigation,
"importance": mem.metadata.importance,
"trust": mem.metadata.source_trust,
"score": serde_json::Value::Null,
"source": "cold",
"cold": true,
"integrity": "verified",
"model_visible": !mem.metadata.model_exclude,
"tags": &mem.metadata.tags,
}),
&mem.content,
));
appended += 1;
}
if appended > 0 && recall_mode == "none" {
recall_mode = "cold";
}
Some(json!({
"enabled": true,
"scanned": outcome.scanned,
"candidates": outcome.candidates,
"matched": outcome.matched,
"appended": appended,
"integrity_rejected": outcome.integrity_rejected,
"private_skipped": outcome.private_skipped,
"non_current_skipped": outcome.non_current_skipped,
"eligibility_skipped": outcome.eligibility_skipped,
"stop_reason": outcome.stop_reason,
"exhausted": outcome.stop_reason == wm_memory::cold_storage::ColdDiscoveryStop::Exhausted,
"ranked": false,
"scan_order": "uuid_key",
"score_floors": "not_applicable_unscored_recovery",
"no_thaw": true,
}))
}
} else {
None
};
let evidence_bundle = build_evidence_bundle(&self.store, &results);
let bytes_injected: usize = results
.iter()
.filter_map(|r| r.get("content").and_then(serde_json::Value::as_str))
.map(str::len)
.sum();
let bytes_available: usize = results
.iter()
.filter_map(|r| r.get("id").and_then(serde_json::Value::as_str))
.filter_map(|id| uuid::Uuid::parse_str(id).ok())
.filter_map(|id| resolve_memory_across_galaxies(&self.store, id))
.map(|(_, mem)| mem.content.len())
.sum();
super::common::append_savings_row(
&self.store,
&json!({
"ts_ms": wm_core::time::now_unix_millis(),
"op": "recall",
"tool": self.route_name,
"recall_mode": recall_mode,
"results": results.len(),
"bytes_available": bytes_available,
"bytes_injected": bytes_injected,
}),
);
let hint = if results.is_empty() && !query.is_empty() {
Some(if galaxy_explicit {
empty_result_hint(&self.store, galaxy)
} else {
empty_result_hint_all(&self.store)
})
} else {
None
};
let mut out = json!({
"status": "success",
"galaxy": if galaxy_explicit {
serde_json::Value::from(galaxy_name(galaxy))
} else {
serde_json::Value::from("all")
},
"count": results.len(),
"recall_mode": recall_mode,
"results": results,
"hint": hint,
});
if let Some(extra) = result_extra {
out["conformal_set"] = extra;
}
if let Some(td) = trust_disclosure {
out["trust_weighting"] = td;
}
if let Some(reason) = hybrid_degraded {
out["hybrid_degraded"] = json!(reason);
}
if let Some(min) = min_trust {
out["min_trust"] = json!(min);
out["min_trust_filtered"] = json!(min_trust_filtered);
}
if let Some(cd) = cold_discovery {
out["cold_discovery"] = cd;
}
out["evidence_bundle"] = evidence_bundle;
if !query.is_empty() && results.is_empty() {
out["abstention"] = json!({
"status": "insufficient_evidence",
"reason": "no_results_above_floors",
"scope": "retrieval",
});
} else if !query.is_empty() {
if let Some(abstention) = weak_evidence_abstention(&results, query) {
out["abstention"] = abstention;
}
}
Ok(out)
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryRecallFeedbackTool {
recall: Option<Arc<RecallEngine>>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryRecallFeedbackTool {
#[must_use]
pub fn new(recall: Option<Arc<RecallEngine>>) -> Self {
Self {
recall,
stats: ToolStats::default(),
effects: EffectRow {
writes: vec![Resource::Filesystem],
..Default::default()
},
}
}
}
#[async_trait]
impl Tool for MemoryRecallFeedbackTool {
fn name(&self) -> &str {
"memory.recall_feedback"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Record relevance feedback for conformal retrieval calibration (V8 S8). Args: samples (array of {score: number 0-1, relevant: bool}) or score+relevant for a single sample. Requires WM_RECALL_CONFORMAL_ALPHA."
}
fn input_schema(&self) -> Value {
schema(
&json!({
"samples": {"type": "array", "items": {"type": "object"}, "description": "Feedback samples: [{score: 0-1 fused score, relevant: bool}]"},
"score": num_prop("Single-sample fused score (0-1)"),
"relevant": {"type": "boolean", "description": "Single-sample relevance label"},
}),
&[],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let Some(ref recall) = self.recall else {
return Ok(json!({
"status": "error",
"message": "no recall engine on this server (hybrid search unavailable) — nothing to calibrate",
}));
};
let mut samples: Vec<(f32, bool)> = Vec::new();
if let Some(list) = args.get("samples").and_then(Value::as_array) {
for s in list {
let score = s.get("score").and_then(Value::as_f64).unwrap_or(-1.0);
let relevant = s.get("relevant").and_then(Value::as_bool);
if !(0.0..=1.0).contains(&score) || relevant.is_none() {
return Err(wm_core::CoreError::InvalidArgs(
"each sample needs score in [0,1] and a boolean 'relevant'".into(),
));
}
samples.push((score as f32, relevant.unwrap_or(false)));
}
} else if let Some(score) = args.get("score").and_then(Value::as_f64) {
let relevant = args
.get("relevant")
.and_then(Value::as_bool)
.ok_or_else(|| {
wm_core::CoreError::InvalidArgs("'relevant' is required with 'score'".into())
})?;
if !(0.0..=1.0).contains(&score) {
return Err(wm_core::CoreError::InvalidArgs(
"'score' must be within [0,1]".into(),
));
}
samples.push((score as f32, relevant));
} else {
return Err(wm_core::CoreError::InvalidArgs(
"provide 'samples' (array of {score, relevant}) or a single 'score' + 'relevant'"
.into(),
));
}
let mut recorded = 0usize;
let mut count = 0usize;
for (score, relevant) in samples {
count = recall.record_relevance_feedback(score, relevant)?;
recorded += 1;
}
let status = recall
.conformal_disclosure(&mut Vec::new())?
.map_or_else(|| "off".into(), |info| info.status);
Ok(json!({
"status": "success",
"recorded": recorded,
"calibration_samples": count,
"conformal_status": status,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryEpisodicSearchTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryEpisodicSearchTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow::read_only(vec![Resource::Galaxy("episodic_records".into())]),
}
}
}
#[async_trait]
impl Tool for MemoryEpisodicSearchTool {
fn name(&self) -> &str {
"memory.episodic_search"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"[V6 Experimental] Search explicit episodic records with provenance and lifecycle filtering"
}
fn input_schema(&self) -> Value {
schema(
&json!({
"query": str_prop("Full-text query"),
"limit": int_prop("Maximum results (default 10)"),
"candidate_limit": int_prop("Maximum candidates to score (default 2x limit). This is the deterministic retrieval width — keep it wide for tail recall; bound rerank embedding cost with rerank_pool instead (measured on the 50q set: wide retrieval + small pool beat narrowed retrieval on R@1 and R@10)"),
"include_historical": {
"type": "boolean",
"description": "Include superseded, revoked, and archived records",
},
"rerank": {
"type": "boolean",
"description": "Enable vector reranking (requires embedder, default false)",
},
"rerank_alpha": {
"type": "number",
"description": "Rerank mode selector (default 0.7): <1.0 hybrid blend weight; >=1.0 near-tie cosine tiebreaker; >=2.0 protected top-K full cosine reorder (recall@limit preserved by construction when the candidate set is not narrowed). Blending alphas can drop correct items out of top-K — prefer >=2.0 when recall@limit matters",
},
"rerank_pool": int_prop("Embedding/reorder width for rerank (default 0 = auto: min(50, max(limit, candidate_limit)), capped at 50). Lower it on CPU embedders to cut latency without narrowing candidate_limit (the deterministic retrieval width)"),
"min_score": {
"type": "number",
"description": "Minimum score threshold; results below this are dropped (abstention). Default 0.0 (no threshold)",
},
"min_coverage": {
"type": "number",
"description": "Minimum query-term coverage ratio (0.0-1.0); results with lower coverage are dropped. E.g. 0.5 requires at least half the query terms to match. Default 0.0 (no threshold)",
},
}),
&["query"],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let query = args.get("query").and_then(Value::as_str).unwrap_or("");
let limit = args.get("limit").and_then(Value::as_u64).unwrap_or(10) as usize;
let candidate_limit =
args.get("candidate_limit")
.and_then(Value::as_u64)
.unwrap_or_else(|| limit.saturating_mul(2) as u64) as usize;
let include_historical = args
.get("include_historical")
.and_then(Value::as_bool)
.unwrap_or(false);
let rerank = args.get("rerank").and_then(Value::as_bool).unwrap_or(false);
let rerank_alpha = args
.get("rerank_alpha")
.and_then(Value::as_f64)
.unwrap_or(0.7) as f32;
let rerank_pool = args.get("rerank_pool").and_then(Value::as_u64).unwrap_or(0) as usize;
let min_score = args
.get("min_score")
.and_then(Value::as_f64)
.map(|v| v as f32);
let min_coverage = args
.get("min_coverage")
.and_then(Value::as_f64)
.map(|v| v as f32);
let query_term_count: usize = {
const STOPWORDS: &[&str] = &[
"the", "a", "an", "is", "are", "was", "were", "be", "been", "being", "have", "has",
"had", "do", "does", "did", "will", "would", "could", "should", "may", "might",
"must", "can", "shall", "to", "of", "in", "on", "at", "by", "for", "with", "about",
"as", "into", "like", "through", "after", "over", "between", "out", "against",
"during", "without", "before", "under", "around", "among", "i", "me", "my", "we",
"us", "our", "you", "your", "he", "him", "his", "she", "her", "it", "its", "they",
"them", "their", "what", "whats", "who", "when", "where", "why", "how", "and",
"or", "but", "not", "no", "nor", "so", "yet", "both", "either", "neither", "this",
"that", "these", "those", "there", "here", "now", "then", "than",
];
query
.split(|c: char| !c.is_alphanumeric())
.filter(|t| t.len() > 1)
.map(str::to_ascii_lowercase)
.filter(|t| !STOPWORDS.contains(&t.as_str()))
.collect::<std::collections::HashSet<_>>()
.len()
};
let raw_results = if rerank {
self.store.episodic().search_with_rerank(
query,
limit,
candidate_limit,
include_historical,
rerank_alpha,
rerank_pool,
)?
} else {
self.store.episodic().search_with_limits(
query,
limit,
candidate_limit,
include_historical,
)?
};
let is_count_query = query.to_ascii_lowercase().contains("how many");
let abstain = min_coverage.is_some()
&& !is_count_query
&& query_term_count >= 3
&& !raw_results.iter().any(|hit| hit.matched_terms >= 2);
let visible: Vec<_> = raw_results
.into_iter()
.filter(|hit| !hit.record.is_private && !hit.record.model_exclude)
.filter(|hit| min_score.is_none_or(|ms| hit.score >= ms))
.filter(|_| !abstain)
.take(limit)
.collect();
let conflicts = detect_conflicts(&visible);
let results = visible
.into_iter()
.map(|hit| {
json!({
"id": hit.record.id,
"content": wm_memory::scrub_text(&hit.record.content),
"score": hit.score,
"matched_terms": hit.matched_terms,
"session_id": hit.record.session_id,
"sequence": hit.record.sequence,
"created_at": hit.record.created_at,
"validity": hit.record.validity,
"provenance": hit.record.provenance,
"source": "episodic",
})
})
.collect::<Vec<_>>();
Ok(json!({
"status": "success",
"count": results.len(),
"current_resolution": wm_memory::episodic::is_current_query(query),
"conflicts": conflicts,
"results": results,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryAggregateTool {
search: Option<Arc<SearchEngine>>,
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryAggregateTool {
pub fn new(search: Option<Arc<SearchEngine>>, store: Arc<MemoryStore>) -> Self {
Self {
search,
store,
stats: ToolStats::default(),
effects: EffectRow::read_only(vec![Resource::Galaxy("codex".into())]),
}
}
}
fn session_ordinal(tags: &[String]) -> Option<u64> {
tags.iter().find_map(|tag| {
let rest = tag.strip_prefix("session_")?;
rest.parse::<u64>().ok()
})
}
fn contains_term(content: &str, term: &str) -> bool {
let lowered = content.to_ascii_lowercase();
let variants = [term.to_string(), strip_suffix(term)];
for variant in &variants {
if variant.len() < 2 {
continue;
}
let mut start = 0;
while let Some(pos) = lowered[start..].find(variant.as_str()) {
let before_ok = pos == 0
|| !lowered[start + pos - 1..start + pos]
.chars()
.next()
.is_some_and(char::is_alphanumeric);
let end = start + pos + variant.len();
let after_ok = end >= lowered.len()
|| !lowered[end..]
.chars()
.next()
.is_some_and(char::is_alphanumeric);
if before_ok && after_ok {
return true;
}
start += pos + variant.len();
}
}
false
}
fn strip_suffix(term: &str) -> String {
for suffix in ["ing", "ed", "es", "s"] {
if let Some(stem) = term.strip_suffix(suffix) {
if stem.len() >= 2 {
return stem.to_string();
}
}
}
term.to_string()
}
#[async_trait]
impl Tool for MemoryAggregateTool {
fn name(&self) -> &str {
"memory.aggregate"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Aggregate over memories matching a query: count, distinct session count, or session span (cross-session synthesis)"
}
fn input_schema(&self) -> Value {
schema(
&json!({
"query": str_prop("Full-text query selecting the memories to aggregate over"),
"metric": str_prop("Aggregate metric: count | session_count | session_span"),
"limit": super::common::positive_int_prop("Maximum candidates considered (default 50; must be >= 1)"),
}),
&["query", "metric"],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let query = args
.get("query")
.and_then(Value::as_str)
.ok_or_else(|| wm_core::CoreError::InvalidArgs("query (string) required".into()))?;
let metric = args
.get("metric")
.and_then(Value::as_str)
.ok_or_else(|| wm_core::CoreError::InvalidArgs("metric (string) required".into()))?;
let limit = super::common::positive_usize_arg(&args, "limit", 50)
.map_err(wm_core::CoreError::InvalidArgs)?;
let Some(search) = self.search.as_ref() else {
return Err(wm_core::CoreError::Memory(
"search engine unavailable for aggregation".into(),
));
};
let results = search.search(query, limit)?;
let mut memories = Vec::new();
for r in &results {
let Some(galaxy) = wm_core::Galaxy::from_db_name(&r.galaxy) else {
continue;
};
let Ok(id) = uuid::Uuid::parse_str(&r.memory_id) else {
continue;
};
let Ok(Some(mem)) = self.store.get(galaxy, id) else {
continue;
};
if super::common::mcp_visible(&mem) && super::common::validity_visible(&mem) {
memories.push((r.score, mem));
}
}
let evidence: Vec<Value> = memories
.iter()
.map(|(score, mem)| {
json!({
"memory_id": mem.metadata.id.to_string(),
"score": score,
"content": wm_memory::scrub_text(&mem.content),
"tags": mem.metadata.tags,
})
})
.collect();
let session_tagged: Vec<_> = memories
.iter()
.filter(|(_, mem)| session_ordinal(&mem.metadata.tags).is_some())
.collect();
let (anchored, anchor): (Vec<_>, &str) = if metric == "count" {
(Vec::new(), "none")
} else if session_tagged.len() < 2 {
(session_tagged.clone(), "session_tagged_fallback")
} else {
let terms: Vec<String> = wm_memory::strip_stopwords(query)
.split(|c: char| !c.is_alphanumeric())
.filter(|t| t.len() > 1)
.map(str::to_ascii_lowercase)
.collect();
let mut best: Option<(String, usize)> = None;
for term in &terms {
let count = session_tagged
.iter()
.filter(|(_, mem)| contains_term(&mem.content, term))
.count();
if count == 0 {
continue;
}
let better = best
.as_ref()
.is_none_or(|(_, best_count)| count < *best_count);
if better {
best = Some((term.clone(), count));
}
}
match best {
Some((term, _)) => (
session_tagged
.iter()
.filter(|(_, mem)| contains_term(&mem.content, &term))
.copied()
.collect(),
"rarest_term",
),
None => (session_tagged.clone(), "session_tagged_fallback"),
}
};
let aggregate = match metric {
"count" => json!({
"metric": "count",
"value": memories.len(),
"content": format!("{} memories", memories.len()),
}),
"session_count" => {
let sessions: std::collections::HashSet<u64> = anchored
.iter()
.filter_map(|(_, mem)| session_ordinal(&mem.metadata.tags))
.collect();
json!({
"metric": "session_count",
"value": sessions.len(),
"content": format!("{} distinct sessions", sessions.len()),
})
}
"session_span" => {
let ordinals: Vec<u64> = anchored
.iter()
.filter_map(|(_, mem)| session_ordinal(&mem.metadata.tags))
.collect();
if ordinals.is_empty() {
json!({
"metric": "session_span",
"value": null,
"content": "no session-tagged evidence found",
})
} else {
let span = ordinals.iter().max().unwrap() - ordinals.iter().min().unwrap();
json!({
"metric": "session_span",
"value": span,
"unit": "sessions",
"content": format!("{span} sessions"),
})
}
}
other => {
return Err(wm_core::CoreError::InvalidArgs(format!(
"unknown metric '{other}' (count | session_count | session_span)"
)));
}
};
Ok(json!({
"status": "success",
"query": query,
"total": memories.len(),
"session_tagged": session_tagged.len(),
"anchor": anchor,
"limit_hit": results.len() >= limit,
"aggregate": aggregate,
"results": evidence,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemorySortTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemorySortTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow::read_only(vec![Resource::Galaxy("codex".into())]),
}
}
}
#[async_trait]
impl Tool for MemorySortTool {
fn input_schema(&self) -> Value {
schema(
&json!({
"galaxy": super::common::str_prop("Galaxy to sort (optional; default codex)"),
"sort_by": super::common::str_prop("Sort key: importance | created_at | accessed_at | access_count"),
"order": super::common::str_prop("Order: asc | desc (default desc)"),
"limit": super::common::int_prop("Maximum entries (default 50)"),
}),
&[],
)
}
fn name(&self) -> &str {
"memory.sort"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Sort memories by importance, recency, or access count"
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = parse_galaxy_or(args.get("galaxy").and_then(|v| v.as_str()), Galaxy::Codex)?;
let sort_by = args
.get("sort_by")
.and_then(|v| v.as_str())
.unwrap_or("importance");
let order = args.get("order").and_then(|v| v.as_str()).unwrap_or("desc");
let limit = args
.get("limit")
.and_then(serde_json::Value::as_u64)
.unwrap_or(50) as usize;
let mut memories = self.store.scan(galaxy, 10_000)?;
memories.retain(|m| {
crate::expansion::common::mcp_visible(m)
&& crate::expansion::common::validity_visible(m)
});
match sort_by {
"importance" => memories.sort_by(|a, b| {
b.metadata
.importance
.partial_cmp(&a.metadata.importance)
.unwrap_or(std::cmp::Ordering::Equal)
}),
"recency" => memories.sort_by_key(|x| std::cmp::Reverse(x.metadata.created_at)),
"accessed" => {
memories.sort_by_key(|x| std::cmp::Reverse(x.metadata.accessed_at));
}
"access_count" => {
memories.sort_by_key(|x| std::cmp::Reverse(x.metadata.access_count));
}
_ => {
return Err(wm_core::CoreError::InvalidArgs(format!(
"Unknown sort_by: '{sort_by}'. Use importance, recency, accessed, or access_count"
)));
}
}
if order == "asc" {
memories.reverse();
}
let total = memories.len();
memories.truncate(limit);
let results: Vec<Value> = memories
.iter()
.map(|m| {
json!({
"id": m.metadata.id,
"content": &m.content,
"importance": m.metadata.importance,
"created_at": m.metadata.created_at.to_rfc3339(),
"accessed_at": m.metadata.accessed_at.to_rfc3339(),
"access_count": m.metadata.access_count,
"tags": &m.metadata.tags,
})
})
.collect();
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"sort_by": sort_by,
"order": order,
"total": total,
"returned": results.len(),
"memories": results,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryFilterTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryFilterTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow::read_only(vec![Resource::Galaxy("codex".into())]),
}
}
}
#[async_trait]
impl Tool for MemoryFilterTool {
fn name(&self) -> &str {
"memory.filter"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Filter memories by tags, date range, importance thresholds, and a content query substring"
}
fn input_schema(&self) -> Value {
super::common::schema(
&json!({
"galaxy": super::common::str_prop("Galaxy to filter (default codex)"),
"tags": super::common::str_array_prop("Filter: memories with all of these tags"),
"exclude_tags": super::common::str_array_prop("Filter: drop memories carrying any of these tags"),
"min_importance": super::common::num_prop("Filter: minimum importance (0-1)"),
"max_importance": super::common::num_prop("Filter: maximum importance (0-1)"),
"query": super::common::str_prop("Filter: every whitespace-separated term must appear (case-insensitive) in the content or title"),
"created_after": super::common::str_prop("Filter: only memories created at or after this RFC 3339 timestamp (e.g. 2026-08-01T00:00:00Z)"),
"created_before": super::common::str_prop("Filter: only memories created at or before this RFC 3339 timestamp"),
"limit": super::common::int_prop("Maximum entries (default 50)"),
"offset": super::common::int_prop("Skip this many matching entries before returning (default 0)"),
}),
&[],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = parse_galaxy_or(args.get("galaxy").and_then(|v| v.as_str()), Galaxy::Codex)?;
let tags: Vec<String> = args
.get("tags")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|t| t.as_str().map(String::from))
.collect()
})
.unwrap_or_default();
let exclude_tags: Vec<String> = args
.get("exclude_tags")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|t| t.as_str().map(String::from))
.collect()
})
.unwrap_or_default();
let min_importance = args
.get("min_importance")
.and_then(serde_json::Value::as_f64)
.unwrap_or(0.0) as f32;
let max_importance = args
.get("max_importance")
.and_then(serde_json::Value::as_f64)
.unwrap_or(1.0) as f32;
let limit = args
.get("limit")
.and_then(serde_json::Value::as_u64)
.unwrap_or(50) as usize;
let offset = args
.get("offset")
.and_then(serde_json::Value::as_u64)
.unwrap_or(0) as usize;
let parse_bound = |name: &str| -> wm_core::Result<Option<chrono::DateTime<chrono::Utc>>> {
match args.get(name).and_then(|v| v.as_str()) {
Some(s) if !s.trim().is_empty() => chrono::DateTime::parse_from_rfc3339(s.trim())
.map(|t| Some(t.with_timezone(&chrono::Utc)))
.map_err(|_| {
wm_core::CoreError::InvalidArgs(format!(
"{name} must be an RFC 3339 timestamp (e.g. \"2026-08-01T00:00:00Z\"), got: {s}"
))
}),
_ => Ok(None),
}
};
let created_after = parse_bound("created_after")?;
let created_before = parse_bound("created_before")?;
let query_terms: Vec<String> = args
.get("query")
.and_then(|v| v.as_str())
.map(|q| q.split_whitespace().map(str::to_lowercase).collect())
.unwrap_or_default();
let memories = self.store.scan(galaxy, 10_000)?;
let matched: Vec<&wm_memory::Memory> = memories
.iter()
.filter(|m| {
if !crate::expansion::common::mcp_visible(m) {
return false;
}
if !crate::expansion::common::validity_visible(m) {
return false;
}
if m.metadata.importance < min_importance || m.metadata.importance > max_importance
{
return false;
}
if !tags.is_empty() && !tags.iter().all(|t| m.metadata.tags.contains(t)) {
return false;
}
if exclude_tags
.iter()
.any(|t| m.metadata.tags.iter().any(|mt| mt == t))
{
return false;
}
if let Some(after) = created_after {
if m.metadata.created_at < after {
return false;
}
}
if let Some(before) = created_before {
if m.metadata.created_at > before {
return false;
}
}
if !query_terms.is_empty() {
let haystack = match &m.metadata.title {
Some(title) => format!("{}\n{}", m.content, title).to_lowercase(),
None => m.content.to_lowercase(),
};
if !query_terms.iter().all(|t| haystack.contains(t)) {
return false;
}
}
true
})
.collect();
let filtered: Vec<&&wm_memory::Memory> = matched.iter().skip(offset).take(limit).collect();
let total_scanned = memories.len();
let results: Vec<Value> = filtered
.iter()
.map(|m| {
json!({
"id": m.metadata.id,
"content": &m.content,
"importance": m.metadata.importance,
"tags": &m.metadata.tags,
"created_at": m.metadata.created_at.to_rfc3339(),
})
})
.collect();
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"scanned": total_scanned,
"matched": matched.len(),
"offset": offset,
"returned": results.len(),
"filters": {
"tags": tags,
"exclude_tags": exclude_tags,
"min_importance": min_importance,
"max_importance": max_importance,
"query_terms": query_terms,
"created_after": created_after.map(|t| t.to_rfc3339()),
"created_before": created_before.map(|t| t.to_rfc3339()),
},
"memories": results,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryDeduplicateTool {
store: Arc<MemoryStore>,
search: Option<Arc<SearchEngine>>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryDeduplicateTool {
pub fn new(store: Arc<MemoryStore>, search: Option<Arc<SearchEngine>>) -> Self {
Self {
store,
search,
stats: ToolStats::default(),
effects: EffectRow {
writes: super::common::memory_galaxy_writes(),
reads: super::common::memory_galaxy_reads(),
destructive: true,
..Default::default()
},
}
}
}
#[async_trait]
impl Tool for MemoryDeduplicateTool {
fn name(&self) -> &str {
"memory.deduplicate"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Find and merge duplicate memories by content hash or similarity"
}
fn input_schema(&self) -> Value {
super::common::schema(
&json!({
"galaxy": super::common::str_prop("Galaxy to deduplicate"),
"mode": super::common::str_prop("Strategy: hash | similarity (default: hash)"),
"limit": super::common::int_prop("Maximum entries to scan"),
"dry_run": super::common::bool_prop("Preview only (default: true)"),
}),
&["galaxy"],
)
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = parse_galaxy_or(args.get("galaxy").and_then(|v| v.as_str()), Galaxy::Codex)?;
let mode = args.get("mode").and_then(|v| v.as_str()).unwrap_or("hash");
let dry_run = args
.get("dry_run")
.and_then(serde_json::Value::as_bool)
.unwrap_or(true);
let limit = args
.get("limit")
.and_then(serde_json::Value::as_u64)
.unwrap_or(10_000) as usize;
let memories = self.store.scan(galaxy, limit)?;
match mode {
"hash" => {
let mut seen_hashes: HashMap<String, uuid::Uuid> = HashMap::new();
let mut duplicates: Vec<Value> = Vec::new();
for mem in &memories {
let hash = &mem.metadata.content_hash;
if let Some(existing_id) = seen_hashes.get(hash) {
if *existing_id != mem.metadata.id {
duplicates.push(json!({
"id": mem.metadata.id,
"duplicate_of": existing_id,
"content_preview": mem.content.chars().take(100).collect::<String>(),
"importance": mem.metadata.importance,
}));
if !dry_run {
self.store.delete(galaxy, mem.metadata.id)?;
super::common::deindex(
self.search.as_deref(),
&mem.metadata.id.to_string(),
);
}
}
} else {
seen_hashes.insert(hash.clone(), mem.metadata.id);
}
}
let removed = if dry_run { 0 } else { duplicates.len() };
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"mode": mode,
"dry_run": dry_run,
"scanned": memories.len(),
"duplicates_found": duplicates.len(),
"removed": removed,
"duplicates": duplicates,
}))
}
"content" => {
let mut duplicates: Vec<Value> = Vec::new();
let mut removed_count = 0u32;
for i in 0..memories.len() {
for j in (i + 1)..memories.len() {
if memories[i].content == memories[j].content {
duplicates.push(json!({
"id": memories[j].metadata.id,
"duplicate_of": memories[i].metadata.id,
"content_preview": memories[j].content.chars().take(100).collect::<String>(),
}));
if !dry_run {
self.store.delete(galaxy, memories[j].metadata.id)?;
super::common::deindex(
self.search.as_deref(),
&memories[j].metadata.id.to_string(),
);
removed_count += 1;
}
break;
}
}
}
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"mode": mode,
"dry_run": dry_run,
"scanned": memories.len(),
"duplicates_found": duplicates.len(),
"removed": removed_count,
"duplicates": duplicates,
}))
}
_ => Err(wm_core::CoreError::InvalidArgs(format!(
"Unknown mode: '{mode}'. Use 'hash' or 'content'"
))),
}
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
pub struct MemoryExportTool {
store: Arc<MemoryStore>,
stats: ToolStats,
effects: EffectRow,
}
impl MemoryExportTool {
pub fn new(store: Arc<MemoryStore>) -> Self {
Self {
store,
stats: ToolStats::default(),
effects: EffectRow::read_only(vec![Resource::Galaxy("codex".into())]),
}
}
}
#[async_trait]
impl Tool for MemoryExportTool {
fn input_schema(&self) -> Value {
schema(
&json!({
"galaxy": super::common::str_prop("Galaxy to export (optional; default codex)"),
"format": super::common::str_prop("Export format: json | jsonl | markdown"),
"limit": super::common::int_prop("Maximum entries to export"),
}),
&[],
)
}
fn name(&self) -> &str {
"memory.export"
}
fn gana(&self) -> Gana {
Gana::WinnowingBasket
}
fn effects(&self) -> &EffectRow {
&self.effects
}
fn description(&self) -> &str {
"Export memories in JSON, CSV, or Markdown format"
}
async fn call(&self, _ctx: &mut Context, args: Value) -> wm_core::Result<Value> {
let galaxy = parse_galaxy_or(args.get("galaxy").and_then(|v| v.as_str()), Galaxy::Codex)?;
let format = args
.get("format")
.and_then(|v| v.as_str())
.unwrap_or("json");
let limit = args
.get("limit")
.and_then(serde_json::Value::as_u64)
.unwrap_or(1000) as usize;
let memories = self.store.scan(galaxy, limit)?;
let exported = match format {
"json" => {
let entries: Vec<Value> = memories
.iter()
.map(|m| {
json!({
"id": m.metadata.id,
"content": &m.content,
"tags": &m.metadata.tags,
"importance": m.metadata.importance,
"created_at": m.metadata.created_at.to_rfc3339(),
"access_count": m.metadata.access_count,
})
})
.collect();
serde_json::to_string_pretty(&entries).unwrap_or_default()
}
"csv" => {
let mut csv = String::from("id,content,tags,importance,created_at,access_count\n");
for m in &memories {
let tags = m.metadata.tags.join(";");
let content = m.content.replace('\n', " ").replace('"', "'");
let _ = writeln!(
csv,
"{},{},{},{:.3},{},{}",
m.metadata.id,
content,
tags,
m.metadata.importance,
m.metadata.created_at.to_rfc3339(),
m.metadata.access_count,
);
}
csv
}
"markdown" => {
let mut md = format!("# Memory Export: {}\n\n", galaxy_name(galaxy));
let _ = write!(md, "Total memories: {}\n\n", memories.len());
for m in &memories {
let _ = write!(
md,
"## {}\n\n- **Importance**: {:.2}\n- **Tags**: {}\n- **Created**: {}\n- **Access Count**: {}\n\n{}\n\n---\n\n",
m.metadata.id,
m.metadata.importance,
m.metadata.tags.join(", "),
m.metadata.created_at.to_rfc3339(),
m.metadata.access_count,
m.content,
);
}
md
}
_ => {
return Err(wm_core::CoreError::InvalidArgs(format!(
"Unknown format: '{format}'. Use json, csv, or markdown"
)));
}
};
Ok(json!({
"status": "success",
"galaxy": galaxy_name(galaxy),
"format": format,
"count": memories.len(),
"export": exported,
}))
}
fn stats(&self) -> &ToolStats {
&self.stats
}
}
#[cfg(test)]
mod tests {
use super::*;
use wm_core::{EpisodicKind, EpisodicRecord, Galaxy, Provenance, ProvenanceSource};
use wm_memory::{Association, AssociationStore, LinkType, Memory, MemoryStore};
fn test_store() -> Arc<MemoryStore> {
let dir = tempfile::tempdir().unwrap();
Arc::new(MemoryStore::open_default(dir.path()).unwrap())
}
fn populate_memories(store: &Arc<MemoryStore>, galaxy: Galaxy) {
let mut m1 = Memory::new(galaxy, "First memory about rust".into());
m1.metadata.importance = 0.9;
m1.metadata.tags = vec!["rust".into(), "programming".into()];
let _ = store.put(galaxy, &m1);
let mut m2 = Memory::new(galaxy, "Second memory about python".into());
m2.metadata.importance = 0.5;
m2.metadata.tags = vec!["python".into()];
let _ = store.put(galaxy, &m2);
let mut m3 = Memory::new(galaxy, "Third memory about rust".into());
m3.metadata.importance = 0.3;
m3.metadata.tags = vec!["rust".into(), "tutorial".into()];
let _ = store.put(galaxy, &m3);
}
#[tokio::test]
async fn episodic_search_filters_private_records() {
let store = test_store();
let public = EpisodicRecord::new(
None,
1,
EpisodicKind::Observation,
"public retrieval evidence",
Provenance::new(ProvenanceSource::User),
);
let private = EpisodicRecord::new(
None,
2,
EpisodicKind::Observation,
"private retrieval evidence",
Provenance::new(ProvenanceSource::User),
)
.with_visibility(true, false);
store.episodic().append(&public).unwrap();
store.episodic().append(&private).unwrap();
let tool = MemoryEpisodicSearchTool::new(store);
let mut ctx = Context::default();
let result = tool
.call(
&mut ctx,
json!({"query": "retrieval evidence", "limit": 10}),
)
.await
.unwrap();
assert_eq!(result["count"], 1);
assert_eq!(result["results"][0]["id"], json!(public.id));
}
fn mirror_memory(
store: &Arc<MemoryStore>,
mem: &Memory,
session: Option<uuid::Uuid>,
sequence: u64,
) {
use wm_core::EpisodicCapturePolicy;
let record = EpisodicRecord::new(
session,
sequence,
EpisodicKind::Observation,
mem.content.clone(),
Provenance::new(ProvenanceSource::User),
)
.with_id(mem.metadata.id)
.with_visibility(mem.metadata.is_private, mem.metadata.model_exclude);
store
.episodic()
.append_explicit(&record, EpisodicCapturePolicy::explicit_only())
.unwrap();
}
fn default_search_tool(
store: Arc<MemoryStore>,
search: Option<Arc<SearchEngine>>,
) -> MemoryHybridRecallTool {
MemoryHybridRecallTool::as_search(store, search, None)
}
#[tokio::test]
async fn default_route_prefers_episodic_and_discloses_mode() {
let (_dir, store, search) = hybrid_fixture();
let needle = Memory::new(
Galaxy::Codex,
"Kotlin coroutine budget meeting notes".into(),
);
let needle_id = needle.metadata.id;
let other = Memory::new(Galaxy::Codex, "Grocery list eggs and flour".into());
store.put(Galaxy::Codex, &needle).unwrap();
store.put(Galaxy::Codex, &other).unwrap();
mirror_memory(&store, &needle, None, 1);
mirror_memory(&store, &other, None, 2);
let tool = default_search_tool(store.clone(), Some(search));
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "kotlin coroutine budget", "limit": 10}),
)
.await
.unwrap();
assert_eq!(v["recall_mode"], "episodic");
assert_eq!(v["results"][0]["source"], "episodic");
assert_eq!(v["results"][0]["id"], json!(needle_id.to_string()));
assert!(v["results"][0]["score"].as_f64().unwrap() > 0.0);
}
#[test]
fn fused_mode_label_follows_the_configured_vector_weight() {
assert_eq!(fused_mode_label(0.3), "hybrid");
assert_eq!(fused_mode_label(1.0), "hybrid");
assert_eq!(fused_mode_label(0.0), "bm25");
}
#[tokio::test]
async fn zero_vector_weight_discloses_bm25_not_hybrid() {
let (_dir, store, search) = hybrid_fixture();
let needle = Memory::new(
Galaxy::Codex,
"Kotlin coroutine budget meeting notes".into(),
);
let needle_id = needle.metadata.id;
store.put(Galaxy::Codex, &needle).unwrap();
wm_memory::reindex::rebuild_index(&store, &search, &[]).unwrap();
struct NoEmbedEmbedder;
impl wm_memory::Embedder for NoEmbedEmbedder {
fn embed_batch(&self, _texts: &[&str]) -> wm_core::Result<Vec<Vec<f32>>> {
panic!("zeroed vector weight must not call the embedder");
}
fn dimension(&self) -> usize {
16
}
fn is_available(&self) -> bool {
true
}
fn backend_name(&self) -> &'static str {
"no-embed-test"
}
}
let config = wm_memory::RecallConfig {
bm25_weight: 1.0,
vector_weight: 0.0,
importance_weight: 0.0,
..wm_memory::RecallConfig::default()
};
let recall = Arc::new(
RecallEngine::new(
store.clone(),
search.clone(),
wm_memory::VectorStore::new(),
Arc::new(NoEmbedEmbedder),
config,
)
.unwrap(),
);
let tool = MemoryHybridRecallTool::as_search(store.clone(), Some(search), Some(recall));
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "kotlin coroutine budget", "limit": 10}),
)
.await
.unwrap();
assert_eq!(v["recall_mode"], "bm25", "{v}");
assert_eq!(v["results"][0]["source"], "bm25", "{v}");
assert_eq!(v["results"][0]["id"], json!(needle_id.to_string()));
}
#[tokio::test]
async fn search_rejects_non_positive_limit() {
let (_dir, store, search) = hybrid_fixture();
index_memory(&store, &search, Galaxy::Codex, "kotlin coroutine budget");
let tool = MemoryHybridRecallTool::as_search(store, Some(search), None);
let mut ctx = Context::default();
for bad in [json!(0), json!(-1), json!("0"), json!(0.5)] {
let err = tool
.call(&mut ctx, json!({"query": "kotlin", "limit": bad}))
.await
.unwrap_err();
assert!(
err.to_string().contains("limit"),
"limit={bad} must be rejected, got: {err}"
);
}
let v = tool
.call(&mut ctx, json!({"query": "kotlin", "limit": 1}))
.await
.unwrap();
assert_eq!(v["status"], "success", "{v}");
assert_eq!(v["results"].as_array().map(Vec::len), Some(1), "{v}");
}
#[tokio::test]
async fn default_route_matches_the_episodic_machinery_ranking() {
let (_dir, store, search) = hybrid_fixture();
let contents = [
"Deployed the telemetry agent on Tuesday",
"Cancun trip booked for the twelfth",
"Telemetry agent rollout postponed to Friday",
"Deadline for the quarterly report moved",
];
let mut memories: Vec<(uuid::Uuid, &str)> = Vec::new();
for (i, content) in contents.iter().enumerate() {
let mem = Memory::new(Galaxy::Codex, (*content).to_string());
memories.push((mem.metadata.id, content));
store.put(Galaxy::Codex, &mem).unwrap();
mirror_memory(&store, &mem, None, i as u64 + 1);
}
let query = "when was the telemetry agent deployed";
let default_tool = default_search_tool(store.clone(), Some(search.clone()));
let mut ctx = Context::default();
let default_v = default_tool
.call(&mut ctx, json!({"query": query, "limit": 10}))
.await
.unwrap();
let episodic_tool = MemoryEpisodicSearchTool::new(store);
let episodic_v = episodic_tool
.call(&mut ctx, json!({"query": query, "limit": 10}))
.await
.unwrap();
assert_eq!(
default_v["results"][0]["id"], episodic_v["results"][0]["id"],
"default route must rank exactly like the episodic machinery"
);
let top = memories
.iter()
.find(|(id, _)| id.to_string() == default_v["results"][0]["id"])
.map(|(_, c)| *c)
.unwrap();
assert_eq!(top, "Deployed the telemetry agent on Tuesday");
}
#[tokio::test]
async fn default_route_falls_back_to_fts_when_episodic_yields_nothing() {
let (_dir, store, search) = hybrid_fixture();
let mem = Memory::new(Galaxy::Codex, "Zebra quotas revised upward".into());
let id = mem.metadata.id;
store.put(Galaxy::Codex, &mem).unwrap();
{
let mut writer = search.writer().unwrap();
search
.add_document(
&mut writer,
&id.to_string(),
"codex",
"Zebra quotas revised upward",
&mem.metadata.tags,
mem.metadata.created_at.timestamp(),
)
.unwrap();
search.commit(&mut writer).unwrap();
}
let tool = default_search_tool(store, Some(search));
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "zebra quotas", "limit": 10}))
.await
.unwrap();
assert_eq!(v["recall_mode"], "fts");
assert_eq!(v["results"][0]["source"], "fts");
assert_eq!(v["results"][0]["id"], json!(id.to_string()));
}
#[tokio::test]
async fn episodic_default_route_respects_galaxy_filter() {
let (_dir, store, search) = hybrid_fixture();
let in_galaxy = Memory::new(Galaxy::Codex, "Marble fountain restoration plan".into());
let other_galaxy =
Memory::new(Galaxy::Sessions, "Marble fountain restoration notes".into());
store.put(Galaxy::Codex, &in_galaxy).unwrap();
store.put(Galaxy::Sessions, &other_galaxy).unwrap();
mirror_memory(&store, &in_galaxy, None, 1);
mirror_memory(&store, &other_galaxy, None, 2);
let tool = default_search_tool(store, Some(search));
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "marble fountain", "galaxy": "sessions", "limit": 10}),
)
.await
.unwrap();
assert_eq!(v["recall_mode"], "episodic");
for r in v["results"].as_array().unwrap() {
assert_eq!(r["galaxy"], "sessions", "galaxy filter must hold");
}
assert_eq!(
v["results"][0]["id"],
json!(other_galaxy.metadata.id.to_string())
);
}
#[tokio::test]
async fn episodic_default_route_filters_private_and_stale() {
let (_dir, store, search) = hybrid_fixture();
let public = Memory::new(
Galaxy::Codex,
"Lighthouse maintenance schedule confirmed".into(),
);
let mut private = Memory::new(Galaxy::Codex, "Lighthouse access code renewal".into());
private.metadata.is_private = true;
let stale = Memory::new(Galaxy::Codex, "Lighthouse inspection legacy draft".into());
let stale_id = stale.metadata.id;
store.put(Galaxy::Codex, &public).unwrap();
store.put(Galaxy::Codex, &private).unwrap();
store.put(Galaxy::Codex, &stale).unwrap();
mirror_memory(&store, &public, None, 1);
mirror_memory(&store, &private, None, 2);
mirror_memory(&store, &stale, None, 3);
store.delete(Galaxy::Codex, stale_id).unwrap();
let tool = default_search_tool(store, Some(search));
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "lighthouse", "limit": 10}))
.await
.unwrap();
assert_eq!(v["recall_mode"], "episodic");
let ids: Vec<&str> = v["results"]
.as_array()
.unwrap()
.iter()
.filter_map(|r| r["id"].as_str())
.collect();
assert!(
!ids.contains(&private.metadata.id.to_string().as_str()),
"private memories must never surface on the default route"
);
assert!(
!ids.contains(&stale_id.to_string().as_str()),
"episodic records without a live v5 memory must be skipped"
);
assert!(!ids.is_empty(), "the public hit must still surface");
}
#[tokio::test]
async fn memory_sort_by_importance_desc() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemorySortTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"sort_by": "importance", "order": "desc"}))
.await
.unwrap();
assert_eq!(v["status"], "success");
assert_eq!(v["returned"], 3);
let mems = v["memories"].as_array().unwrap();
assert!(mems[0]["importance"].as_f64().unwrap() >= mems[1]["importance"].as_f64().unwrap());
}
#[tokio::test]
async fn memory_update_cannot_mutate_tier() {
let store = test_store();
let mem = Memory::new(Galaxy::Codex, "tier is not client-settable".into());
let id = mem.metadata.id;
store.put(Galaxy::Codex, &mem).unwrap();
let tool = MemoryUpdateTool::new(store.clone(), None);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"galaxy": "codex", "id": id.to_string(), "tier": "archival", "tags": ["x"]}),
)
.await
.unwrap();
assert_eq!(v["status"], "success");
let after = store.get(Galaxy::Codex, id).unwrap().unwrap();
assert_eq!(
after.metadata.tier,
wm_memory::Tier::Working,
"memory.update must never move the tier"
);
assert_eq!(
after.metadata.tags,
vec!["x".to_string()],
"whitelisted fields still apply"
);
}
#[tokio::test]
async fn empty_search_hints_at_populated_galaxies() {
let (_dir, store, search) = hybrid_fixture();
index_memory(&store, &search, Galaxy::Sessions, "gate plan decision");
let tool = MemoryHybridRecallTool::new(store.clone(), Some(search), None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "gate plan"}))
.await
.unwrap();
assert_eq!(
v["count"], 1,
"unfiltered search must find cross-galaxy content: {v}"
);
assert_eq!(v["galaxy"], "all");
assert_eq!(v["results"][0]["galaxy"], "sessions");
assert!(v["hint"].is_null());
let v2 = tool
.call(
&mut ctx,
json!({"query": "zzz-no-match", "galaxy": "sessions"}),
)
.await
.unwrap();
assert_eq!(v2["count"], 0);
let hint2 = v2["hint"].as_str().unwrap();
assert!(hint2.contains("no matches for this query"), "{hint2}");
let v3 = tool
.call(&mut ctx, json!({"query": "zzz-no-match"}))
.await
.unwrap();
assert_eq!(v3["count"], 0);
let hint3 = v3["hint"].as_str().expect("hint present on empty result");
assert!(hint3.contains("across all memory galaxies"), "{hint3}");
}
#[tokio::test]
async fn unfiltered_search_labels_hits_from_every_galaxy() {
let (_dir, store, search) = hybrid_fixture();
index_memory(
&store,
&search,
Galaxy::Sessions,
"lineage ledger phase four",
);
index_memory(
&store,
&search,
Galaxy::Codex,
"lineage ledger codex mirror note",
);
index_memory(&store, &search, Galaxy::Dreams, "lineage ledger dream echo");
let tool = MemoryHybridRecallTool::new(store.clone(), Some(search), None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "lineage ledger", "limit": 10}))
.await
.unwrap();
assert_eq!(v["count"], 3, "got: {v}");
let galaxies: Vec<&str> = v["results"]
.as_array()
.unwrap()
.iter()
.map(|r| r["galaxy"].as_str().unwrap())
.collect();
assert!(galaxies.contains(&"sessions"), "got: {galaxies:?}");
assert!(galaxies.contains(&"codex"), "got: {galaxies:?}");
assert!(galaxies.contains(&"dreams"), "got: {galaxies:?}");
let v2 = tool
.call(
&mut ctx,
json!({"query": "lineage ledger", "galaxy": "dreams"}),
)
.await
.unwrap();
assert_eq!(v2["count"], 1, "got: {v2}");
assert_eq!(v2["results"][0]["galaxy"], "dreams");
}
#[tokio::test]
async fn galaxy_all_alias_matches_unfiltered_search() {
let (_dir, store, search) = hybrid_fixture();
index_memory(&store, &search, Galaxy::Sessions, "alias probe session");
index_memory(&store, &search, Galaxy::Dreams, "alias probe dream");
let tool = MemoryHybridRecallTool::new(store, Some(search), None);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "alias probe", "galaxy": "all", "limit": 10}),
)
.await
.unwrap();
assert_eq!(v["count"], 2, "got: {v}");
assert_eq!(v["galaxy"], "all");
let galaxies: Vec<&str> = v["results"]
.as_array()
.unwrap()
.iter()
.map(|r| r["galaxy"].as_str().unwrap())
.collect();
assert!(galaxies.contains(&"sessions"), "got: {galaxies:?}");
assert!(galaxies.contains(&"dreams"), "got: {galaxies:?}");
let v2 = tool
.call(
&mut ctx,
json!({"query": "alias probe", "galaxy": "ALL", "limit": 10}),
)
.await
.unwrap();
assert_eq!(v2["count"], 2, "got: {v2}");
}
#[tokio::test]
async fn unfiltered_search_excludes_telemetry_unless_explicit() {
let (_dir, store, search) = hybrid_fixture();
index_memory(
&store,
&search,
Galaxy::Codex,
"telemetry probe project note",
);
index_memory(
&store,
&search,
Galaxy::Telemetry,
"telemetry probe diagnostic record",
);
let tool = MemoryHybridRecallTool::new(store, Some(search), None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "telemetry probe", "limit": 10}))
.await
.unwrap();
assert_eq!(v["count"], 1, "unfiltered search must skip telemetry: {v}");
assert_eq!(v["results"][0]["galaxy"], "codex");
let v2 = tool
.call(
&mut ctx,
json!({"query": "telemetry probe", "galaxy": "telemetry", "limit": 10}),
)
.await
.unwrap();
assert_eq!(v2["count"], 1, "explicit telemetry must still work: {v2}");
assert_eq!(v2["results"][0]["galaxy"], "telemetry");
}
#[tokio::test]
async fn successful_search_carries_no_hint() {
let (_dir, store, search) = hybrid_fixture();
index_memory(&store, &search, Galaxy::Sessions, "gate plan decision");
let tool = MemoryHybridRecallTool::new(store, Some(search), None);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "gate plan", "galaxy": "sessions"}),
)
.await
.unwrap();
assert_eq!(v["count"], 1);
assert!(v["hint"].is_null());
}
#[tokio::test]
async fn associative_expansion_surfaces_linked_memory() {
let dir = tempfile::tempdir().unwrap();
let store = Arc::new(MemoryStore::open_default(dir.path()).unwrap());
let tantivy_dir = dir.path().join("tantivy");
std::fs::create_dir_all(&tantivy_dir).unwrap();
let search = Arc::new(SearchEngine::open(&tantivy_dir).unwrap());
index_memory(
&store,
&search,
Galaxy::Codex,
"gate plan for the v7 alpha release",
);
let mut linked = Memory::new(
Galaxy::Codex,
"backup automation runs nightly at 03:30".into(),
);
linked.metadata.importance = 0.7;
let linked_id = linked.metadata.id;
store.put(Galaxy::Codex, &linked).unwrap();
search
.writer()
.and_then(|mut w| {
search.add_document(
&mut w,
&linked_id.to_string(),
"codex",
&linked.content,
&linked.metadata.tags,
linked.metadata.created_at.timestamp(),
)?;
search.commit(&mut w)
})
.unwrap();
let assoc = Association::new(
find_id(&store, "gate plan"),
linked_id,
LinkType::Extends,
0.8,
);
let associations = Arc::new(AssociationStore::open(store.env()).unwrap());
associations.put(store.env(), &assoc).unwrap();
let tool = MemoryHybridRecallTool::as_search(store.clone(), Some(search), None)
.with_associations(Some(associations));
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "gate plan alpha release"}))
.await
.unwrap();
assert_eq!(v["count"], 2, "direct hit + associated memory: {v}");
let assoc_hit = v["results"]
.as_array()
.unwrap()
.iter()
.find(|r| r["source"] == "association")
.expect("association-sourced result present");
assert_eq!(assoc_hit["id"], json!(linked_id.to_string()));
assert_eq!(assoc_hit["link_type"], "extends");
assert!(assoc_hit["via"].is_string());
assert!(assoc_hit["weight"].as_f64().unwrap() > 0.7);
}
fn find_id(store: &MemoryStore, needle: &str) -> uuid::Uuid {
store
.scan(Galaxy::Codex, 100)
.unwrap()
.into_iter()
.find(|m| m.content.contains(needle))
.map(|m| m.metadata.id)
.unwrap()
}
#[tokio::test]
async fn associative_expansion_skips_private_and_dedupes() {
let dir = tempfile::tempdir().unwrap();
let store = Arc::new(MemoryStore::open_default(dir.path()).unwrap());
let mut a = Memory::new(Galaxy::Codex, "quarterly revenue planning notes".into());
a.metadata.importance = 0.8;
let a_id = a.metadata.id;
store.put(Galaxy::Codex, &a).unwrap();
let mut private = Memory::new(Galaxy::Codex, "private salary bands".into());
private.metadata.is_private = true;
private.metadata.importance = 0.8;
store.put(Galaxy::Codex, &private).unwrap();
let mut b = Memory::new(Galaxy::Codex, "hiring plan for next quarter".into());
b.metadata.importance = 0.7;
let b_id = b.metadata.id;
store.put(Galaxy::Codex, &b).unwrap();
let associations = Arc::new(AssociationStore::open(store.env()).unwrap());
associations
.put(
store.env(),
&Association::new(a_id, private.metadata.id, LinkType::Related, 0.9),
)
.unwrap();
associations
.put(
store.env(),
&Association::new(a_id, b_id, LinkType::Related, 0.9),
)
.unwrap();
associations
.put(
store.env(),
&Association::new(b_id, a_id, LinkType::Related, 0.9),
)
.unwrap();
let tool = MemoryHybridRecallTool::as_search(store.clone(), None, None)
.with_associations(Some(associations.clone()));
let mut ctx = Context::default();
let _ = &tool;
let tantivy_dir = dir.path().join("tantivy");
std::fs::create_dir_all(&tantivy_dir).unwrap();
let search = Arc::new(SearchEngine::open(&tantivy_dir).unwrap());
for (content, id) in [
("quarterly revenue planning notes", a_id),
("private salary bands", private.metadata.id),
("hiring plan for next quarter", b_id),
] {
search
.writer()
.and_then(|mut w| {
search.add_document(&mut w, &id.to_string(), "codex", content, &[], 0)?;
search.commit(&mut w)
})
.unwrap();
}
let tool = MemoryHybridRecallTool::as_search(store.clone(), Some(search), None)
.with_associations(Some(associations));
let v = tool
.call(&mut ctx, json!({"query": "quarterly revenue planning"}))
.await
.unwrap();
let ids: Vec<&str> = v["results"]
.as_array()
.unwrap()
.iter()
.filter_map(|r| r["id"].as_str())
.collect();
assert!(
!ids.iter().any(|id| *id == private.metadata.id.to_string()),
"private memory must not surface via association: {ids:?}"
);
assert_eq!(
ids.iter().filter(|id| **id == b_id.to_string()).count(),
1,
"neighbor linked both directions appears exactly once: {ids:?}"
);
}
#[tokio::test]
async fn memory_update_content_recomputes_hash() {
let store = test_store();
let mem = Memory::new(Galaxy::Codex, "original text".into());
store.put(Galaxy::Codex, &mem).unwrap();
let id = mem.metadata.id;
let original_hash = mem.metadata.content_hash.clone();
let tool = MemoryUpdateTool::new(store.clone(), None);
let v = tool
.call(
&mut Context::default(),
json!({"galaxy": "codex", "id": id.to_string(), "content": "changed text"}),
)
.await
.unwrap();
assert_eq!(v["status"], "success");
let stored = store.get(Galaxy::Codex, id).unwrap().unwrap();
assert_eq!(stored.content, "changed text");
assert_eq!(
stored.metadata.content_hash,
wm_memory::content_hash("changed text")
);
assert_ne!(stored.metadata.content_hash, original_hash);
}
#[tokio::test]
async fn memory_update_discloses_hash_timeline() {
let store = test_store();
let mem = Memory::new(Galaxy::Codex, "original text".into());
store.put(Galaxy::Codex, &mem).unwrap();
let id = mem.metadata.id;
let original_hash = mem.metadata.content_hash.clone();
let tool = MemoryUpdateTool::new(store.clone(), None);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"galaxy": "codex", "id": id.to_string(), "content": "changed text"}),
)
.await
.unwrap();
assert_eq!(
v["content_hash"],
json!(wm_memory::content_hash("changed text"))
);
assert_eq!(v["prev_content_hash"], json!(original_hash));
let v = tool
.call(
&mut ctx,
json!({"galaxy": "codex", "id": id.to_string(), "tags": ["amended"]}),
)
.await
.unwrap();
assert_eq!(
v["content_hash"],
json!(wm_memory::content_hash("changed text"))
);
assert!(v.get("prev_content_hash").is_none());
}
#[tokio::test]
async fn memory_update_appends_revision_chain() {
let store = test_store();
let mem = Memory::new(Galaxy::Codex, "original text".into());
store.put(Galaxy::Codex, &mem).unwrap();
let id = mem.metadata.id;
let tool = MemoryUpdateTool::new(store.clone(), None);
let mut ctx = Context {
user_id: Some("agent-b".to_string()),
session_id: Some(uuid::Uuid::nil()),
compartment: Some("production".to_string()),
..Default::default()
};
let v = tool
.call(
&mut ctx,
json!({"galaxy": "codex", "id": id.to_string(), "content": "second text"}),
)
.await
.unwrap();
assert_eq!(v["revision"]["seq"], 0);
assert_eq!(
v["revision"]["old_hash"],
json!(wm_memory::content_hash("original text"))
);
assert_eq!(
v["revision"]["new_hash"],
json!(wm_memory::content_hash("second text"))
);
let v = tool
.call(
&mut ctx,
json!({"galaxy": "codex", "id": id.to_string(), "content": "third text"}),
)
.await
.unwrap();
assert_eq!(v["revision"]["seq"], 1);
let revisions = store.revisions(Galaxy::Codex, id).unwrap();
assert_eq!(revisions.len(), 2);
assert_eq!(revisions[1].old_hash, revisions[0].new_hash, "chain links");
assert_eq!(revisions[0].actor_user.as_deref(), Some("agent-b"));
assert_eq!(
revisions[0].actor_compartment.as_deref(),
Some("production")
);
assert_eq!(
revisions[0].actor_session.as_deref(),
Some(uuid::Uuid::nil().to_string().as_str())
);
let stored = store.get(Galaxy::Codex, id).unwrap().unwrap();
assert_eq!(stored.metadata.revision_count, 2);
let report = store
.verify_revision_chain(Galaxy::Codex, id, &stored.metadata.content_hash)
.unwrap();
assert!(report.valid, "{:?}", report.breaks);
assert!(report.matches_head);
}
#[tokio::test]
async fn memory_update_out_of_band_edit_breaks_chain() {
let store = test_store();
let mem = Memory::new(Galaxy::Codex, "original text".into());
store.put(Galaxy::Codex, &mem).unwrap();
let id = mem.metadata.id;
let tool = MemoryUpdateTool::new(store.clone(), None);
tool.call(
&mut Context::default(),
json!({"galaxy": "codex", "id": id.to_string(), "content": "second text"}),
)
.await
.unwrap();
let mut row = store.get(Galaxy::Codex, id).unwrap().unwrap();
row.content = "smuggled text".to_string();
row.metadata.content_hash = wm_memory::content_hash("smuggled text");
store.put(Galaxy::Codex, &row).unwrap();
let report = store
.verify_revision_chain(Galaxy::Codex, id, &row.metadata.content_hash)
.unwrap();
assert!(!report.valid);
assert!(!report.matches_head);
assert!(report.breaks.iter().any(|b| b.contains("head mismatch")));
}
#[tokio::test]
async fn memory_revisions_tool_list_and_verify() {
let store = test_store();
let mem = Memory::new(Galaxy::Codex, "v1".into());
store.put(Galaxy::Codex, &mem).unwrap();
let id = mem.metadata.id;
let update = MemoryUpdateTool::new(store.clone(), None);
update
.call(
&mut Context::default(),
json!({"galaxy": "codex", "id": id.to_string(), "content": "v2"}),
)
.await
.unwrap();
let tool = MemoryRevisionsTool::new(store.clone());
let v = tool
.call(&mut Context::default(), json!({"id": id.to_string()}))
.await
.unwrap();
assert_eq!(v["action"], "list");
assert_eq!(v["count"], 1);
let v = tool
.call(
&mut Context::default(),
json!({"id": id.to_string(), "action": "verify"}),
)
.await
.unwrap();
assert_eq!(v["valid"], true);
assert_eq!(v["entries"], 1);
store
.record_revision(
Galaxy::Codex,
id,
"forged_old_hash",
&wm_memory::content_hash("v2"),
wm_memory::RevisionActor::default(),
)
.unwrap();
let v = tool
.call(
&mut Context::default(),
json!({"id": id.to_string(), "action": "verify"}),
)
.await
.unwrap();
assert_eq!(v["valid"], false);
let breaks: Vec<String> = v["breaks"]
.as_array()
.unwrap()
.iter()
.map(|b| b.as_str().unwrap().to_string())
.collect();
assert!(
breaks.iter().any(|b| b.contains("hash-linkage")),
"{breaks:?}"
);
}
#[tokio::test]
async fn memory_update_applies_importance_verbatim() {
let store = test_store();
let tel = Memory::new(
Galaxy::Codex,
"## Auto-logged Friction: dispatch error\n\nbody".into(),
);
store.put(Galaxy::Codex, &tel).unwrap();
let tool = MemoryUpdateTool::new(store.clone(), None);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"galaxy": "codex", "id": tel.metadata.id.to_string(), "importance": 0.9}),
)
.await
.unwrap();
assert!(v.get("class_policy").is_none());
assert!(v.get("write_gate").is_none());
let stored = store.get(Galaxy::Codex, tel.metadata.id).unwrap().unwrap();
assert!((stored.metadata.importance - 0.9).abs() < 1e-5);
}
#[tokio::test]
async fn memory_search_min_trust_filter_drops_low_trust() {
let (_dir, store, search) = hybrid_fixture();
let mut confirmed = Memory::new(Galaxy::Codex, "Quantum foal registry minutes".into());
confirmed.metadata.source_trust = 1.0;
confirmed.metadata.source = "user".to_string();
let mut ingested = Memory::new(Galaxy::Codex, "Quantum foal registry draft".into());
ingested.metadata.source_trust = 0.7;
ingested.metadata.source = "tool".to_string();
store.put(Galaxy::Codex, &confirmed).unwrap();
store.put(Galaxy::Codex, &ingested).unwrap();
mirror_memory(&store, &confirmed, None, 1);
mirror_memory(&store, &ingested, None, 2);
let tool = default_search_tool(store, Some(search));
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "quantum foal registry", "limit": 10}),
)
.await
.unwrap();
assert_eq!(v["count"], 2, "no floor: both results surface");
assert!(v.get("min_trust").is_none());
let v = tool
.call(
&mut ctx,
json!({"query": "quantum foal registry", "limit": 10, "min_trust": 0.9}),
)
.await
.unwrap();
assert_eq!(v["min_trust"], 0.9);
assert_eq!(v["min_trust_filtered"], 1);
let trusts: Vec<f64> = v["results"]
.as_array()
.unwrap()
.iter()
.map(|r| r["trust"].as_f64().unwrap())
.collect();
assert!(trusts.iter().all(|t| *t >= 0.9), "{trusts:?}");
}
#[tokio::test]
async fn memory_sort_by_importance_asc() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemorySortTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"sort_by": "importance", "order": "asc"}))
.await
.unwrap();
let mems = v["memories"].as_array().unwrap();
assert!(mems[0]["importance"].as_f64().unwrap() <= mems[1]["importance"].as_f64().unwrap());
}
#[tokio::test]
async fn memory_sort_by_recency() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemorySortTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"sort_by": "recency"}))
.await
.unwrap();
assert_eq!(v["returned"], 3);
}
#[tokio::test]
async fn memory_sort_invalid_field() {
let store = test_store();
let tool = MemorySortTool::new(store);
let mut ctx = Context::default();
let result = tool.call(&mut ctx, json!({"sort_by": "invalid"})).await;
assert!(result.is_err());
}
#[tokio::test]
async fn memory_sort_with_limit() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemorySortTool::new(store);
let mut ctx = Context::default();
let v = tool.call(&mut ctx, json!({"limit": 2})).await.unwrap();
assert_eq!(v["returned"], 2);
assert_eq!(v["total"], 3);
}
#[tokio::test]
async fn memory_filter_by_tag() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemoryFilterTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"tags": ["rust"]}))
.await
.unwrap();
assert_eq!(v["matched"], 2);
}
#[tokio::test]
async fn memory_filter_by_importance_range() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemoryFilterTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"min_importance": 0.4, "max_importance": 0.6}),
)
.await
.unwrap();
assert_eq!(v["matched"], 1);
}
#[tokio::test]
async fn memory_filter_query_matches_content_and_title() {
let store = test_store();
let tool = MemoryFilterTool::new(store.clone());
let mut ctx = Context::default();
let mut titled = Memory::new(Galaxy::Codex, "unrelated body text".into());
titled.metadata.title = Some("Rust Borrow Checker".into());
let plain = Memory::new(Galaxy::Codex, "rust ownership rules".into());
let other = Memory::new(Galaxy::Codex, "gardening tips".into());
for m in [&titled, &plain, &other] {
store.put(Galaxy::Codex, m).unwrap();
}
let v = tool.call(&mut ctx, json!({"query": "RUST"})).await.unwrap();
assert_eq!(v["matched"], 2, "content hit + title hit: {v}");
assert_eq!(v["filters"]["query_terms"], json!(["rust"]));
let v = tool
.call(&mut ctx, json!({"query": "rust borrow"}))
.await
.unwrap();
assert_eq!(v["matched"], 1, "all terms must match: {v}");
assert_eq!(v["memories"][0]["content"], "unrelated body text");
}
#[tokio::test]
async fn memory_filter_no_matches() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemoryFilterTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"tags": ["nonexistent"]}))
.await
.unwrap();
assert_eq!(v["matched"], 0);
}
#[tokio::test]
async fn memory_filter_combined_tags_and_importance() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemoryFilterTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"tags": ["rust"], "min_importance": 0.5}))
.await
.unwrap();
assert_eq!(v["matched"], 1);
}
#[tokio::test]
async fn memory_filter_offset_exclude_tags_and_date_range() {
let store = test_store();
let tool = MemoryFilterTool::new(store.clone());
let mut ctx = Context::default();
let mut recent_a = Memory::new(Galaxy::Codex, "recent a".into());
recent_a.metadata.created_at = chrono::Utc::now() - chrono::Duration::hours(2);
let mut recent_b = Memory::new(Galaxy::Codex, "recent b".into());
recent_b.metadata.created_at = chrono::Utc::now() - chrono::Duration::hours(1);
recent_b.metadata.tags = vec!["noise".into()];
let mut recent_priv = Memory::new(Galaxy::Codex, "recent private".into());
recent_priv.metadata.created_at = chrono::Utc::now() - chrono::Duration::minutes(90);
recent_priv.metadata.is_private = true;
let mut old = Memory::new(Galaxy::Codex, "old relic".into());
old.metadata.created_at = chrono::Utc::now() - chrono::Duration::days(60);
for m in [&recent_a, &recent_b, &recent_priv, &old] {
store.put(Galaxy::Codex, m).unwrap();
}
let cutoff = (chrono::Utc::now() - chrono::Duration::days(1))
.to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
let v = tool
.call(
&mut ctx,
json!({
"galaxy": "codex",
"created_after": cutoff,
"exclude_tags": ["noise"],
}),
)
.await
.unwrap();
assert_eq!(v["matched"], 1, "only recent-a is visible in range: {v}");
assert_eq!(v["returned"], 1);
assert_eq!(v["memories"][0]["content"], "recent a");
assert_eq!(v["filters"]["exclude_tags"], json!(["noise"]));
assert!(v["filters"]["created_after"].is_string());
let page2 = tool
.call(
&mut ctx,
json!({"galaxy": "codex", "offset": 3, "limit": 2}),
)
.await
.unwrap();
assert_eq!(
page2["matched"], 3,
"private memory must not count: {page2}"
);
assert_eq!(
page2["returned"], 0,
"offset past the match set is an honest empty page"
);
assert_eq!(page2["offset"], 3);
let bad = tool
.call(
&mut ctx,
json!({"galaxy": "codex", "created_before": "yesterday"}),
)
.await;
assert!(bad.is_err(), "non-RFC-3339 bound must be refused");
}
#[tokio::test]
async fn memory_deduplicate_hash_dry_run() {
let store = test_store();
let m1 = Memory::new(Galaxy::Codex, "duplicate content".into());
let m2 = Memory::new(Galaxy::Codex, "duplicate content".into());
let _ = store.put(Galaxy::Codex, &m1);
let _ = store.put(Galaxy::Codex, &m2);
let _ = store.put(
Galaxy::Codex,
&Memory::new(Galaxy::Codex, "unique content".into()),
);
let tool = MemoryDeduplicateTool::new(store.clone(), None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"mode": "hash", "dry_run": true}))
.await
.unwrap();
assert_eq!(v["duplicates_found"], 1);
assert_eq!(v["removed"], 0);
let memories = store.scan(Galaxy::Codex, 100).unwrap();
assert_eq!(memories.len(), 3);
}
#[tokio::test]
async fn memory_deduplicate_hash_execute() {
let store = test_store();
let m1 = Memory::new(Galaxy::Codex, "duplicate content".into());
let m2 = Memory::new(Galaxy::Codex, "duplicate content".into());
let _ = store.put(Galaxy::Codex, &m1);
let _ = store.put(Galaxy::Codex, &m2);
let _ = store.put(
Galaxy::Codex,
&Memory::new(Galaxy::Codex, "unique content".into()),
);
let tool = MemoryDeduplicateTool::new(store.clone(), None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"mode": "hash", "dry_run": false}))
.await
.unwrap();
assert_eq!(v["duplicates_found"], 1);
assert_eq!(v["removed"], 1);
let memories = store.scan(Galaxy::Codex, 100).unwrap();
assert_eq!(memories.len(), 2);
}
#[tokio::test]
async fn memory_deduplicate_deindexes_removed_memories() {
let (_dir, store, search) = hybrid_fixture();
let m1 = Memory::new(Galaxy::Codex, "index drift duplicate".into());
let m2 = Memory::new(Galaxy::Codex, "index drift duplicate".into());
let id1 = m1.metadata.id;
let id2 = m2.metadata.id;
let _ = store.put(Galaxy::Codex, &m1);
let _ = store.put(Galaxy::Codex, &m2);
for mem in [&m1, &m2] {
let mut writer = search.writer().unwrap();
search
.add_document(
&mut writer,
&mem.metadata.id.to_string(),
mem.metadata.galaxy.db_name(),
&mem.content,
&mem.metadata.tags,
mem.metadata.created_at.timestamp(),
)
.unwrap();
search.commit(&mut writer).unwrap();
}
let before = search.search_ids("index drift", 100).unwrap();
assert_eq!(before.len(), 2);
let tool = MemoryDeduplicateTool::new(store.clone(), Some(search.clone()));
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"mode": "hash", "dry_run": false}))
.await
.unwrap();
assert_eq!(v["removed"], 1);
let after = search.search_ids("index drift", 100).unwrap();
assert_eq!(after.len(), 1, "search index should only contain survivors");
assert!(
after.contains(&id1) || after.contains(&id2),
"survivor should be one of the original memories"
);
}
#[tokio::test]
async fn memory_deduplicate_content_mode() {
let store = test_store();
let m1 = Memory::new(Galaxy::Codex, "same text".into());
let m2 = Memory::new(Galaxy::Codex, "same text".into());
let _ = store.put(Galaxy::Codex, &m1);
let _ = store.put(Galaxy::Codex, &m2);
let tool = MemoryDeduplicateTool::new(store, None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"mode": "content", "dry_run": true}))
.await
.unwrap();
assert_eq!(v["duplicates_found"], 1);
}
#[tokio::test]
async fn memory_deduplicate_no_duplicates() {
let store = test_store();
let _ = store.put(
Galaxy::Codex,
&Memory::new(Galaxy::Codex, "content a".into()),
);
let _ = store.put(
Galaxy::Codex,
&Memory::new(Galaxy::Codex, "content b".into()),
);
let tool = MemoryDeduplicateTool::new(store, None);
let mut ctx = Context::default();
let v = tool.call(&mut ctx, json!({})).await.unwrap();
assert_eq!(v["duplicates_found"], 0);
}
#[tokio::test]
async fn memory_deduplicate_invalid_mode() {
let store = test_store();
let tool = MemoryDeduplicateTool::new(store, None);
let mut ctx = Context::default();
let result = tool.call(&mut ctx, json!({"mode": "invalid"})).await;
assert!(result.is_err());
}
#[tokio::test]
async fn memory_export_json() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemoryExportTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"format": "json"}))
.await
.unwrap();
assert_eq!(v["format"], "json");
assert_eq!(v["count"], 3);
assert!(v["export"].as_str().unwrap().contains("First memory"));
}
#[tokio::test]
async fn memory_export_csv() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemoryExportTool::new(store);
let mut ctx = Context::default();
let v = tool.call(&mut ctx, json!({"format": "csv"})).await.unwrap();
let csv = v["export"].as_str().unwrap();
assert!(csv.contains("id,content,tags"));
assert!(csv.contains("First memory"));
}
#[tokio::test]
async fn memory_export_markdown() {
let store = test_store();
populate_memories(&store, Galaxy::Codex);
let tool = MemoryExportTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"format": "markdown"}))
.await
.unwrap();
let md = v["export"].as_str().unwrap();
assert!(md.contains("# Memory Export"));
assert!(md.contains("First memory"));
}
#[tokio::test]
async fn memory_export_invalid_format() {
let store = test_store();
let tool = MemoryExportTool::new(store);
let mut ctx = Context::default();
let result = tool.call(&mut ctx, json!({"format": "xml"})).await;
assert!(result.is_err());
}
#[tokio::test]
async fn memory_export_empty_galaxy() {
let store = test_store();
let tool = MemoryExportTool::new(store);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"format": "json"}))
.await
.unwrap();
assert_eq!(v["count"], 0);
}
#[tokio::test]
async fn memory_sort_and_filter_are_winnowing_basket_gana() {
let store = test_store();
assert_eq!(
MemorySortTool::new(store.clone()).gana(),
Gana::WinnowingBasket
);
assert_eq!(
MemoryFilterTool::new(store.clone()).gana(),
Gana::WinnowingBasket
);
assert_eq!(
MemoryDeduplicateTool::new(store.clone(), None).gana(),
Gana::WinnowingBasket
);
assert_eq!(MemoryExportTool::new(store).gana(), Gana::WinnowingBasket);
}
#[test]
fn hybrid_recall_routes_expose_query_schema() {
let dir = tempfile::tempdir().unwrap();
let store = Arc::new(MemoryStore::open_default(dir.path()).unwrap());
for tool in [
MemoryHybridRecallTool::new(store.clone(), None, None),
MemoryHybridRecallTool::as_search(store, None, None),
] {
let schema = tool.input_schema();
assert_eq!(schema["type"], "object");
assert!(schema["properties"].get("query").is_some());
assert_eq!(schema["required"], json!(["query"]));
}
}
fn hybrid_fixture() -> (tempfile::TempDir, Arc<MemoryStore>, Arc<SearchEngine>) {
let dir = tempfile::tempdir().unwrap();
let store = Arc::new(MemoryStore::open_default(dir.path()).unwrap());
let tantivy_dir = dir.path().join("tantivy");
std::fs::create_dir_all(&tantivy_dir).unwrap();
let search = Arc::new(SearchEngine::open(&tantivy_dir).unwrap());
(dir, store, search)
}
fn index_memory(
store: &Arc<MemoryStore>,
search: &Arc<SearchEngine>,
galaxy: Galaxy,
content: &str,
) {
let mem = Memory::new(galaxy, content.to_string());
let id = mem.metadata.id;
store.put(galaxy, &mem).unwrap();
let mut writer = search.writer().unwrap();
search
.add_document(
&mut writer,
&id.to_string(),
galaxy.db_name(),
content,
&mem.metadata.tags,
mem.metadata.created_at.timestamp(),
)
.unwrap();
search.commit(&mut writer).unwrap();
}
#[tokio::test]
async fn hybrid_recall_excludes_private_memories() {
let (_dir, store, search) = hybrid_fixture();
let mut priv_mem = Memory::new(Galaxy::Codex, "private secret plan alpha".to_string());
priv_mem.metadata.is_private = true;
let id = priv_mem.metadata.id;
store.put(Galaxy::Codex, &priv_mem).unwrap();
{
let mut writer = search.writer().unwrap();
search
.add_document(
&mut writer,
&id.to_string(),
"codex",
"private secret plan alpha",
&[],
priv_mem.metadata.created_at.timestamp(),
)
.unwrap();
search.commit(&mut writer).unwrap();
}
index_memory(
&store,
&search,
Galaxy::Codex,
"public plan alpha documentation",
);
let tool = MemoryHybridRecallTool::new(store.clone(), Some(search.clone()), None);
let v = tool
.call(
&mut Context::default(),
json!({"query": "plan alpha", "galaxy": "codex"}),
)
.await
.unwrap();
let results = v["results"].as_array().unwrap();
let contents: Vec<&str> = results
.iter()
.filter_map(|r| r["content"].as_str())
.collect();
assert!(
!contents.iter().any(|c| c.contains("private")),
"private memory leaked through hybrid recall: {results:?}"
);
assert!(
contents.iter().any(|c| c.contains("public")),
"public memory missing from hybrid recall: {results:?}"
);
}
#[tokio::test]
async fn batch_read_treats_private_as_miss() {
let store = test_store();
let mut priv_mem = Memory::new(Galaxy::Codex, "private batch note".into());
priv_mem.metadata.is_private = true;
let priv_id = priv_mem.metadata.id;
store.put(Galaxy::Codex, &priv_mem).unwrap();
let pub_mem = Memory::new(Galaxy::Codex, "public batch note".into());
let pub_id = pub_mem.metadata.id;
store.put(Galaxy::Codex, &pub_mem).unwrap();
let tool = MemoryBatchReadTool::new(store);
let v = tool
.call(
&mut Context::default(),
json!({"galaxy": "codex", "ids": [priv_id.to_string(), pub_id.to_string()]}),
)
.await
.unwrap();
assert_eq!(v["found"], 1);
assert_eq!(v["misses"], 1);
assert!(
!v["memories"]
.as_array()
.unwrap()
.iter()
.any(|m| m["content"].as_str().unwrap_or("").contains("private")),
"private memory leaked through batch_read: {v}"
);
}
#[tokio::test]
async fn hybrid_recall_incident_query_returns_only_relevant() {
let (_dir, store, search) = hybrid_fixture();
index_memory(
&store,
&search,
Galaxy::Codex,
"smoke test from wmClient: verify recall",
);
index_memory(
&store,
&search,
Galaxy::Codex,
"NES Evolution and Impact: a history of the console wars",
);
index_memory(
&store,
&search,
Galaxy::Codex,
"Insights on The Gateless Gate: koans and zen practice",
);
let tool = MemoryHybridRecallTool::new(store, Some(search), None);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "smoke test", "galaxy": "codex", "limit": 5}),
)
.await
.unwrap();
let results = v["results"].as_array().unwrap();
assert_eq!(
results.len(),
1,
"incident query must not return unrelated memories: {results:?}"
);
let hit = &results[0];
assert_eq!(hit["source"], "fts");
assert!(
hit["content"]
.as_str()
.unwrap()
.contains("smoke test from wmClient")
);
assert!(hit["normalized_score"].as_f64().unwrap() > 0.0);
assert_eq!(v["count"], 1);
}
#[tokio::test]
async fn hybrid_recall_filters_stale_index_entries() {
let (_dir, store, search) = hybrid_fixture();
index_memory(
&store,
&search,
Galaxy::Codex,
"rust memory about ownership",
);
{
let mut writer = search.writer().unwrap();
search
.add_document(
&mut writer,
"99999999-9999-9999-9999-999999999999",
"codex",
"rust ghost memory",
&[],
1700000000,
)
.unwrap();
search.commit(&mut writer).unwrap();
}
let tool = MemoryHybridRecallTool::new(store, Some(search), None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "rust", "limit": 10}))
.await
.unwrap();
let results = v["results"].as_array().unwrap();
assert_eq!(results.len(), 1);
assert_ne!(
results[0]["id"].as_str().unwrap(),
"99999999-9999-9999-9999-999999999999"
);
}
#[tokio::test]
async fn hybrid_recall_respects_min_score_arg() {
let (_dir, store, search) = hybrid_fixture();
index_memory(&store, &search, Galaxy::Codex, "alpha");
let filler = format!("alpha {}", "zzz ".repeat(400));
index_memory(&store, &search, Galaxy::Codex, &filler);
let tool = MemoryHybridRecallTool::new(store.clone(), Some(search.clone()), None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "alpha", "limit": 10}))
.await
.unwrap();
assert_eq!(v["count"], 2);
let scores: Vec<f64> = v["results"]
.as_array()
.unwrap()
.iter()
.map(|r| r["score"].as_f64().unwrap())
.collect();
let lo = scores.iter().copied().fold(f64::MAX, f64::min);
let hi = scores.iter().copied().fold(0.0, f64::max);
let mid = f64::midpoint(hi, lo);
let v = tool
.call(
&mut ctx,
json!({"query": "alpha", "limit": 10, "min_score": mid}),
)
.await
.unwrap();
assert_eq!(v["count"], 1);
assert!((v["results"][0]["score"].as_f64().unwrap() - hi).abs() < 1e-3);
}
#[tokio::test]
async fn hybrid_recall_or_coverage_finds_partial_matches() {
let (_dir, store, search) = hybrid_fixture();
index_memory(&store, &search, Galaxy::Codex, "alpha only here");
index_memory(&store, &search, Galaxy::Codex, "alpha beta gamma delta");
let tool = MemoryHybridRecallTool::new(store, Some(search), None);
let mut ctx = Context::default();
let v = tool
.call(&mut ctx, json!({"query": "alpha beta gamma", "limit": 10}))
.await
.unwrap();
let results = v["results"].as_array().unwrap();
assert_eq!(results.len(), 1);
assert!(
results[0]["content"]
.as_str()
.unwrap()
.contains("alpha beta gamma")
);
let v = tool
.call(
&mut ctx,
json!({"query": "alpha beta gamma zeta", "limit": 10}),
)
.await
.unwrap();
let results = v["results"].as_array().unwrap();
assert_eq!(
results.len(),
1,
"OR + coverage must require 2/4 token coverage: {results:?}"
);
assert!(
results[0]["content"]
.as_str()
.unwrap()
.contains("alpha beta gamma")
);
for r in results {
assert!(
matches!(r["source"].as_str(), Some("fts")),
"results should be tagged fts"
);
}
}
fn index_tagged_memory(
store: &Arc<MemoryStore>,
search: &Arc<SearchEngine>,
galaxy: Galaxy,
content: &str,
tags: &[&str],
) {
let mut mem = Memory::new(galaxy, content.to_string());
mem.metadata.tags = tags.iter().map(ToString::to_string).collect();
let id = mem.metadata.id;
store.put(galaxy, &mem).unwrap();
let mut writer = search.writer().unwrap();
search
.add_document(
&mut writer,
&id.to_string(),
galaxy.db_name(),
content,
&mem.metadata.tags,
mem.metadata.created_at.timestamp(),
)
.unwrap();
search.commit(&mut writer).unwrap();
}
fn aggregate_fixture() -> (tempfile::TempDir, Arc<MemoryStore>, Arc<SearchEngine>) {
let (dir, store, search) = hybrid_fixture();
index_tagged_memory(
&store,
&search,
Galaxy::Codex,
"I started learning Rust.",
&["user", "session_002"],
);
index_tagged_memory(
&store,
&search,
Galaxy::Codex,
"I finished my first Rust project, a CLI tool.",
&["user", "session_007"],
);
index_tagged_memory(
&store,
&search,
Galaxy::Codex,
"I got a job as a systems engineer using Rust.",
&["user", "session_012"],
);
index_tagged_memory(
&store,
&search,
Galaxy::Codex,
"I started learning Go.",
&["user", "session_003"],
);
index_tagged_memory(
&store,
&search,
Galaxy::Codex,
"I got a job as a backend engineer using Go.",
&["user", "session_015"],
);
(dir, store, search)
}
#[tokio::test]
async fn aggregate_session_span_isolated_by_rarest_term() {
let (_dir, store, search) = aggregate_fixture();
let tool = MemoryAggregateTool::new(Some(search), store);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({
"query": "How long did it take from starting Rust to getting a job using it?",
"metric": "session_span",
}),
)
.await
.unwrap();
assert_eq!(v["aggregate"]["value"], 10, "session_012 - session_002");
assert_eq!(v["aggregate"]["unit"], "sessions");
assert_eq!(v["aggregate"]["content"], "10 sessions");
}
#[tokio::test]
async fn aggregate_session_count() {
let (_dir, store, search) = aggregate_fixture();
let tool = MemoryAggregateTool::new(Some(search), store);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({
"query": "How long did it take from starting Rust to getting a job using it?",
"metric": "session_count",
}),
)
.await
.unwrap();
assert_eq!(v["aggregate"]["value"], 2);
}
#[tokio::test]
async fn aggregate_count_needs_no_session_tags() {
let (_dir, store, search) = aggregate_fixture();
let tool = MemoryAggregateTool::new(Some(search), store);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "Rust project", "metric": "count"}),
)
.await
.unwrap();
assert_eq!(v["aggregate"]["value"], 3);
}
#[tokio::test]
async fn aggregate_single_session_falls_back_honestly() {
let (_dir, store, search) = aggregate_fixture();
let tool = MemoryAggregateTool::new(Some(search), store);
let mut ctx = Context::default();
let v = tool
.call(
&mut ctx,
json!({"query": "CLI tool", "metric": "session_count"}),
)
.await
.unwrap();
assert_eq!(v["aggregate"]["value"], 1, "one session in evidence: {v}");
assert_eq!(v["anchor"], "session_tagged_fallback");
let v = tool
.call(
&mut ctx,
json!({"query": "CLI tool", "metric": "session_span"}),
)
.await
.unwrap();
assert_eq!(v["aggregate"]["value"], 0, "single point spans 0: {v}");
}
#[tokio::test]
async fn aggregate_rejects_unknown_metric() {
let (_dir, store, search) = aggregate_fixture();
let tool = MemoryAggregateTool::new(Some(search), store);
let mut ctx = Context::default();
let err = tool
.call(&mut ctx, json!({"query": "x", "metric": "median"}))
.await
.unwrap_err();
assert!(err.to_string().contains("unknown metric"));
}
#[tokio::test]
async fn hybrid_recall_rejects_out_of_range_floors() {
let (_dir, store, search) = hybrid_fixture();
let tool = MemoryHybridRecallTool::new(store, Some(search), None);
let mut ctx = Context::default();
for (key, value) in [
("min_trust", json!(2.0)),
("min_trust", json!(-0.1)),
("min_importance", json!(1.5)),
("min_importance", json!(-0.5)),
("min_score_ratio", json!(1.5)),
("min_score_ratio", json!(-1.0)),
("min_score", json!(-3.0)),
("min_trust", json!("2.0")),
] {
let err = tool
.call(&mut ctx, json!({"query": "x", key: value}))
.await
.unwrap_err();
assert!(
err.to_string().contains(key),
"{key}={value} must be rejected by name, got: {err}"
);
}
let v = tool
.call(
&mut ctx,
json!({
"query": "x",
"min_trust": 1.0,
"min_importance": 0.0,
"min_score_ratio": 0.0,
"min_score": 0.0,
}),
)
.await
.unwrap();
assert_eq!(v["min_trust"], 1.0, "valid boundary floor disclosed: {v}");
}
#[tokio::test]
async fn memory_update_rejects_out_of_range_importance() {
let store = test_store();
let mut mem = Memory::new(Galaxy::Codex, "audit validation target".into());
mem.metadata.importance = 0.5;
store.put(Galaxy::Codex, &mem).unwrap();
let id = mem.metadata.id;
let tool = MemoryUpdateTool::new(store.clone(), None);
let mut ctx = Context::default();
for bad in [json!(2.0), json!(999), json!(-0.25), json!("1.5")] {
let err = tool
.call(&mut ctx, json!({"id": id, "importance": bad}))
.await
.unwrap_err();
assert!(
err.to_string().contains("importance"),
"importance={bad} must be rejected, got: {err}"
);
}
let stored = store.get(Galaxy::Codex, id).unwrap().unwrap();
assert!(
(stored.metadata.importance - 0.5).abs() < f32::EPSILON,
"rejected updates must not mutate the record: {}",
stored.metadata.importance
);
let v = tool
.call(&mut ctx, json!({"id": id, "importance": 0.75}))
.await
.unwrap();
assert_eq!(v["status"], "success", "{v}");
let stored = store.get(Galaxy::Codex, id).unwrap().unwrap();
assert!((stored.metadata.importance - 0.75).abs() < f32::EPSILON);
}
#[test]
fn weak_evidence_abstention_floor_and_coverage() {
let fts = vec![json!({"source": "fts", "raw_score": 0.5})];
assert!(weak_evidence_abstention_with(&fts, "alpha", 0.0, 0.0).is_none());
let weak =
weak_evidence_abstention_with(&fts, "alpha", 2.0, 0.0).expect("below floor abstains");
assert_eq!(weak["reason"], "top_below_floor");
assert_eq!(weak["signal"], "bm25");
assert_eq!(weak["scope"], "retrieval");
assert_eq!(weak["status"], "insufficient_evidence");
let strong = vec![json!({"source": "hybrid", "raw_score": 4.0})];
assert!(weak_evidence_abstention_with(&strong, "alpha", 2.0, 0.0).is_none());
let vector_only = vec![json!({"source": "hybrid", "raw_score": 0.0, "vector_score": 0.9})];
assert!(weak_evidence_abstention_with(&vector_only, "alpha", 2.0, 0.0).is_none());
let episodic = vec![json!({"source": "episodic", "matched_terms": 1})];
let low = weak_evidence_abstention_with(&episodic, "alpha beta gamma delta", 0.0, 0.5)
.expect("low coverage abstains");
assert_eq!(low["reason"], "coverage_below_floor");
assert_eq!(low["signal"], "coverage");
assert_eq!(low["matched_terms"], 1);
let full = vec![json!({"source": "episodic", "matched_terms": 4})];
assert!(weak_evidence_abstention_with(&full, "alpha beta gamma delta", 0.0, 0.5).is_none());
let association = vec![json!({"source": "association", "score": 0.1})];
assert!(weak_evidence_abstention_with(&association, "alpha", 2.0, 0.5).is_none());
}
}