memoryd: recall asks Honcho after the local index, outside the state lock
Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
This commit is contained in:
@@ -24,6 +24,22 @@ 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 `recall` reads in Honcho: the sources, the owner's peer, and each source's limit
|
||||
/// (`SOURCE_WAIT` in `memoryd`).
|
||||
pub struct Reading {
|
||||
pub readers: Arc<Readers>,
|
||||
pub owner: String,
|
||||
pub wait: Duration,
|
||||
}
|
||||
|
||||
impl Reading {
|
||||
/// `recall`'s answer from the local index's items and Honcho's: `search`, then `answer`.
|
||||
pub fn recall(&self, local: Local, args: &RecallArgs, log: &Log) -> Recalled {
|
||||
let searched = search(&self.readers, &self.owner, args, self.wait, log);
|
||||
answer(local, &args.query, args.k, searched)
|
||||
}
|
||||
}
|
||||
|
||||
/// What the Honcho part found.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Searched {
|
||||
|
||||
+37
-20
@@ -19,6 +19,7 @@ use crate::files::Store;
|
||||
use crate::honcho::ledger::{self, LedgerError};
|
||||
use crate::honcho::mirror::Mirror;
|
||||
use crate::honcho::report;
|
||||
use crate::honcho::search;
|
||||
use crate::honcho::state;
|
||||
use crate::honcho::turns::TurnSync;
|
||||
use crate::honcho::worker::{self, Wake};
|
||||
@@ -47,6 +48,8 @@ pub struct Memoryd {
|
||||
wake: Arc<Wake>,
|
||||
/// With `[honcho]`: decides memory writes into the ledger, under the ledger's own lock.
|
||||
mirror: Option<Mirror>,
|
||||
/// With `[honcho]`: what `recall` reads beside the local index, outside `state`.
|
||||
reading: Option<search::Reading>,
|
||||
log: Log,
|
||||
}
|
||||
|
||||
@@ -96,9 +99,12 @@ enum Which {
|
||||
/// under `memory/` (and, the first time, write when sync began), open the index and catch it up,
|
||||
/// bind the three sockets, and start the Honcho worker. The error names what failed.
|
||||
pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
|
||||
let opened = match config.honcho_config()? {
|
||||
Some(honcho) => Some(worker::open(&config, &honcho, &log)?),
|
||||
None => None,
|
||||
let (opened, owner) = match config.honcho_config()? {
|
||||
Some(honcho) => (
|
||||
Some(worker::open(&config, &honcho, &log)?),
|
||||
honcho.owner.clone(),
|
||||
),
|
||||
None => (None, String::new()),
|
||||
};
|
||||
let store = Store::new(&config.memory_dir());
|
||||
store.init(&config.memory.soul_starter)?;
|
||||
@@ -119,14 +125,19 @@ pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
|
||||
(admin_listener, Which::Admin),
|
||||
];
|
||||
let wake = Arc::new(Wake::default());
|
||||
let (honcho, mirror) = match opened {
|
||||
let (honcho, mirror, reading) = match opened {
|
||||
Some(opened) => {
|
||||
let sync = Arc::clone(&opened.sync);
|
||||
let mirror = opened.mirror();
|
||||
let reading = search::Reading {
|
||||
readers: Arc::clone(&opened.readers),
|
||||
owner,
|
||||
wait: search::SOURCE_WAIT,
|
||||
};
|
||||
worker::spawn(opened, &config, Arc::clone(&wake), Arc::clone(&log))?;
|
||||
(Some(sync), Some(mirror))
|
||||
(Some(sync), Some(mirror), Some(reading))
|
||||
}
|
||||
None => (None, None),
|
||||
None => (None, None, None),
|
||||
};
|
||||
Ok(Running {
|
||||
memoryd: Arc::new(Memoryd {
|
||||
@@ -136,6 +147,7 @@ pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
|
||||
honcho,
|
||||
wake,
|
||||
mirror,
|
||||
reading,
|
||||
log,
|
||||
}),
|
||||
listeners,
|
||||
@@ -305,13 +317,13 @@ impl Memoryd {
|
||||
Message::Ok(Empty {})
|
||||
}
|
||||
|
||||
/// A decided call from `brokerd`: `recall` or `memory`, under the one lock. `Failed` carries the
|
||||
/// sentence for the model.
|
||||
/// A decided call from `brokerd`: `recall` or `memory`. `memory` runs under the one lock;
|
||||
/// `recall` takes it for the local index only (`recall`). `Failed` carries the sentence for
|
||||
/// the model.
|
||||
fn call(&self, call: MemoryCall) -> ToolResponse {
|
||||
let state = self.lock();
|
||||
let result = match call.request.tool.as_str() {
|
||||
bxcore::memory::RECALL => self.recall(&state, &call),
|
||||
bxcore::memory::MEMORY => self.memory(&state, &call),
|
||||
bxcore::memory::RECALL => self.recall(&call),
|
||||
bxcore::memory::MEMORY => self.memory(&self.lock(), &call),
|
||||
_ => {
|
||||
return ToolResponse::Failed {
|
||||
message: "memoryd runs only recall and memory".to_string(),
|
||||
@@ -329,19 +341,24 @@ impl Memoryd {
|
||||
}
|
||||
}
|
||||
|
||||
/// `recall`: search, labelled by what it found.
|
||||
fn recall(&self, state: &State, call: &MemoryCall) -> Result<Recalled, String> {
|
||||
/// `recall`: search, labelled by what it found. Without `[honcho]`, the local index under the
|
||||
/// lock, as before. With it, the local index under the lock, then, with the lock released,
|
||||
/// Honcho's sources (M5c spec, section 8): a source that hangs holds up no other request.
|
||||
fn recall(&self, call: &MemoryCall) -> Result<Recalled, String> {
|
||||
let args: RecallArgs = match serde_json::from_str(&call.request.arguments) {
|
||||
Ok(args) => args,
|
||||
Err(e) => return Err(format!("recall's arguments: {e}")),
|
||||
};
|
||||
recall::recall(
|
||||
&state.index,
|
||||
&self.store,
|
||||
&args,
|
||||
call.request.session.as_str(),
|
||||
Timestamp::now(),
|
||||
)
|
||||
let session = call.request.session.as_str();
|
||||
let Some(reading) = &self.reading else {
|
||||
let state = self.lock();
|
||||
return recall::recall(&state.index, &self.store, &args, session, Timestamp::now());
|
||||
};
|
||||
let local = {
|
||||
let state = self.lock();
|
||||
recall::local(&state.index, &self.store, &args, session, Timestamp::now())?
|
||||
};
|
||||
Ok(reading.recall(local, &args, &self.log))
|
||||
}
|
||||
|
||||
/// `memory`: refused, staged for review, or applied and journalled; then what it appended to
|
||||
|
||||
@@ -0,0 +1,241 @@
|
||||
//! `memoryd` answering `recall` with `[honcho]` (M5c spec, section 8), through its `brokerd`
|
||||
//! socket, against a fake Honcho: the local index's hits, then the read workspace's and the
|
||||
//! fleet's, the answer untrusted when a fleet hit is shown (T58); a hanging source holds up no
|
||||
//! memory call, because the state lock is released before Honcho is asked; without `[honcho]`,
|
||||
//! `recall` answers as before. Every wait has a deadline. Do not edit.
|
||||
|
||||
#[path = "support/honcho.rs"]
|
||||
mod honcho;
|
||||
#[path = "support/logs.rs"]
|
||||
mod logs;
|
||||
#[path = "support/tmp.rs"]
|
||||
mod tmp;
|
||||
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::Path;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use bxcore::{
|
||||
CallId, DataClass, Envelope, MemoryCall, Message, SessionId, ToolRequest, ToolResponse,
|
||||
read_frame, write_frame,
|
||||
};
|
||||
use honcho::{FakeHoncho, Mode};
|
||||
use logs::{user, write_log};
|
||||
use memoryd::config::Config;
|
||||
use memoryd::serve;
|
||||
use tmp::TempDir;
|
||||
|
||||
/// `[honcho]` for `url`, reading `boxmaker1`'s workspace, labelled trusted.
|
||||
fn section(url: &str) -> String {
|
||||
format!(
|
||||
"[honcho]\nurl = \"{url}\"\nagent = \"boxmaker0\"\nowner = \"kyle\"\ntrusted = \"boxmaker0-trusted\"\nuntrusted = \"boxmaker0-untrusted\"\nfleet = \"fleet\"\n[[honcho.read]]\nagent = \"boxmaker1\"\nworkspace = \"boxmaker1-trusted\"\ntrust = \"trusted\"\n"
|
||||
)
|
||||
}
|
||||
|
||||
struct Home {
|
||||
config: Config,
|
||||
/// Removed when the test ends; declared last, so it outlives what writes into it (T44).
|
||||
_dir: TempDir,
|
||||
}
|
||||
|
||||
impl Home {
|
||||
/// A home whose index holds one line the owner said, `tea time`; with `[honcho]` at `url`
|
||||
/// when there is one.
|
||||
fn new(tag: &str, url: Option<&str>) -> Home {
|
||||
let dir = TempDir::new(tag);
|
||||
let soul = dir.write("etc/soul.md", "You are Boxmaker.\n");
|
||||
write_log(dir.path(), "mm-a", &[user(1, "tea time")]);
|
||||
let text = format!(
|
||||
"[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n{}",
|
||||
dir.path().display(),
|
||||
soul.display(),
|
||||
url.map(section).unwrap_or_default()
|
||||
);
|
||||
Home {
|
||||
config: Config::parse(&text).unwrap(),
|
||||
_dir: dir,
|
||||
}
|
||||
}
|
||||
|
||||
/// `serve::start`, run on a thread; its log lines are collected.
|
||||
fn serve(&self) -> Arc<Mutex<Vec<String>>> {
|
||||
let lines = Arc::new(Mutex::new(Vec::new()));
|
||||
let sink = Arc::clone(&lines);
|
||||
let log: serve::Log = Arc::new(move |l: &str| sink.lock().unwrap().push(l.to_string()));
|
||||
let running = serve::start(self.config.clone(), log).unwrap();
|
||||
std::thread::spawn(move || running.run());
|
||||
lines
|
||||
}
|
||||
}
|
||||
|
||||
/// One call on `brokerd`'s socket and its answer, within 20 s; how long it took.
|
||||
fn call(socket: &Path, tool: &str, arguments: &str) -> (ToolResponse, Duration) {
|
||||
let started = Instant::now();
|
||||
let mut stream = UnixStream::connect(socket).unwrap();
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(20)))
|
||||
.unwrap();
|
||||
let env = Envelope {
|
||||
v: 1,
|
||||
id: 3,
|
||||
r#final: true,
|
||||
msg: Message::MemoryCall(MemoryCall {
|
||||
request: ToolRequest {
|
||||
session: SessionId::new("mm-now").unwrap(),
|
||||
call: CallId(3),
|
||||
tool: tool.to_string(),
|
||||
arguments: arguments.to_string(),
|
||||
},
|
||||
taint: DataClass::Private,
|
||||
untrusted: false,
|
||||
}),
|
||||
};
|
||||
write_frame(&mut stream, &env).unwrap();
|
||||
match read_frame(&mut stream).unwrap().msg {
|
||||
Message::ToolResponse(response) => (response, started.elapsed()),
|
||||
other => panic!("{other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn result(response: ToolResponse) -> (String, DataClass, bool) {
|
||||
match response {
|
||||
ToolResponse::Result {
|
||||
content,
|
||||
class,
|
||||
untrusted,
|
||||
..
|
||||
} => (content, class, untrusted),
|
||||
other => panic!("{other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// The paths of the fake's search and conclusion-query requests.
|
||||
fn reads(fake: &FakeHoncho) -> Vec<String> {
|
||||
fake.requests()
|
||||
.into_iter()
|
||||
.map(|r| r.path)
|
||||
.filter(|p| p.ends_with("/search") || p.ends_with("/conclusions/query"))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// T58, end to end: a fleet hit shown makes `memoryd`'s answer untrusted, as `brokerd` will label
|
||||
/// the session (task 02); hits only from a workspace labelled trusted do not.
|
||||
#[test]
|
||||
fn a_fleet_hit_makes_the_answer_untrusted() {
|
||||
let fake = FakeHoncho::start();
|
||||
let home = Home::new("recall-serve-fleet", Some(&fake.url));
|
||||
let _lines = home.serve();
|
||||
let socket = home.config.broker_socket();
|
||||
let (content, class, untrusted) =
|
||||
result(call(&socket, "recall", r#"{"query":"tea","k":10}"#).0);
|
||||
assert!(
|
||||
content.starts_with("[1] memory · session mm-a, 2026-09-20 14:01 UTC, the owner said · private\n tea time\n[2] memory · boxmaker1: kyle said, "),
|
||||
"{content}"
|
||||
);
|
||||
assert!(
|
||||
content.contains("\n[3] memory-tainted · fleet: kyle said, "),
|
||||
"{content}"
|
||||
);
|
||||
// The conclusions about the owner's peer, `[honcho] owner`, are asked for and shown.
|
||||
assert!(
|
||||
content.contains("memory · boxmaker1: a conclusion about kyle, "),
|
||||
"{content}"
|
||||
);
|
||||
assert_eq!((class, untrusted), (DataClass::Private, true));
|
||||
assert!(
|
||||
!reads(&fake).iter().any(|p| p.contains("boxmaker0")),
|
||||
"{:?}",
|
||||
reads(&fake)
|
||||
);
|
||||
fake.set_workspace("fleet", Mode::Missing);
|
||||
let (content, _, untrusted) = result(call(&socket, "recall", r#"{"query":"tea","k":10}"#).0);
|
||||
assert!(!content.contains("fleet"), "{content}");
|
||||
assert!(!untrusted, "{content}");
|
||||
}
|
||||
|
||||
/// `recall` releases the state lock before it asks Honcho: while a source hangs, a memory call is
|
||||
/// answered at once, and the `recall` answers after the source's limit with its `unavailable`
|
||||
/// line.
|
||||
#[test]
|
||||
fn a_hanging_source_holds_up_no_memory_call() {
|
||||
let fake = FakeHoncho::start();
|
||||
fake.set_workspace("fleet", Mode::Hang);
|
||||
let home = Home::new("recall-serve-hang", Some(&fake.url));
|
||||
let lines = home.serve();
|
||||
let socket = home.config.broker_socket();
|
||||
let for_recall = socket.clone();
|
||||
let recall = std::thread::spawn(move || call(&for_recall, "recall", r#"{"query":"tea"}"#));
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
while !reads(&fake)
|
||||
.iter()
|
||||
.any(|p| p == "/v3/workspaces/fleet/search")
|
||||
{
|
||||
assert!(Instant::now() < deadline, "the fleet was never searched");
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
let (answer, took) = call(
|
||||
&socket,
|
||||
"memory",
|
||||
r#"{"action":"note","text":"tea is kept in the blue tin"}"#,
|
||||
);
|
||||
assert!(matches!(answer, ToolResponse::Result { .. }), "{answer:?}");
|
||||
assert!(
|
||||
took < Duration::from_secs(2),
|
||||
"the memory call waited {took:?}"
|
||||
);
|
||||
assert!(!recall.is_finished(), "the recall is still waiting");
|
||||
let (answer, took) = recall.join().unwrap();
|
||||
let (content, _, untrusted) = result(answer);
|
||||
assert!(content.ends_with("\nfleet: unavailable"), "{content}");
|
||||
assert!(took < Duration::from_secs(15), "{took:?}");
|
||||
assert!(!untrusted, "no fleet hit was shown: {content}");
|
||||
assert!(
|
||||
lines
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.any(|l| l.starts_with("memoryd: recall: fleet unavailable: ")),
|
||||
"{:?}",
|
||||
lines.lock().unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
/// Nothing found anywhere: the sentence local `recall` gives.
|
||||
#[test]
|
||||
fn nothing_found_anywhere() {
|
||||
let fake = FakeHoncho::start();
|
||||
let home = Home::new("recall-serve-nothing", Some(&fake.url));
|
||||
let _lines = home.serve();
|
||||
let (content, class, untrusted) = result(
|
||||
call(
|
||||
&home.config.broker_socket(),
|
||||
"recall",
|
||||
r#"{"query":"zebra"}"#,
|
||||
)
|
||||
.0,
|
||||
);
|
||||
assert_eq!(content, "(no matches for \"zebra\")");
|
||||
assert_eq!((class, untrusted), (DataClass::Private, false));
|
||||
}
|
||||
|
||||
/// Without `[honcho]`, `recall` is the local index's, and nothing else is asked.
|
||||
#[test]
|
||||
fn without_honcho_recall_answers_as_before() {
|
||||
let home = Home::new("recall-serve-off", None);
|
||||
let lines = home.serve();
|
||||
let socket = home.config.broker_socket();
|
||||
let (content, class, untrusted) = result(call(&socket, "recall", r#"{"query":"tea"}"#).0);
|
||||
assert_eq!(
|
||||
content,
|
||||
"[1] memory · session mm-a, 2026-09-20 14:01 UTC, the owner said · private\n tea time"
|
||||
);
|
||||
assert_eq!((class, untrusted), (DataClass::Private, false));
|
||||
let (content, _, _) = result(call(&socket, "recall", r#"{"query":"zebra"}"#).0);
|
||||
assert_eq!(content, "(no matches for \"zebra\")");
|
||||
assert!(
|
||||
lines.lock().unwrap().is_empty(),
|
||||
"{:?}",
|
||||
lines.lock().unwrap()
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user