From 6c01dbf6b60eabee09f438c5fbfd6133e513f54d Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Thu, 15 Oct 2026 12:19:42 +0530 Subject: [PATCH 1/2] feat(cortex): evidence recall keeps the packs that arrive within 12 s Evidence-mode recall built one pack per kind scope and waited for all of them, so one slow scope cost the whole recall inside a host's tool-call limit (OpenHuman allows 15 s; openhuman-benchmarks #38 lost 2 of 13 answers that way). Packs now get 12 s from the start of the recall; a pack still building then is left out with a warning and the evidence is rendered from the rest, most specific first. No pack in time is Unavailable, never an empty answer; a failed pack still fails the recall. Co-Authored-By: Claude Opus 5.5 --- .../src/cortex/engine/engine_test_support.rs | 1 + .../src/cortex/engine/evidence.rs | 57 ++++++++++++++++--- .../src/cortex/engine/evidence_tests.rs | 44 +++++++++++++- .../src/cortex/log/mod.rs | 6 ++ .../src/cortex/testing/mod.rs | 3 + .../src/cortex/testing/routes.rs | 6 ++ docs/architecture/cortex-flows.md | 6 ++ 7 files changed, 115 insertions(+), 8 deletions(-) diff --git a/crates/tinymemory-integrations/src/cortex/engine/engine_test_support.rs b/crates/tinymemory-integrations/src/cortex/engine/engine_test_support.rs index fb860f78..8d5f78cb 100644 --- a/crates/tinymemory-integrations/src/cortex/engine/engine_test_support.rs +++ b/crates/tinymemory-integrations/src/cortex/engine/engine_test_support.rs @@ -12,6 +12,7 @@ impl CortexEngine { settle: visibility, poll: Duration::from_millis(5), erasure: Duration::from_millis(300), + evidence: visibility, }; self.log.client.set_read_backoff(Duration::from_millis(5)); self diff --git a/crates/tinymemory-integrations/src/cortex/engine/evidence.rs b/crates/tinymemory-integrations/src/cortex/engine/evidence.rs index 5f840227..2e7c288e 100644 --- a/crates/tinymemory-integrations/src/cortex/engine/evidence.rs +++ b/crates/tinymemory-integrations/src/cortex/engine/evidence.rs @@ -27,10 +27,20 @@ //! //! The citations are the conversation's items, at most `limit`; the evidence //! may hold more events than that (each pack's events budget is `2 * limit`). +//! +//! **Deadline.** The packs are built at once and waited for at most 12 +//! seconds from the start of the recall. A pack still building then is left +//! out (logged as a warning), and the evidence is rendered from the packs +//! that arrived, still most specific first. A host bounds a memory tool call +//! (OpenHuman: 15 seconds), and one scope's pack can take that long on its +//! own, so without the deadline the slowest scope decided whether any +//! evidence arrived. If no pack arrives in time the recall fails as +//! unavailable rather than claim memory holds nothing. A pack that fails +//! still fails the recall, as before. use std::collections::HashSet; -use futures::{StreamExt, TryStreamExt, stream}; +use futures::{StreamExt, stream}; use serde_json::Value; use tinymemory_api::{ Citation, ItemId, ItemKind, MetaFilter, RecallAnswer, RecallRequest, Role, ThreadWindow, @@ -41,7 +51,7 @@ use super::fetch::interleave; use super::items::{keeps, located_meta}; use super::recall::PACKS_AT_ONCE; use crate::cortex::envelope::{Envelope, decode_event}; -use crate::cortex::error::Result; +use crate::cortex::error::{Error, Result}; /// What one pack contributes to the evidence. #[derive(Debug, Default)] @@ -256,6 +266,43 @@ fn render(facts: &[String], beliefs: &[String], events: &[Envelope]) -> String { } impl CortexEngine { + /// The packs for `paths` that arrive within the evidence deadline (see + /// the module docs), in `paths` order. + async fn packs_by_deadline(&self, req: &RecallRequest, paths: &[String]) -> Result> { + let deadline = tokio::time::Instant::now() + self.log.timing.evidence; + let mut building = stream::iter(paths.iter().cloned().enumerate()) + .map(|(index, path)| async move { (index, self.pack(req, &path).await) }) + .buffer_unordered(PACKS_AT_ONCE); + let mut arrived = Vec::new(); + while let Ok(next) = tokio::time::timeout_at(deadline, building.next()).await { + let Some((index, pack)) = next else { break }; + arrived.push((index, pack?)); + } + let late: Vec<&str> = paths + .iter() + .enumerate() + .filter(|(index, _)| !arrived.iter().any(|(got, _)| got == index)) + .map(|(_, path)| path.as_str()) + .collect(); + if !paths.is_empty() && arrived.is_empty() { + return Err(Error::Unavailable(format!( + "no recall pack arrived within {:?}", + self.log.timing.evidence + ))); + } + if !late.is_empty() { + log::warn!( + "[cortex] evidence recall left out {} of {} scopes still building after {:?}: {}", + late.len(), + paths.len(), + self.log.timing.evidence, + late.join(", ") + ); + } + arrived.sort_by_key(|(index, _)| *index); + Ok(arrived.into_iter().map(|(_, pack)| pack).collect()) + } + /// See the module docs. `paths` are the kind scopes to read, most /// specific node first. pub(super) async fn recall_evidence( @@ -263,11 +310,7 @@ impl CortexEngine { req: &RecallRequest, paths: &[String], ) -> Result { - let packs: Vec = stream::iter(paths.to_vec()) - .map(|path| async move { self.pack(req, &path).await }) - .buffered(PACKS_AT_ONCE) - .try_collect() - .await?; + let packs = self.packs_by_deadline(req, paths).await?; let mut facts = Vec::new(); let mut beliefs = Vec::new(); let mut per_pack = Vec::new(); diff --git a/crates/tinymemory-integrations/src/cortex/engine/evidence_tests.rs b/crates/tinymemory-integrations/src/cortex/engine/evidence_tests.rs index 58163698..d015cacc 100644 --- a/crates/tinymemory-integrations/src/cortex/engine/evidence_tests.rs +++ b/crates/tinymemory-integrations/src/cortex/engine/evidence_tests.rs @@ -1,8 +1,9 @@ //! Tests for rendering recall packs as evidence. use super::*; +use crate::cortex::testing::{direct_double, direct_engine, sample_items}; use serde_json::json; -use tinymemory_api::{MemoryMeta, Reach, StoreItem, Turn, TurnRange}; +use tinymemory_api::{MemoryEngine, MemoryMeta, Reach, StoreItem, Turn, TurnRange}; fn turn(thread: &str, role: Role, text: &str) -> StoreItem { let mut meta = MemoryMeta { @@ -183,3 +184,44 @@ fn a_learning_made_in_an_excluded_thread_stays_in_the_evidence() { let texts: Vec<&str> = gathered.events.iter().map(|e| e.text.as_str()).collect(); assert_eq!(texts, vec!["the user prefers green tea"]); } + +/// A Direct engine holding [`sample_items`], with a short evidence +/// deadline, and recalls over `slow` scopes held past it. +async fn engine_with_slow(slow: &'static str) -> CortexEngine { + let (endpoint, state) = direct_double().await; + let mut engine = direct_engine(&endpoint); + engine.log.timing.evidence = std::time::Duration::from_millis(300); + for item in sample_items() { + engine.store(item).await.unwrap(); + } + *state.slow_recall.lock().unwrap() = Some((slow, 3_000)); + engine +} + +fn evidence_request() -> RecallRequest { + let mut req = RecallRequest::new("helix", 5); + req.mode = tinymemory_api::RecallMode::Evidence; + req +} + +#[tokio::test] +async fn a_scope_still_building_at_the_deadline_is_left_out() { + let engine = engine_with_slow("app:learnings").await; + let started = std::time::Instant::now(); + let got = engine.recall(evidence_request()).await.unwrap(); + assert!( + started.elapsed() < std::time::Duration::from_secs(2), + "the recall waited for the slow pack" + ); + assert!(got.answer.contains("## Conversation"), "{}", got.answer); + assert!(!got.citations.is_empty()); + assert!(!got.answer.contains("prefers helix"), "{}", got.answer); + assert!(got.citations.iter().all(|c| c.kind != ItemKind::Learning)); +} + +#[tokio::test] +async fn no_pack_by_the_deadline_is_unavailable_not_empty() { + let engine = engine_with_slow("app:").await; + let error = engine.recall(evidence_request()).await.unwrap_err(); + assert!(matches!(error, Error::Unavailable(_)), "{error:?}"); +} diff --git a/crates/tinymemory-integrations/src/cortex/log/mod.rs b/crates/tinymemory-integrations/src/cortex/log/mod.rs index e4011d20..78edee7e 100644 --- a/crates/tinymemory-integrations/src/cortex/log/mod.rs +++ b/crates/tinymemory-integrations/src/cortex/log/mod.rs @@ -81,6 +81,11 @@ pub(crate) struct Timing { /// How long a running Direct erasure is polled before it is reported /// as not done (the status requests included). pub(crate) erasure: Duration, + /// How long an evidence-mode recall waits for its packs. Packs still + /// building then are left out, so one slow scope does not cost the + /// others. Under the 15s a host typically allows one memory tool call + /// (OpenHuman's), so the evidence arrives inside it. + pub(crate) evidence: Duration, } /// Longest gap between hosted polls. A fixed 250ms poll would spend a fifth @@ -94,6 +99,7 @@ impl Default for Timing { settle: Duration::from_secs(10), poll: Duration::from_millis(250), erasure: Duration::from_secs(300), + evidence: Duration::from_secs(12), } } } diff --git a/crates/tinymemory-integrations/src/cortex/testing/mod.rs b/crates/tinymemory-integrations/src/cortex/testing/mod.rs index dd19fd61..5479ed30 100644 --- a/crates/tinymemory-integrations/src/cortex/testing/mod.rs +++ b/crates/tinymemory-integrations/src/cortex/testing/mod.rs @@ -161,6 +161,9 @@ pub(crate) struct Double { pub(crate) scoped_erase_answer: Mutex>, /// The status a settled erasure ends with; `completed` when unset. pub(crate) erasure_ends: Mutex>, + /// Recalls over a scope containing this text are held this long (ms) + /// before they answer, as a slow pack is. + pub(crate) slow_recall: Mutex>, } /// The shared handle the routes and tests hold. diff --git a/crates/tinymemory-integrations/src/cortex/testing/routes.rs b/crates/tinymemory-integrations/src/cortex/testing/routes.rs index a81294fa..b02831cf 100644 --- a/crates/tinymemory-integrations/src/cortex/testing/routes.rs +++ b/crates/tinymemory-integrations/src/cortex/testing/routes.rs @@ -335,6 +335,12 @@ async fn recall( if state.recall_down.load(Ordering::SeqCst) { return fail(&state, 500, "INTERNAL"); } + let slow = *state.slow_recall.lock().unwrap(); + if let Some((scope, ms)) = slow + && body["scope"].as_str().unwrap_or_default().contains(scope) + { + tokio::time::sleep(std::time::Duration::from_millis(ms)).await; + } let pack = state.log.lock().unwrap().recall(&body); ok(&state, 200, pack) } diff --git a/docs/architecture/cortex-flows.md b/docs/architecture/cortex-flows.md index 8c8aba96..0eabcc65 100644 --- a/docs/architecture/cortex-flows.md +++ b/docs/architecture/cortex-flows.md @@ -241,6 +241,12 @@ and **never asks the answer route**: no model runs after retrieval, and no answer is grounded on one pack while citing others. The engine renders the evidence itself (`engine/evidence.rs`): +0. Wait for the packs at most 12 seconds from the start of the recall + (`Timing::evidence`). A pack still building then is left out with a + warning, so one slow scope no longer costs the evidence of the others + inside a host's tool-call limit (OpenHuman's is 15 seconds); no pack in + time is `Unavailable`, never an empty answer. A pack that fails still + fails the recall. 1. Decode each pack's events, keep what the full `MetaFilter` admits, and drop the turns `exclude_thread` covers, noting their event ids. 2. When the filter narrows by reach and kinds alone, render the `facts` From 2db85f68f2d6f37c0e3ab968c6a08604d454f8a4 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Fri, 16 Oct 2026 02:55:27 +0530 Subject: [PATCH 2/2] fix(cortex): a recall that left a late pack out says so With the 12 s deadline, the conversation scope's pack could miss it while empty scopes (documents, learnings) answered, and the recall returned empty evidence: the host then told its agent nothing is stored in memory. The openhuman-benchmarks #38 retrieval test hit this on 2-4 of 199 questions. A recall that left a pack out and found nothing is now Unavailable; one that found something ends its evidence with a note that memory was only partly searched. Co-Authored-By: Claude Opus 5.5 --- .../src/cortex/engine/evidence.rs | 41 +++++++++++++++---- .../src/cortex/engine/evidence_tests.rs | 24 +++++++++++ docs/architecture/cortex-flows.md | 9 ++-- 3 files changed, 63 insertions(+), 11 deletions(-) diff --git a/crates/tinymemory-integrations/src/cortex/engine/evidence.rs b/crates/tinymemory-integrations/src/cortex/engine/evidence.rs index 2e7c288e..d19cc83f 100644 --- a/crates/tinymemory-integrations/src/cortex/engine/evidence.rs +++ b/crates/tinymemory-integrations/src/cortex/engine/evidence.rs @@ -34,9 +34,11 @@ //! that arrived, still most specific first. A host bounds a memory tool call //! (OpenHuman: 15 seconds), and one scope's pack can take that long on its //! own, so without the deadline the slowest scope decided whether any -//! evidence arrived. If no pack arrives in time the recall fails as -//! unavailable rather than claim memory holds nothing. A pack that fails -//! still fails the recall, as before. +//! evidence arrived. A recall that left a pack out says so: if the packs +//! that arrived hold nothing, it fails as unavailable (so does one where no +//! pack arrived) rather than claim memory holds nothing; otherwise the +//! evidence ends with a note that memory was only partly searched. A pack +//! that fails still fails the recall, as before. use std::collections::HashSet; @@ -267,8 +269,12 @@ fn render(facts: &[String], beliefs: &[String], events: &[Envelope]) -> String { impl CortexEngine { /// The packs for `paths` that arrive within the evidence deadline (see - /// the module docs), in `paths` order. - async fn packs_by_deadline(&self, req: &RecallRequest, paths: &[String]) -> Result> { + /// the module docs), in `paths` order, and how many were left out. + async fn packs_by_deadline( + &self, + req: &RecallRequest, + paths: &[String], + ) -> Result<(Vec, usize)> { let deadline = tokio::time::Instant::now() + self.log.timing.evidence; let mut building = stream::iter(paths.iter().cloned().enumerate()) .map(|(index, path)| async move { (index, self.pack(req, &path).await) }) @@ -300,7 +306,10 @@ impl CortexEngine { ); } arrived.sort_by_key(|(index, _)| *index); - Ok(arrived.into_iter().map(|(_, pack)| pack).collect()) + Ok(( + arrived.into_iter().map(|(_, pack)| pack).collect(), + late.len(), + )) } /// See the module docs. `paths` are the kind scopes to read, most @@ -310,7 +319,7 @@ impl CortexEngine { req: &RecallRequest, paths: &[String], ) -> Result { - let packs = self.packs_by_deadline(req, paths).await?; + let (packs, late) = self.packs_by_deadline(req, paths).await?; let mut facts = Vec::new(); let mut beliefs = Vec::new(); let mut per_pack = Vec::new(); @@ -321,7 +330,23 @@ impl CortexEngine { per_pack.push(gathered.events); } let events = interleave(per_pack); - let answer = render(&facts, &beliefs, &events); + let mut answer = render(&facts, &beliefs, &events); + if late > 0 { + if answer.is_empty() { + return Err(Error::Unavailable(format!( + "{late} of {} memory areas were still being searched after {:?} and the \ + rest hold nothing relevant; memory was not fully searched", + paths.len(), + self.log.timing.evidence + ))); + } + answer.push_str(&format!( + "\n\n(Memory was only partly searched: {late} of {} areas did not answer \ + within {:?}, so this evidence may be incomplete.)", + paths.len(), + self.log.timing.evidence + )); + } let citations = events .into_iter() .take(req.limit) diff --git a/crates/tinymemory-integrations/src/cortex/engine/evidence_tests.rs b/crates/tinymemory-integrations/src/cortex/engine/evidence_tests.rs index d015cacc..65aa90a7 100644 --- a/crates/tinymemory-integrations/src/cortex/engine/evidence_tests.rs +++ b/crates/tinymemory-integrations/src/cortex/engine/evidence_tests.rs @@ -216,9 +216,33 @@ async fn a_scope_still_building_at_the_deadline_is_left_out() { assert!(got.answer.contains("## Conversation"), "{}", got.answer); assert!(!got.citations.is_empty()); assert!(!got.answer.contains("prefers helix"), "{}", got.answer); + assert!( + got.answer.contains("only partly searched: 1 of"), + "{}", + got.answer + ); assert!(got.citations.iter().all(|c| c.kind != ItemKind::Learning)); } +#[tokio::test] +async fn a_late_scope_beside_empty_ones_is_unavailable_not_nothing_stored() { + let engine = engine_with_slow("app:conversations").await; + let mut req = evidence_request(); + req.question = "editor".into(); + let error = engine.recall(req).await.unwrap_err(); + assert!( + matches!(&error, Error::Unavailable(m) if m.contains("not fully searched")), + "{error:?}" + ); +} + +#[tokio::test] +async fn a_recall_with_every_pack_in_time_carries_no_note() { + let engine = engine_with_slow("app:no-such-scope").await; + let got = engine.recall(evidence_request()).await.unwrap(); + assert!(!got.answer.contains("partly searched"), "{}", got.answer); +} + #[tokio::test] async fn no_pack_by_the_deadline_is_unavailable_not_empty() { let engine = engine_with_slow("app:").await; diff --git a/docs/architecture/cortex-flows.md b/docs/architecture/cortex-flows.md index 0eabcc65..f7172efe 100644 --- a/docs/architecture/cortex-flows.md +++ b/docs/architecture/cortex-flows.md @@ -244,9 +244,12 @@ evidence itself (`engine/evidence.rs`): 0. Wait for the packs at most 12 seconds from the start of the recall (`Timing::evidence`). A pack still building then is left out with a warning, so one slow scope no longer costs the evidence of the others - inside a host's tool-call limit (OpenHuman's is 15 seconds); no pack in - time is `Unavailable`, never an empty answer. A pack that fails still - fails the recall. + inside a host's tool-call limit (OpenHuman's is 15 seconds). A recall + that left a pack out says so: if what arrived holds nothing it is + `Unavailable` (as is one where no pack arrived), never an empty answer + that reads as "nothing stored"; otherwise the evidence ends with a note + that memory was only partly searched. A pack that fails still fails the + recall. 1. Decode each pack's events, keep what the full `MetaFilter` admits, and drop the turns `exclude_thread` covers, noting their event ids. 2. When the filter narrows by reach and kinds alone, render the `facts`