M5c-1 plan: task 13, the Honcho worker sends memory writes

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-28 09:27:51 -07:00
co-authored by Claude Opus 5.5
parent d3d37c955a
commit 687d07a9a7
10 changed files with 1938 additions and 14 deletions
@@ -0,0 +1,78 @@
//! Memory writes sent to Honcho (M5c spec, sections 4 and 5), for the worker: a conclusion made
//! in the Boxmaker session's memory session, `<id>_memory`; a conclusion deleted by the id its
//! making was given; an untrusted note added to the fleet workspace as a message from the agent.
//! The text is neutralised here, as it leaves.
use bxcore::SessionId;
use bxcore::sanitize::neutralise;
use roncho::{ConclusionBatchCreate, ConclusionCreate, MessageCreate};
use serde_json::{Map, json};
use crate::honcho::client::{Clients, HonchoError, classify};
use crate::honcho::ledger::Workspace;
use crate::honcho::naming::HonchoSession;
use crate::honcho::worker::parts;
/// What names a Boxmaker session's memory session: `_` is in no Boxmaker id, so the name is no
/// other session's.
pub const MEMORY_SUFFIX: &str = "_memory";
/// The memory session of `session`: `<id>_memory`.
pub fn memory_session(session: &SessionId) -> String {
// `format!("{}{MEMORY_SUFFIX}", session.as_str())`.
todo!()
}
/// Make conclusion `text` about `peer` (its observer and the observed) in `workspace`, stamped
/// with `session`'s memory session, which is made sure of first. Its one id.
pub fn conclude(
clients: &mut Clients,
workspace: Workspace,
peer: &str,
session: &SessionId,
text: &str,
) -> Result<Vec<String>, HonchoError> {
// `let name = memory_session(session);`, then
// `clients.ensure_memory_session(workspace, &name)?;`. The conclusion:
// `ConclusionCreate::new(neutralise(text), peer, peer).with_session_id(name)` (observer and
// observed both `peer`), in `ConclusionBatchCreate::new(vec![conclusion])`. Then `match
// clients.client(workspace).create_conclusions(&batch)`: `Ok(made)` ->
// `Ok(made.into_iter().map(|c| c.id).collect())`; `Err(e)` -> `Err(classify(&e))`.
todo!()
}
/// Delete conclusion `id` in `workspace`; with none to delete, nothing is asked. The ids of what
/// was deleted.
pub fn unconclude(
clients: &Clients,
workspace: Workspace,
id: Option<&str>,
) -> Result<Vec<String>, HonchoError> {
// `match id`: `Some(id)` -> `clients.delete_conclusion(workspace, id)?;` then
// `Ok(vec![id.to_string()])`; `None` -> `Ok(Vec::new())`: nothing is asked.
todo!()
}
/// An untrusted note's messages for the fleet workspace: from the agent's peer, neutralised, in
/// parts of at most `MAX_PART` characters, each with the note's metadata.
pub fn fleet_batch(text: &str, agent: &str, session: &SessionId, edit: &str) -> Vec<MessageCreate> {
// A `Map` `metadata` with `"agent"` -> `json!(agent)`, `"boxmaker_session"` ->
// `json!(session.as_str())`, `"edit"` -> `json!(edit)` (in this order). Then
// `parts(&neutralise(text)).into_iter().map(|part| MessageCreate::new(part, agent)
// .with_metadata(metadata.clone())).collect()`.
todo!()
}
/// Add `messages` to `session`'s Honcho session in the fleet workspace, which holds only the
/// agent's peer, made sure of first. The ids of the messages.
pub fn fleet(
clients: &mut Clients,
session: &SessionId,
messages: &[MessageCreate],
) -> Result<Vec<String>, HonchoError> {
// `let honcho = HonchoSession { name: session.as_str().to_string(), speakers: Vec::new() };`
// then `clients.ensure_session(Workspace::Fleet, &honcho)?;`. Then `match
// clients.client(Workspace::Fleet).add_messages(&honcho.name, messages)`: `Ok(made)` ->
// `Ok(made.into_iter().map(|m| m.id).collect())`; `Err(e)` -> `Err(classify(&e))`.
todo!()
}
@@ -0,0 +1,85 @@
//! `memoryd`'s Honcho side at start (M5c spec, sections 5 and 10): the ledger read through, the
//! sessions that wait again, the clients with their keys, and the worker's thread.
use std::sync::{Arc, Mutex};
use bxcore::SessionId;
use crate::config::Config;
use crate::honcho::client::{Clients, Timeouts};
use crate::honcho::config::HonchoConfig;
use crate::honcho::ledger::Ledger;
use crate::honcho::mirror::Mirror;
use crate::honcho::state;
use crate::honcho::turns::{LEDGER, TurnSync};
use crate::honcho::worker::{AskBox, Tuning, Wake, Worker};
use crate::serve::{Log, StartError};
/// With `[honcho]`, at start: the ledger read through, the sessions that wait again, and the
/// clients with their keys read. Nothing is bound or started yet.
pub struct Opened {
pub sync: Arc<Mutex<TurnSync>>,
clients: Clients,
agent: String,
owner: String,
}
impl Opened {
/// The memory writes' side of the ledger (task 12).
pub fn mirror(&self) -> Mirror {
let ledger = self
.sync
.lock()
.unwrap_or_else(|p| p.into_inner())
.shared_ledger();
Mirror::new(ledger, &self.agent, &self.owner)
}
}
/// Open the ledger (a cut-off last line is logged), resume the waiting sessions, and build the
/// clients (their key files' warnings are logged). A key that cannot be read is `Secret`.
pub fn open(config: &Config, honcho: &HonchoConfig, log: &Log) -> Result<Opened, StartError> {
let path = config.memory_dir().join(LEDGER);
let ledger = Ledger::open(&path).map_err(StartError::Ledger)?;
if ledger.repaired() {
(log)(&format!(
"memoryd: {}: a last line cut off by a crash was dropped",
path.display()
));
}
let mut sync = TurnSync::new(ledger, &honcho.agent, &config.sessions_dir());
sync.resume();
let env = |name: &str| std::env::var_os(name);
let (clients, warnings) =
Clients::new(honcho, &env, Timeouts::WRITES).map_err(StartError::Secret)?;
for warning in warnings {
(log)(&warning);
}
Ok(Opened {
sync: Arc::new(Mutex::new(sync)),
clients,
agent: honcho.agent.clone(),
owner: honcho.owner.clone(),
})
}
/// Start the worker's thread, asking `brokerd` on `config`'s state socket.
pub fn spawn(opened: Opened, config: &Config, wake: Arc<Wake>, log: Log) -> Result<(), String> {
let socket = config.state_socket();
let ask: AskBox =
Box::new(move |session: &SessionId| state::ask(&socket, session, state::WAIT));
let worker = Worker::new(
opened.sync,
opened.clients,
wake,
ask,
log,
Tuning::DEFAULT,
&opened.agent,
);
std::thread::Builder::new()
.name("honcho".to_string())
.spawn(move || worker.run())
.map(|_| ())
.map_err(|e| format!("cannot start the Honcho worker: {e}"))
}
@@ -0,0 +1,411 @@
//! The worker sending memory writes (M5c spec, sections 4 and 5), stepped by hand against the fake
//! Honcho: a conclusion made in the session's `<id>_memory` session with the agent's and the
//! owner's peers; one deleted by the id its making got, or nothing to delete; a 404 on delete as
//! done; an untrusted note as a fleet message from the agent; the text neutralised; one outbox
//! across turns and edits; errors as for turns; a restart sending nothing again. Do not edit.
#[path = "support/honcho.rs"]
mod honcho;
#[path = "support/logs.rs"]
mod logs;
#[path = "support/tmp.rs"]
mod tmp;
#[path = "support/turns.rs"]
mod turns;
#[path = "support/worker.rs"]
mod worker;
use bxcore::sanitize::TAGS;
use bxcore::{SessionId, Timestamp};
use honcho::{FakeHoncho, Mode, Request};
use logs::assistant;
use memoryd::honcho::edits::{MEMORY_SUFFIX, memory_session};
use memoryd::honcho::ledger::{Action, Item, Record, Workspace};
use memoryd::honcho::worker::{MAX_PART, Step};
use serde_json::{Value, json};
use turns::spoken;
use worker::{FAST, Home};
/// The id fixture 11 gives the first conclusion it makes.
const CONCLUSION: &str = "YUBMqIqHhuJprV4o-inrg";
fn sid(s: &str) -> SessionId {
SessionId::new(s).unwrap()
}
fn edit(id: &str, part: u32) -> Item {
Item::Edit {
id: id.to_string(),
part,
}
}
fn conclude(peer: &str, text: &str) -> Action {
Action::Conclude {
workspace: Workspace::Trusted,
peer: peer.to_string(),
session: sid("mm-a"),
text: text.to_string(),
}
}
fn unconclude(text: &str) -> Action {
Action::Unconclude {
workspace: Workspace::Trusted,
text: text.to_string(),
}
}
fn fleet(text: &str) -> Action {
Action::Fleet {
session: sid("mm-c"),
text: text.to_string(),
}
}
/// Decide `action` for `item`, as task 12 does when an edit is journalled.
fn decide(home: &Home, item: Item, action: Action) {
let sync = home.sync.lock().unwrap();
sync.ledger()
.append(Record::Decided {
time: Timestamp::parse("2026-09-28T10:00:00.000Z").unwrap(),
item,
action,
})
.unwrap();
}
fn sent_ids(home: &Home, item: &Item) -> Option<Vec<String>> {
home.records().into_iter().find_map(|r| match r {
Record::Sent { item: i, ids, .. } if i == *item => Some(ids),
_ => None,
})
}
fn posts(fake: &FakeHoncho, suffix: &str) -> Vec<Request> {
fake.requests()
.into_iter()
.filter(|r| r.path.ends_with(suffix))
.collect()
}
#[test]
fn the_memory_session_is_named_for_the_boxmaker_session() {
assert_eq!(MEMORY_SUFFIX, "_memory");
assert_eq!(memory_session(&sid("mm-a")), "mm-a_memory");
}
#[test]
fn a_conclusion_goes_into_the_sessions_memory_session() {
let fake = FakeHoncho::start();
let home = Home::new("edits-conclude");
decide(&home, edit("e1", 0), conclude("kyle", "Prefers UTC."));
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Done);
let w_path = "/v3/workspaces/boxmaker0-trusted";
let seen: Vec<(String, Value)> = fake
.requests()
.into_iter()
.map(|r| (r.path, r.body.unwrap_or(Value::Null)))
.collect();
assert_eq!(
seen,
vec![
(
"/v3/workspaces".to_string(),
json!({ "id": "boxmaker0-trusted" })
),
(
format!("{w_path}/peers"),
json!({ "id": "boxmaker0", "configuration": { "observe_me": false } })
),
(format!("{w_path}/peers"), json!({ "id": "kyle" })),
(
format!("{w_path}/sessions"),
json!({ "id": "mm-a_memory", "peers": {
"boxmaker0": { "observe_me": false, "observe_others": false },
"kyle": { "observe_me": false, "observe_others": false },
} })
),
(
format!("{w_path}/conclusions"),
json!({ "conclusions": [{
"content": "Prefers UTC.",
"observer_id": "kyle",
"observed_id": "kyle",
"session_id": "mm-a_memory",
}] })
),
]
);
assert_eq!(
sent_ids(&home, &edit("e1", 0)),
Some(vec![CONCLUSION.to_string()])
);
let sync = home.sync.lock().unwrap();
assert_eq!(
sync.ledger().conclusion(Workspace::Trusted, "Prefers UTC."),
Some((edit("e1", 0), CONCLUSION.to_string()))
);
}
#[test]
fn a_replace_deletes_then_makes_and_an_undo_reverses_it() {
let fake = FakeHoncho::start();
let home = Home::new("edits-replace");
decide(&home, edit("e1", 0), conclude("kyle", "old"));
decide(&home, edit("e2", 0), unconclude("old"));
decide(&home, edit("e2", 1), conclude("kyle", "new"));
// The owner's undo of e2: the new deleted, the old made again.
decide(&home, edit("e3", 0), unconclude("new"));
decide(&home, edit("e3", 1), conclude("kyle", "old"));
let mut w = home.worker(&fake, FAST);
for _ in 0..5 {
assert_eq!(w.step(), Step::Done);
}
assert_eq!(w.step(), Step::Idle);
let order: Vec<(String, String)> = fake
.requests()
.into_iter()
.filter(|r| r.path.contains("/conclusions"))
.map(|r| {
let text = r.body.map_or(String::new(), |b| {
b["conclusions"][0]["content"].as_str().unwrap().to_string()
});
(r.method, text)
})
.collect();
let post = |t: &str| ("POST".to_string(), t.to_string());
let delete = ("DELETE".to_string(), String::new());
assert_eq!(
order,
vec![
post("old"),
delete.clone(),
post("new"),
delete,
post("old")
]
);
// Each deletion names the id its conclusion got, and records it.
let deletes = fake
.requests()
.into_iter()
.filter(|r| r.method == "DELETE")
.map(|r| r.path)
.collect::<Vec<_>>();
let path = format!("/v3/workspaces/boxmaker0-trusted/conclusions/{CONCLUSION}");
assert_eq!(deletes, vec![path.clone(), path]);
// The memory session, its workspace and peers are made sure of once in this run.
assert_eq!(posts(&fake, "/sessions").len(), 1);
assert_eq!(posts(&fake, "/peers").len(), 2);
assert_eq!(
sent_ids(&home, &edit("e2", 0)),
Some(vec![CONCLUSION.to_string()])
);
let sync = home.sync.lock().unwrap();
assert_eq!(sync.ledger().conclusion(Workspace::Trusted, "new"), None);
assert_eq!(
sync.ledger().conclusion(Workspace::Trusted, "old"),
Some((edit("e3", 1), CONCLUSION.to_string()))
);
}
#[test]
fn a_deletion_with_nothing_to_delete_is_sent_with_no_id() {
let fake = FakeHoncho::start();
let home = Home::new("edits-nothing");
decide(&home, edit("e1", 0), unconclude("never made"));
// A conclusion that failed is not live either.
decide(&home, edit("e2", 0), conclude("kyle", "refused"));
decide(&home, edit("e3", 0), unconclude("refused"));
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Done);
assert!(fake.requests().is_empty());
assert_eq!(sent_ids(&home, &edit("e1", 0)), Some(Vec::new()));
fake.set(Mode::Status(422));
assert_eq!(w.step(), Step::Done);
fake.set(Mode::Fixtures);
fake.clear();
assert_eq!(w.step(), Step::Done);
assert!(fake.requests().is_empty(), "{:?}", fake.requests());
assert_eq!(sent_ids(&home, &edit("e3", 0)), Some(Vec::new()));
assert_eq!(
home.lines().len(),
1,
"only the refusal: {:?}",
home.lines()
);
}
#[test]
fn a_404_on_delete_is_done() {
let fake = FakeHoncho::start();
let home = Home::new("edits-404");
decide(&home, edit("e1", 0), conclude("kyle", "x"));
decide(&home, edit("e2", 0), unconclude("x"));
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Done);
fake.set(Mode::Status(404));
assert_eq!(w.step(), Step::Done);
assert_eq!(
sent_ids(&home, &edit("e2", 0)),
Some(vec![CONCLUSION.to_string()])
);
assert!(home.lines().is_empty(), "{:?}", home.lines());
}
#[test]
fn an_untrusted_note_is_a_fleet_message_from_the_agent() {
let fake = FakeHoncho::start();
let home = Home::new("edits-fleet");
decide(&home, edit("e1", 0), fleet("The page says 3.2."));
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Done);
let seen: Vec<(String, Value)> = fake
.requests()
.into_iter()
.map(|r| (r.path, r.body.unwrap_or(Value::Null)))
.collect();
assert_eq!(seen.len(), 4);
assert_eq!(seen[0].1["configuration"]["reasoning"]["enabled"], false);
assert_eq!(seen[2].0, "/v3/workspaces/fleet/sessions");
assert_eq!(
seen[2].1,
json!({ "id": "mm-c", "peers": {
"boxmaker0": { "observe_me": false, "observe_others": false },
} })
);
assert_eq!(
seen[3],
(
"/v3/workspaces/fleet/sessions/mm-c/messages".to_string(),
json!({ "messages": [{
"content": "The page says 3.2.",
"peer_id": "boxmaker0",
"metadata": { "agent": "boxmaker0", "boxmaker_session": "mm-c", "edit": "e1" },
}] })
)
);
assert_eq!(sent_ids(&home, &edit("e1", 0)).map(|i| i.len()), Some(1));
// A long note in parts, in one batch.
decide(&home, edit("e2", 0), fleet(&"n".repeat(MAX_PART + 1)));
assert_eq!(w.step(), Step::Done);
let last = posts(&fake, "/messages").pop().unwrap().body.unwrap();
assert_eq!(last["messages"].as_array().unwrap().len(), 2);
}
#[test]
fn a_note_too_long_for_one_batch_is_marked_failed() {
let fake = FakeHoncho::start();
let home = Home::new("edits-long");
decide(&home, edit("e1", 0), fleet(&"n".repeat(MAX_PART * 100 + 1)));
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Done);
assert!(matches!(
home.records().last(),
Some(Record::Failed { status: None, kind, .. }) if kind == "too-long"
));
assert!(fake.requests().is_empty());
}
#[test]
fn the_text_is_neutralised() {
let fake = FakeHoncho::start();
let home = Home::new("edits-neutral");
decide(
&home,
edit("e1", 0),
conclude("boxmaker0", &format!("a{}b", TAGS[0].text)),
);
decide(&home, edit("e2", 0), fleet(&format!("c{}d", TAGS[1].text)));
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Done);
assert_eq!(w.step(), Step::Done);
let concluded = posts(&fake, "/conclusions")[0].body.clone().unwrap();
let content = concluded["conclusions"][0]["content"].as_str().unwrap();
assert!(
!content.contains(TAGS[0].text) && content.contains("&lt;"),
"{content}"
);
let message = posts(&fake, "/messages")[0].body.clone().unwrap();
let content = message["messages"][0]["content"].as_str().unwrap();
assert!(
!content.contains(TAGS[1].text) && content.contains("&lt;"),
"{content}"
);
}
#[test]
fn one_outbox_in_order_across_turns_and_edits() {
let fake = FakeHoncho::start();
let mut home = Home::new("edits-order");
home.log(
"mm-a",
&[
vec![spoken("one", Some("kyle")), assistant(2, Some("a"), &[])],
vec![spoken("two", Some("kyle")), assistant(2, Some("b"), &[])],
],
);
home.report("mm-a", 1);
decide(&home, edit("e1", 0), conclude("kyle", "Prefers UTC."));
home.report("mm-a", 2);
let mut w = home.worker(&fake, FAST);
for _ in 0..3 {
assert_eq!(w.step(), Step::Done);
}
let order: Vec<String> = fake
.requests()
.into_iter()
.filter(|r| r.path.ends_with("/messages") || r.path.ends_with("/conclusions"))
.map(|r| r.path)
.collect();
assert_eq!(
order,
vec![
"/v3/workspaces/boxmaker0-trusted/sessions/mm-a/messages".to_string(),
"/v3/workspaces/boxmaker0-trusted/conclusions".to_string(),
"/v3/workspaces/boxmaker0-trusted/sessions/mm-a/messages".to_string(),
]
);
// A restart sends nothing again.
home.restart();
let fake2 = FakeHoncho::start();
let mut w = home.worker(&fake2, FAST);
assert_eq!(w.step(), Step::Idle);
assert!(fake2.requests().is_empty());
}
#[test]
fn errors_as_for_turns() {
let fake = FakeHoncho::start();
let home = Home::new("edits-errors");
decide(&home, edit("e1", 0), conclude("kyle", "private-fact"));
decide(&home, edit("e2", 0), fleet("private-note"));
decide(&home, edit("e3", 0), conclude("kyle", "second"));
let mut w = home.worker(&fake, FAST);
fake.set(Mode::Status(503));
assert!(matches!(w.step(), Step::Backoff(_)));
fake.set(Mode::Status(422));
assert_eq!(w.step(), Step::Done);
assert!(matches!(
home.records().last(),
Some(Record::Failed { status: Some(422), kind, .. }) if kind == "status"
));
fake.set(Mode::Status(401));
assert_eq!(w.step(), Step::Done, "the fleet's key refused");
fake.set(Mode::Fixtures);
assert_eq!(w.step(), Step::Done, "the trusted workspace goes on");
assert!(sent_ids(&home, &edit("e3", 0)).is_some());
assert_eq!(w.step(), Step::Idle, "the fleet note waits");
let lines = home.lines();
assert_eq!(lines.len(), 4, "{lines:?}");
assert!(
lines[2].starts_with("memoryd: part 0 of edit e1 was not sent: "),
"{lines:?}"
);
assert!(lines[3].contains("workspace fleet"), "{lines:?}");
for line in lines {
assert!(!line.contains("private"), "{line}");
}
}
@@ -1,9 +1,9 @@
//! The Honcho worker's other duties (M5c spec, section 5): with nothing to send it decides what
//! `brokerd` could not answer for; after a restart nothing is sent again, and a session with a
//! decided turn waits again (a session without one does not); a running turn is never decided;
//! actions of later tasks wait; a ledger that cannot be written stops it; no log line holds a
//! turn's text. Then `memoryd` itself: a reported turn reaches Honcho, woken at once, and a key
//! that cannot be read stops the start. Do not edit.
//! a retraction waits (nothing decides one); a ledger that cannot be written stops it; no log
//! line holds a turn's text. Then `memoryd` itself: a reported turn reaches Honcho, woken at
//! once, and a key that cannot be read stops the start. Do not edit.
#[path = "support/brokerd.rs"]
mod brokerd;
@@ -150,15 +150,17 @@ fn a_restart_never_decides_a_running_turn() {
}
#[test]
fn actions_of_later_tasks_wait() {
fn a_retraction_waits() {
// Nothing decides a retraction any more (M5c spec, section 4, 2026-09-28): one in the
// ledger is neither sent nor failed.
let fake = FakeHoncho::start();
let mut home = Home::new("worker-later");
let conclude = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0001","part":0},"action":{"type":"conclude","workspace":"trusted","peer":"kyle","session":"mm-a","text":"a fact"}}"#;
std::fs::write(home.ledger_path(), format!("{conclude}\n")).unwrap();
let retract = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0001","part":0},"action":{"type":"retract","session":"mm-a","text":"a note"}}"#;
std::fs::write(home.ledger_path(), format!("{retract}\n")).unwrap();
home.restart();
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Idle);
assert_eq!(home.records().len(), 1, "the conclusion was failed or sent");
assert_eq!(home.records().len(), 1, "the retraction was failed or sent");
assert!(fake.requests().is_empty());
}
@@ -0,0 +1,327 @@
//! The Honcho worker's other duties (M5c spec, section 5): with nothing to send it decides what
//! `brokerd` could not answer for; after a restart nothing is sent again, and a session with a
//! decided turn waits again (a session without one does not); a running turn is never decided;
//! actions of later tasks wait; a ledger that cannot be written stops it; no log line holds a
//! turn's text. Then `memoryd` itself: a reported turn reaches Honcho, woken at once, and a key
//! that cannot be read stops the start. Do not edit.
#[path = "support/brokerd.rs"]
mod brokerd;
#[path = "support/honcho.rs"]
mod honcho;
#[path = "support/logs.rs"]
mod logs;
#[path = "support/tmp.rs"]
mod tmp;
#[path = "support/turns.rs"]
mod turns;
#[path = "support/worker.rs"]
mod worker;
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use brokerd::{Answer, FakeBrokerd};
use bxcore::{Envelope, Message, SessionId, SessionUpdated, read_frame, write_frame};
use honcho::{FakeHoncho, Mode};
use logs::assistant;
use memoryd::config::Config;
use memoryd::honcho::ledger::{Item, Record};
use memoryd::honcho::worker::{Step, Tuning};
use memoryd::serve::{self, StartError};
use tmp::TempDir;
use turns::{clean, running, sid, spoken};
use worker::{FAST, Home};
fn turn(text: &str, speaker: &str, reply: &str) -> Vec<bxcore::LogRecord> {
vec![spoken(text, Some(speaker)), assistant(2, Some(reply), &[])]
}
fn messages(fake: &FakeHoncho) -> usize {
fake.requests()
.iter()
.filter(|r| r.path.ends_with("/messages"))
.count()
}
fn is_sent(records: &[Record], session: &str, turn: u64) -> bool {
let item = Item::Turn {
session: sid(session),
turn,
};
records
.iter()
.any(|r| matches!(r, Record::Sent { item: i, .. } if *i == item))
}
fn decided(records: &[Record], session: &str, turn: u64) -> bool {
let item = Item::Turn {
session: sid(session),
turn,
};
records
.iter()
.any(|r| matches!(r, Record::Decided { item: i, .. } if *i == item))
}
#[test]
fn the_default_waits() {
assert_eq!(
Tuning::DEFAULT,
Tuning {
backoff: Duration::from_secs(5),
backoff_max: Duration::from_secs(600),
idle: Duration::from_secs(30),
}
);
}
#[test]
fn with_nothing_to_send_undecided_turns_are_decided() {
let fake = FakeHoncho::start();
let home = Home::new("worker-retry");
home.log("mm-a", &[turn("a", "kyle", "a")]);
*home.answer.lock().unwrap() = Err("brokerd is down".to_string());
home.report("mm-a", 1);
assert!(!decided(&home.records(), "mm-a", 1));
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Idle, "brokerd still down");
*home.answer.lock().unwrap() = Ok(clean());
assert_eq!(w.step(), Step::Done, "decided now");
assert!(decided(&home.records(), "mm-a", 1));
assert_eq!(w.step(), Step::Done, "sent");
assert!(is_sent(&home.records(), "mm-a", 1));
}
#[test]
fn an_outage_that_ends_is_followed_by_the_retry() {
let fake = FakeHoncho::start();
let home = Home::new("worker-after-outage");
home.log("mm-a", &[turn("a", "kyle", "a")]);
home.log("mm-b", &[turn("b", "kyle", "b")]);
home.report("mm-a", 1);
*home.answer.lock().unwrap() = Err("brokerd is down".to_string());
home.report("mm-b", 1);
let mut w = home.worker(&fake, FAST);
fake.set(Mode::Status(502));
assert!(matches!(w.step(), Step::Backoff(_)));
*home.answer.lock().unwrap() = Ok(clean());
fake.set(Mode::Fixtures);
assert_eq!(w.step(), Step::Done);
// mm-a is sent, and the outage's end decided mm-b before any idle step.
assert!(decided(&home.records(), "mm-b", 1));
}
#[test]
fn a_restart_sends_nothing_again_and_resumes_the_sessions_it_knows() {
let fake = FakeHoncho::start();
let mut home = Home::new("worker-restart");
home.log("mm-a", &[turn("a", "kyle", "a")]);
home.report("mm-a", 1);
home.worker(&fake, FAST).step();
assert_eq!(messages(&fake), 1);
// While memoryd is down: mm-a has a second turn, never reported; mm-z, never seen, one.
home.log("mm-a", &[turn("a", "kyle", "a"), turn("b", "kyle", "b")]);
home.log("mm-z", &[turn("z", "kyle", "z")]);
home.restart();
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Done, "mm-a's second turn decided");
assert_eq!(w.step(), Step::Done, "and sent");
assert_eq!(w.step(), Step::Idle);
let records = home.records();
assert!(is_sent(&records, "mm-a", 2));
assert!(!decided(&records, "mm-z", 1), "a session never reported");
assert_eq!(messages(&fake), 2, "turn 1 was sent again");
}
#[test]
fn a_restart_never_decides_a_running_turn() {
let fake = FakeHoncho::start();
let mut home = Home::new("worker-running");
home.log("mm-a", &[turn("a", "kyle", "a")]);
home.report("mm-a", 1);
home.worker(&fake, FAST).step();
home.log("mm-a", &[turn("a", "kyle", "a"), running(Some("kyle"))]);
home.restart();
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Idle);
assert!(!decided(&home.records(), "mm-a", 2));
}
#[test]
fn actions_of_later_tasks_wait() {
let fake = FakeHoncho::start();
let mut home = Home::new("worker-later");
let conclude = r#"{"type":"decided","time":"2026-09-28T10:00:00.000Z","item":{"type":"edit","id":"2026-09-28-0001","part":0},"action":{"type":"conclude","workspace":"trusted","peer":"kyle","session":"mm-a","text":"a fact"}}"#;
std::fs::write(home.ledger_path(), format!("{conclude}\n")).unwrap();
home.restart();
let mut w = home.worker(&fake, FAST);
assert_eq!(w.step(), Step::Idle);
assert_eq!(home.records().len(), 1, "the conclusion was failed or sent");
assert!(fake.requests().is_empty());
}
#[test]
fn a_ledger_that_cannot_be_written_stops_the_worker() {
let fake = FakeHoncho::start();
let home = Home::new("worker-ledger");
home.log("mm-a", &[turn("a", "kyle", "a")]);
home.report("mm-a", 1);
let mut w = home.worker(&fake, FAST);
let ledger = home.ledger_path();
let kept = std::fs::read_to_string(&ledger).unwrap();
std::fs::remove_file(&ledger).unwrap();
std::fs::create_dir(&ledger).unwrap();
assert_eq!(w.step(), Step::Stopped);
assert_eq!(w.step(), Step::Stopped);
assert_eq!(
messages(&fake),
1,
"sent once; its sent line could not be written"
);
let lines = home.lines();
assert_eq!(lines.len(), 1, "{lines:?}");
assert!(lines[0].ends_with(
"; the Honcho worker stops until memoryd restarts; see docs/runbook.md#honcho-ledger-damaged"
));
std::fs::remove_dir(&ledger).unwrap();
std::fs::write(&ledger, kept).unwrap();
}
#[test]
fn no_log_line_holds_a_turns_text() {
let fake = FakeHoncho::start();
let home = Home::new("worker-quiet");
for (session, status) in [("mm-a", 503), ("mm-b", 422), ("mm-c", 401)] {
home.log(
session,
&[turn("private-question", "kyle", "private-answer")],
);
home.report(session, 1);
let mut w = home.worker(&fake, FAST);
fake.set(Mode::Status(status));
w.step();
}
let lines = home.lines();
assert_eq!(lines.len(), 3, "{lines:?}");
for line in lines {
assert!(!line.contains("private"), "{line}");
}
}
/// `memoryd serve`'s config for `home`, with `[honcho]` at `url`.
fn serve_config(home: &TempDir, url: &str, extra: &str) -> String {
let soul = home.write("etc/soul.md", "You are Boxmaker.\n");
format!(
"[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n[honcho]\nurl = \"{url}\"\nagent = \"boxmaker0\"\nowner = \"kyle\"\ntrusted = \"boxmaker0-trusted\"\nuntrusted = \"boxmaker0-untrusted\"\nfleet = \"fleet\"\n{extra}",
home.path().display(),
soul.display()
)
}
#[test]
fn memoryd_sends_a_reported_turn_at_once() {
let fake = FakeHoncho::start();
let dir = TempDir::new("worker-serve");
turns::write_turns(dir.path(), "mm-a", &[turn("a", "kyle", "a")]);
let config = Config::parse(&serve_config(&dir, &fake.url, "")).unwrap();
let _brokerd = FakeBrokerd::start(&config.state_socket(), Answer::State(clean()));
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());
// The worker has slept since start (30 s when idle): the report wakes it.
std::thread::sleep(Duration::from_millis(100));
let mut stream = std::os::unix::net::UnixStream::connect(config.loop_socket()).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(10)))
.unwrap();
let report = Envelope {
v: 1,
id: 1,
r#final: true,
msg: Message::SessionUpdated(SessionUpdated {
session: SessionId::new("mm-a").unwrap(),
turn: 1,
}),
};
write_frame(&mut stream, &report).unwrap();
assert!(matches!(
read_frame(&mut stream).unwrap().msg,
Message::Ok(_)
));
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline && messages(&fake) == 0 {
std::thread::sleep(Duration::from_millis(10));
}
assert_eq!(messages(&fake), 1, "{:?}", lines.lock().unwrap());
}
#[test]
fn a_key_that_cannot_be_read_stops_the_start() {
let fake = FakeHoncho::start();
let dir = TempDir::new("worker-key-start");
let text = serve_config(
&dir,
&fake.url,
"[secrets.honcho_trusted]\nenv = \"MEMORYD_TEST_UNSET_KEY\"\n",
);
let config = Config::parse(&text).unwrap();
let log: serve::Log = Arc::new(|_: &str| {});
let err = match serve::start(config.clone(), log) {
Err(err) => err,
Ok(_) => panic!("memoryd started"),
};
assert!(matches!(err, StartError::Secret(_)), "{err:?}");
assert!(!config.loop_socket().exists(), "a socket was bound");
// The process: exit 1 with the secret's own lines and pointer, not the start pointer.
let path = dir.write("memoryd.toml", &text);
let mut child = Daemon(
Command::new(env!("CARGO_BIN_EXE_memoryd"))
.args(["serve", "--config"])
.arg(&path)
.env_remove("MEMORYD_TEST_UNSET_KEY")
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::piped())
.spawn()
.unwrap(),
);
let deadline = Instant::now() + Duration::from_secs(10);
let mut code = None;
while Instant::now() < deadline {
if let Ok(Some(status)) = child.0.try_wait() {
code = status.code();
break;
}
std::thread::sleep(Duration::from_millis(20));
}
let _ = child.0.kill();
let _ = child.0.wait();
let mut stderr = String::new();
if let Some(mut pipe) = child.0.stderr.take() {
let _ = std::io::Read::read_to_string(&mut pipe, &mut stderr);
}
assert_eq!(code, Some(1), "{stderr}");
assert!(
stderr.starts_with("memoryd: secret honcho_trusted: "),
"{stderr}"
);
assert!(
stderr.ends_with("\nsee docs/runbook.md#secret-unavailable\n"),
"{stderr}"
);
assert!(!stderr.contains("memoryd-start-failed"), "{stderr}");
}
/// A `memoryd` process, killed and waited for when dropped.
struct Daemon(std::process::Child);
impl Drop for Daemon {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
@@ -1,6 +1,7 @@
//! A fake Honcho for tests, on 127.0.0.1 and a port of its own: it answers each request from the
//! responses recorded from rift (`tests/fixtures/honcho/`), by method and path, and records every
//! request, body included. It can be set to answer every request with a status instead, or to
//! responses recorded from rift (`tests/fixtures/honcho/`), by method and path (a batch of
//! messages or conclusions answered with as many as it asked for, up to the two recorded), and
//! records every request, body included. It can be set to answer every request with a status instead, or to
//! answer nothing. Stopped, and its held connections released, when dropped; every wait has a
//! deadline. Do not edit.
@@ -214,8 +215,13 @@ fn answer(request: &Request, made: &Mutex<HashSet<String>>) -> (u16, String) {
}
("POST", ["v3", "workspaces", _, "sessions"]) => fixture("06-session-create.json"),
("POST", ["v3", "workspaces", _, "sessions", _, "messages"]) => {
fixture("08-messages-add-batch.json")
first(fixture("08-messages-add-batch.json"), request, "messages")
}
("POST", ["v3", "workspaces", _, "conclusions"]) => first(
fixture("11-conclusions-create-batch.json"),
request,
"conclusions",
),
("DELETE", ["v3", "workspaces", _, "conclusions", _]) if again => {
fixture("16-conclusion-delete-again.json")
}
@@ -226,6 +232,19 @@ fn answer(request: &Request, made: &Mutex<HashSet<String>>) -> (u16, String) {
}
}
/// A recorded batch answer cut to as many items as the request's `field` holds (the recorded ones
/// hold two), so that one message or conclusion gets one id.
fn first(recorded: (u16, String), request: &Request, field: &str) -> (u16, String) {
let asked = request
.body
.as_ref()
.and_then(|b| b[field].as_array())
.map_or(0, Vec::len);
let mut items: Vec<Value> = serde_json::from_str(&recorded.1).unwrap();
items.truncate(asked);
(recorded.0, Value::Array(items).to_string())
}
fn serve(mut stream: TcpStream, shared: &Shared) {
let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
let Some(request) = read_request(&mut stream) else {
@@ -0,0 +1,260 @@
//! A fake Honcho for tests, on 127.0.0.1 and a port of its own: it answers each request from the
//! responses recorded from rift (`tests/fixtures/honcho/`), by method and path, and records every
//! request, body included. It can be set to answer every request with a status instead, or to
//! answer nothing. Stopped, and its held connections released, when dropped; every wait has a
//! deadline. Do not edit.
#![allow(dead_code)] // each test file uses a different part of this module
use std::collections::HashSet;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
use serde_json::Value;
/// One request as the fake read it.
#[derive(Debug, Clone, PartialEq)]
pub struct Request {
pub method: String,
pub path: String,
/// Header names in lower case.
pub headers: Vec<(String, String)>,
pub body: Option<Value>,
}
impl Request {
pub fn header(&self, name: &str) -> Option<&str> {
self.headers
.iter()
.find(|(k, _)| k == name)
.map(|(_, v)| v.as_str())
}
}
/// How the fake answers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Mode {
/// From the fixtures.
Fixtures,
/// Every request with this status. A 422 echoes the request's body, as Honcho does.
Status(u16),
/// Read the request, then answer nothing until the fake is dropped (at most 30 s).
Hang,
}
struct Shared {
mode: Mutex<Mode>,
requests: Mutex<Vec<Request>>,
/// What was made, by method and path and id: a second create answers 200, a second delete
/// 404.
made: Mutex<HashSet<String>>,
stop: AtomicBool,
}
pub struct FakeHoncho {
pub url: String,
port: u16,
shared: Arc<Shared>,
accept: Option<JoinHandle<()>>,
}
/// A recorded fixture: its response's status and body.
fn fixture(name: &str) -> (u16, String) {
let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("tests/fixtures/honcho")
.join(name);
let text = std::fs::read_to_string(&path).unwrap();
let value: Value = serde_json::from_str(&text).unwrap();
let status = u16::try_from(value["response"]["status"].as_u64().unwrap()).unwrap();
let body = match &value["response"]["body"] {
Value::String(s) if s.is_empty() => String::new(),
other => other.to_string(),
};
(status, body)
}
impl FakeHoncho {
pub fn start() -> FakeHoncho {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let shared = Arc::new(Shared {
mode: Mutex::new(Mode::Fixtures),
requests: Mutex::new(Vec::new()),
made: Mutex::new(HashSet::new()),
stop: AtomicBool::new(false),
});
let for_accept = Arc::clone(&shared);
let accept = std::thread::spawn(move || {
for stream in listener.incoming() {
if for_accept.stop.load(Ordering::SeqCst) {
return;
}
let Ok(stream) = stream else { return };
let shared = Arc::clone(&for_accept);
std::thread::spawn(move || serve(stream, &shared));
}
});
FakeHoncho {
url: format!("http://127.0.0.1:{port}"),
port,
shared,
accept: Some(accept),
}
}
pub fn set(&self, mode: Mode) {
*self.shared.mode.lock().unwrap() = mode;
}
/// Every request read so far, in order.
pub fn requests(&self) -> Vec<Request> {
self.shared.requests.lock().unwrap().clone()
}
/// Forget the requests read so far (not what was made).
pub fn clear(&self) {
self.shared.requests.lock().unwrap().clear();
}
/// Waits up to 10 s until at least `n` requests were read; true if they were.
pub fn wait_for(&self, n: usize) -> bool {
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline {
if self.requests().len() >= n {
return true;
}
std::thread::sleep(Duration::from_millis(10));
}
false
}
}
impl Drop for FakeHoncho {
fn drop(&mut self) {
self.shared.stop.store(true, Ordering::SeqCst);
let _ = TcpStream::connect(("127.0.0.1", self.port));
if let Some(accept) = self.accept.take() {
let _ = accept.join();
}
}
}
/// Reads one request: the head up to its blank line, then `Content-Length` bytes of body.
fn read_request(stream: &mut TcpStream) -> Option<Request> {
let mut buf = Vec::new();
let mut byte = [0u8; 1];
while !buf.ends_with(b"\r\n\r\n") {
if stream.read(&mut byte).ok()? == 0 {
return None;
}
buf.push(byte[0]);
}
let head = String::from_utf8(buf).ok()?;
let mut lines = head.split("\r\n");
let mut first = lines.next()?.split(' ');
let (method, path) = (first.next()?.to_string(), first.next()?.to_string());
let headers: Vec<(String, String)> = lines
.filter_map(|l| l.split_once(": "))
.map(|(k, v)| (k.to_ascii_lowercase(), v.to_string()))
.collect();
let length: usize = headers
.iter()
.find(|(k, _)| k == "content-length")
.map_or(Some(0), |(_, v)| v.parse().ok())?;
let mut body = vec![0u8; length];
stream.read_exact(&mut body).ok()?;
let body = if length == 0 {
None
} else {
serde_json::from_slice(&body).ok()
};
Some(Request {
method,
path,
headers,
body,
})
}
/// The fixture that answers `request`, by its method and the shape of its path; the second of a
/// pair when the same thing was made (or deleted) before.
fn answer(request: &Request, made: &Mutex<HashSet<String>>) -> (u16, String) {
let parts: Vec<&str> = request.path.trim_start_matches('/').split('/').collect();
let id = request
.body
.as_ref()
.and_then(|b| b["id"].as_str())
.unwrap_or_default()
.to_string();
let again = !made
.lock()
.unwrap()
.insert(format!("{} {} {id}", request.method, request.path));
let configured = request
.body
.as_ref()
.is_some_and(|b| b.get("configuration").is_some());
match (request.method.as_str(), parts.as_slice()) {
("POST", ["v3", "workspaces"]) if again => fixture("02-workspace-create-again.json"),
("POST", ["v3", "workspaces"]) => fixture("01-workspace-create.json"),
("POST", ["v3", "workspaces", _, "peers"]) if again => {
fixture("05-peer-create-boxmaker0-again.json")
}
("POST", ["v3", "workspaces", _, "peers"]) if configured => {
fixture("04-peer-create-boxmaker0.json")
}
("POST", ["v3", "workspaces", _, "peers"]) => fixture("03-peer-create-kyle.json"),
("POST", ["v3", "workspaces", _, "sessions"]) if again => {
fixture("07-session-create-again.json")
}
("POST", ["v3", "workspaces", _, "sessions"]) => fixture("06-session-create.json"),
("POST", ["v3", "workspaces", _, "sessions", _, "messages"]) => {
fixture("08-messages-add-batch.json")
}
("DELETE", ["v3", "workspaces", _, "conclusions", _]) if again => {
fixture("16-conclusion-delete-again.json")
}
("DELETE", ["v3", "workspaces", _, "conclusions", _]) => {
fixture("15-conclusion-delete.json")
}
_ => (404, r#"{"detail":"Not Found"}"#.to_string()),
}
}
fn serve(mut stream: TcpStream, shared: &Shared) {
let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
let Some(request) = read_request(&mut stream) else {
return;
};
shared.requests.lock().unwrap().push(request.clone());
let mode = *shared.mode.lock().unwrap();
let (status, body) = match mode {
Mode::Fixtures => answer(&request, &shared.made),
Mode::Status(422) => {
let input = request.body.clone().unwrap_or(Value::Null);
(
422,
serde_json::json!({ "detail": [{ "input": input }] }).to_string(),
)
}
Mode::Status(status) => (status, r#"{"detail":"refused"}"#.to_string()),
Mode::Hang => {
let deadline = Instant::now() + Duration::from_secs(30);
while !shared.stop.load(Ordering::SeqCst) && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(10));
}
return;
}
};
let head = format!(
"HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = stream.write_all(head.as_bytes());
let _ = stream.write_all(body.as_bytes());
}