Evidence recall: a 12 s deadline on the packs, and say when one was left out

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

crates/tinymemory-integrations/src/cortex/engine/evidence.rs

@@ -27,10 +27,22 @@
 //!
 //! 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. 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;
 
-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 +53,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,18 +268,58 @@ 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, and how many were left out.
+    async fn packs_by_deadline(
+        &self,
+        req: &RecallRequest,
+        paths: &[String],
+    ) -> Result<(Vec<Value>, 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) })
+            .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(),
+            late.len(),
+        ))
+    }
+
     /// See the module docs. `paths` are the kind scopes to read, most
     /// specific node first.
     pub(super) async fn recall_evidence(
         &self,
         req: &RecallRequest,
         paths: &[String],
     ) -> Result<RecallAnswer> {
-        let packs: Vec<Value> = stream::iter(paths.to_vec())
-            .map(|path| async move { self.pack(req, &path).await })
-            .buffered(PACKS_AT_ONCE)
-            .try_collect()
-            .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();
@@ -278,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)

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,68 @@ 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.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;
+    let error = engine.recall(evidence_request()).await.unwrap_err();
+    assert!(matches!(error, Error::Unavailable(_)), "{error:?}");
+}

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),
         }
     }
 }

crates/tinymemory-integrations/src/cortex/testing/mod.rs

@@ -161,6 +161,9 @@ pub(crate) struct Double {
     pub(crate) scoped_erase_answer: Mutex<Option<serde_json::Value>>,
     /// The status a settled erasure ends with; `completed` when unset.
     pub(crate) erasure_ends: Mutex<Option<&'static str>>,
+    /// 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<Option<(&'static str, u64)>>,
 }
 
 /// The shared handle the routes and tests hold.

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)
 }

docs/architecture/cortex-flows.md

@@ -241,6 +241,15 @@ 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). 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`