M5c-2 plan: task 08, recall's Honcho part; the wiring split out as task 09

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 19:30:58 -07:00
parent 109aacbd1b
commit d41c41eb4a
9 changed files with 1130 additions and 11 deletions
@@ -0,0 +1,165 @@
+//! `recall`'s Honcho part (M5c spec, section 8): every source `Readers::recall_sources` names,
+//! searched at once, each on its own thread with its own limit, then its answer made from the
+//! local index's items followed by Honcho's. A query that holds a secret asks nothing. Runs
+//! outside `memoryd`'s state lock; a source still running at its limit is left to finish, and
+//! what it finds is dropped.
+
+use std::sync::{Arc, mpsc};
+use std::time::{Duration, Instant};
+
+use bxcore::RecallArgs;
+use roncho::{Conclusion, ConclusionQuery, MessageSearch};
+use serde_json::{Map, json};
+
+use crate::honcho::client::{HonchoError, classify};
+use crate::honcho::decide::has_secret;
+use crate::honcho::read::{Reader, Readers, Source};
+use crate::honcho::recalled::{Item, interleave, items, label, render, unavailable};
+use crate::recall::{Local, Recalled};
+use crate::serve::Log;
+
+/// How long each source has, counted from when the Honcho part starts (M5c spec, section 8).
+pub const SOURCE_WAIT: Duration = Duration::from_secs(5);
+
+/// The line in place of Honcho's items when the query holds a known-prefix secret.
+pub const SECRET: &str = "honcho: not asked; the query looks like it holds a secret";
+
+/// What the Honcho part found.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Searched {
+    /// The query holds a secret: no source was asked.
+    Secret,
+    /// Each source's kept items, in `recall_sources` order (none for a source that did not
+    /// answer), and the names of those that did not, in the same order.
+    Asked {
+        found: Vec<Vec<Item>>,
+        unavailable: Vec<String>,
+    },
+}
+
+/// The conclusions about `peer` that `peer` drew, as `memoryd` writes them: the observer and the
+/// observed are the same peer (M5c spec, section 8).
+fn conclusions(
+    reader: &Reader,
+    peer: &str,
+    query: &str,
+    k: u32,
+) -> Result<Vec<Conclusion>, HonchoError> {
+    let mut filters = Map::new();
+    filters.insert("observer_id".to_string(), json!(peer));
+    filters.insert("observed_id".to_string(), json!(peer));
+    let asked = ConclusionQuery::new(query).top_k(k).filters(filters);
+    reader
+        .search()
+        .query_conclusions(&asked)
+        .map_err(|e| classify(&e))
+}
+
+/// One source, its calls one after another: the workspace's messages, then for another agent's
+/// workspace the conclusions about that agent's peer and about the owner's. Its kept items, or
+/// the first call's failure.
+pub fn source(reader: &Reader, owner: &str, query: &str, k: u32) -> Result<Vec<Item>, HonchoError> {
+    // 1. messages: `reader.search().search_workspace(&MessageSearch { limit: Some(k),
+    //    ..MessageSearch::new(query) })`, its error mapped with `classify(&e)`, then `?`.
+    // 2. For `Source::Agent(peer)`: about_agent = `conclusions(reader, peer, query, k)?`, then
+    //    about_owner = `conclusions(reader, owner, query, k)?`, in that order. Any other source:
+    //    both `Vec::new()`, and no conclusion is asked for.
+    // 3. `Ok(items(&reader.label, query, &messages, &about_agent, &about_owner))`.
+    todo!()
+}
+
+/// Ask every source of `readers.recall_sources()` at once for `args`, each on a thread of its
+/// own, and wait for each at most `wait` from now. A source that failed, or that has not answered
+/// by then, is unavailable; each is logged by its name and the error's kind and status, never the
+/// query. `Secret`, with nothing sent, when the query holds a known-prefix secret.
+pub fn search(
+    readers: &Arc<Readers>,
+    owner: &str,
+    args: &RecallArgs,
+    wait: Duration,
+    log: &Log,
+) -> Searched {
+    if has_secret(&args.query) {
+        return Searched::Secret;
+    }
+    let deadline = Instant::now() + wait;
+    let k = u32::try_from(args.k).unwrap_or(u32::MAX);
+    let names: Vec<String> = readers
+        .recall_sources()
+        .iter()
+        .map(|r| r.label.name.clone())
+        .collect();
+    let (tx, rx) = mpsc::channel::<(usize, Result<Vec<Item>, HonchoError>)>();
+    let mut answers: Vec<Option<Result<Vec<Item>, HonchoError>>> = vec![None; names.len()];
+    for (i, slot) in answers.iter_mut().enumerate() {
+        let (readers, sender) = (Arc::clone(readers), tx.clone());
+        let (owner, query) = (owner.to_string(), args.query.clone());
+        let spawned = std::thread::Builder::new()
+            .name("honcho-recall".to_string())
+            .spawn(move || {
+                if let Some(reader) = readers.recall_sources().get(i) {
+                    let _ = sender.send((i, source(reader, &owner, &query, k)));
+                }
+            });
+        if spawned.is_err() {
+            *slot = Some(Err(HonchoError::Unreachable {
+                kind: "thread",
+                status: None,
+            }));
+        }
+    }
+    drop(tx);
+    while answers.iter().any(Option::is_none) {
+        let left = deadline.saturating_duration_since(Instant::now());
+        match rx.recv_timeout(left) {
+            Ok((i, answer)) => {
+                if let Some(slot) = answers.get_mut(i) {
+                    *slot = Some(answer);
+                }
+            }
+            Err(_) => break,
+        }
+    }
+    let mut found = Vec::with_capacity(names.len());
+    let mut missing = Vec::new();
+    for (name, answer) in names.into_iter().zip(answers) {
+        match answer {
+            Some(Ok(kept)) => found.push(kept),
+            Some(Err(e)) => {
+                (log)(&format!("memoryd: recall: {name} unavailable: {e}"));
+                found.push(Vec::new());
+                missing.push(name);
+            }
+            None => {
+                (log)(&format!(
+                    "memoryd: recall: {name} unavailable: no answer in {} ms",
+                    wait.as_millis()
+                ));
+                found.push(Vec::new());
+                missing.push(name);
+            }
+        }
+    }
+    Searched::Asked {
+        found,
+        unavailable: missing,
+    }
+}
+
+/// `recall`'s answer from the local index's items and what Honcho found: the local items, then
+/// Honcho's up to `k` in all, numbered after them; `(no matches for "…")` when there is neither;
+/// then the `unavailable` lines, or the secret line. The label is the stricter of the two parts'.
+pub fn answer(local: Local, query: &str, k: u64, searched: Searched) -> Recalled {
+    // 1. k as a usize (`usize::try_from(k).unwrap_or(usize::MAX)`).
+    // 2. (shown, tail): `Secret` -> (no items, `SECRET.to_string()`); `Asked { found,
+    //    unavailable: missing }` (bound as `missing`: the field has the function's name) ->
+    //    (`interleave(local.items.len(), k, found)`, `unavailable(&names)` where names is
+    //    `missing` as a `Vec<&str>`).
+    // 3. (class, untrusted) = `label(&shown)`; `first` = local.items.len() + 1, taken before the
+    //    items move. parts: `local.items`; then, if `shown` is not empty, `render(first,
+    //    &shown)`; then, if parts is still empty, format!("(no matches for \"{query}\")"); then
+    //    `tail` if it is not empty.
+    // 4. Recalled { text: parts joined by "\n", class: `local.class.max(class)`, untrusted:
+    //    `local.untrusted || untrusted` }.
+    todo!()
+}