diff --git a/crates/memoryd/src/honcho/mirror.rs b/crates/memoryd/src/honcho/mirror.rs index fdc2335..9fbf4b9 100644 --- a/crates/memoryd/src/honcho/mirror.rs +++ b/crates/memoryd/src/honcho/mirror.rs @@ -225,9 +225,16 @@ impl Mirror { /// Decide the journalled edit `id` (the latest when `None`), called under the index lock /// right after it was journalled, so the ledger has the journal's order; the ledger's lock is /// taken inside the index lock, and before an undo is decided against it. Its parts are - /// `Edit { id, part }` from 0, numbering what is decided. An edit already decided, or asking nothing, adds no line. A - /// failure is logged; the edit itself stands. - pub fn journalled(&self, store: &Store, id: Option<&str>, now: Timestamp, log: &dyn Fn(&str)) { + /// `Edit { id, part }` from 0, numbering what is decided. An edit already decided, or asking + /// nothing, adds no line. A failure is logged; the edit itself stands. Whether any line was + /// added, so that the worker is woken only when there is something new. + pub fn journalled( + &self, + store: &Store, + id: Option<&str>, + now: Timestamp, + log: &dyn Fn(&str), + ) -> bool { let records = match journal::records(store) { Ok(records) => records, Err(e) => { @@ -235,7 +242,7 @@ impl Mirror { "memoryd: {e}; the edit is not mirrored in Honcho; {}", ledger::RUNBOOK )); - return; + return false; } }; let found = match id { @@ -243,7 +250,7 @@ impl Mirror { None => records.last(), }; let Some(record) = found else { - return; + return false; }; let session = session_of(record, &records); let mut book = self.ledger.lock().unwrap_or_else(|p| p.into_inner()); @@ -251,9 +258,10 @@ impl Mirror { if record.undoes.is_some() { actions.retain(|a| brought_back(&book, a)); } + let mut added = false; for (part, action) in actions.into_iter().enumerate() { let Ok(part) = u32::try_from(part) else { - return; + return added; }; let item = Item::Edit { id: record.id.clone(), @@ -268,8 +276,10 @@ impl Mirror { action, }) { log(&format!("memoryd: {e}; {}", ledger::RUNBOOK)); - return; + return added; } + added = true; } + added } } diff --git a/crates/memoryd/src/serve.rs b/crates/memoryd/src/serve.rs index 2d4e48e..7d5d691 100644 --- a/crates/memoryd/src/serve.rs +++ b/crates/memoryd/src/serve.rs @@ -228,9 +228,12 @@ impl Memoryd { /// With `[honcho]`: decide journalled edit `id` (the latest when `None`) into the ledger. Called /// under the index lock, so edits are decided in the journal's order (M5c spec, section 4). + /// What it decided is sent at once: the worker is woken. fn mirror(&self, id: Option<&str>) { - if let Some(mirror) = &self.mirror { - mirror.journalled(&self.store, id, Timestamp::now(), &*self.log); + if let Some(mirror) = &self.mirror + && mirror.journalled(&self.store, id, Timestamp::now(), &*self.log) + { + self.wake.notify(); } } diff --git a/crates/memoryd/tests/honcho_wake_serve.rs b/crates/memoryd/tests/honcho_wake_serve.rs new file mode 100644 index 0000000..15b801b --- /dev/null +++ b/crates/memoryd/tests/honcho_wake_serve.rs @@ -0,0 +1,131 @@ +//! A memory write wakes the Honcho worker (M5c spec, section 4: a write reaches Honcho "at once"; +//! review of M5c-1, finding 3): with the worker asleep for its 30 s idle wait, the memory tool's +//! conclusion and the owner's undo's deletion both reach the fake Honcho within 5 s. + +#[path = "support/honcho.rs"] +mod honcho; +#[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, MemoryAdmin, MemoryCall, MemoryCommand, Message, SessionId, + ToolRequest, read_frame, write_frame, +}; +use honcho::FakeHoncho; +use memoryd::config::Config; +use memoryd::honcho::worker::Tuning; +use memoryd::serve; +use tmp::TempDir; + +fn ask(socket: &Path, msg: Message) -> Message { + let mut stream = UnixStream::connect(socket).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let env = Envelope { + v: 1, + id: 1, + r#final: true, + msg, + }; + write_frame(&mut stream, &env).unwrap(); + read_frame(&mut stream).unwrap().msg +} + +/// How many requests of `method` to a path ending `suffix` the fake has had. +fn count(fake: &FakeHoncho, method: &str, suffix: &str) -> usize { + fake.requests() + .iter() + .filter(|r| { + r.method == method && r.path.contains("/conclusions") && r.path.ends_with(suffix) + }) + .count() +} + +/// Waits up to 5 s for `done`; true if it came. +fn within_5s(done: impl Fn() -> bool) -> bool { + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + if done() { + return true; + } + std::thread::sleep(Duration::from_millis(10)); + } + done() +} + +#[test] +fn a_memory_write_and_an_undo_wake_the_worker() { + // The worker `serve` starts waits this long when idle: far past every deadline here. + assert!(Tuning::DEFAULT.idle >= Duration::from_secs(30)); + let fake = FakeHoncho::start(); + let dir = TempDir::new("wake-serve"); + let soul = dir.write("etc/soul.md", "You are Boxmaker.\n"); + let text = format!( + "[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n[honcho]\nurl = \"{}\"\nagent = \"boxmaker0\"\nowner = \"kyle\"\ntrusted = \"boxmaker0-trusted\"\nuntrusted = \"boxmaker0-untrusted\"\nfleet = \"fleet\"\n", + dir.path().display(), + soul.display(), + fake.url + ); + let config = Config::parse(&text).unwrap(); + 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(config.clone(), log).unwrap(); + std::thread::spawn(move || running.run()); + // Let the worker find nothing to send and fall asleep. + std::thread::sleep(Duration::from_millis(200)); + + let call = Message::MemoryCall(MemoryCall { + request: ToolRequest { + session: SessionId::new("mm-a").unwrap(), + call: CallId(1), + tool: "memory".to_string(), + arguments: r#"{"action":"add","file":"user","text":"Prefers UTC."}"#.to_string(), + }, + taint: DataClass::Private, + untrusted: false, + }); + assert!(matches!( + ask(&config.broker_socket(), call), + Message::ToolResponse(_) + )); + assert!( + within_5s(|| count(&fake, "POST", "/conclusions") == 1), + "the conclusion waited: {:?} {:?}", + fake.requests(), + lines.lock().unwrap() + ); + std::thread::sleep(Duration::from_millis(200)); + + let log = ask( + &config.admin_socket(), + Message::MemoryAdmin(MemoryAdmin { + command: MemoryCommand::Log, + arg: String::new(), + }), + ); + let Message::MemoryAdminReply(reply) = log else { + panic!("{log:?}") + }; + let id = reply.text.split(" ").next().unwrap().to_string(); + let undone = ask( + &config.admin_socket(), + Message::MemoryAdmin(MemoryAdmin { + command: MemoryCommand::Undo, + arg: id, + }), + ); + assert!(matches!(undone, Message::MemoryAdminReply(_)), "{undone:?}"); + assert!( + within_5s(|| count(&fake, "DELETE", "") == 1), + "the deletion waited: {:?} {:?}", + fake.requests(), + lines.lock().unwrap() + ); +}