bxctl memory sync: the Honcho ledger's counts and failed items

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
This commit is contained in:
2026-09-28 13:22:26 -07:00
co-authored by ornith-1.5-35b-a3b
parent fb05b26462
commit fb2afa1979
11 changed files with 445 additions and 12 deletions
+2
View File
@@ -282,6 +282,8 @@ pub enum MemoryCommand {
Pending,
Accept,
Reject,
/// What the Honcho ledger holds: its counts and failed items (M5c spec, section 5).
Sync,
}
/// `bxctl` → `memoryd`: one command; `arg` is the count for `log`, the edit or pending id for
+2 -1
View File
@@ -34,8 +34,9 @@ usage: bxctl <command> [options]
bxctl reindex [--memory-socket PATH]
Rebuild memoryd's index from the session logs and notes.
bxctl memory log [N] | undo <id> | pending | accept <id> | reject <id> [--memory-socket PATH]
bxctl memory log [N] | undo <id> | pending | accept <id> | reject <id> | sync [--memory-socket PATH]
List memory edits, undo the latest edit of a file, or answer writes waiting for review.
sync shows what memoryd sends to Honcho: counts, and each failed item by its label.
bxctl sanitize check [TEXT]
Show which sanitizer rules TEXT, or standard input, breaks; writes nothing. Refuse shows
+4 -3
View File
@@ -1,5 +1,5 @@
//! The owner's memory commands over `memoryd`'s owner socket: `reindex`, and `memory log`,
//! `undo`, `pending`, `accept`, `reject` (M5a spec, section 10).
//! `undo`, `pending`, `accept`, `reject` (M5a spec, section 10), and `sync` (M5c spec, section 5).
use std::io::Write;
use std::os::unix::net::UnixStream;
@@ -39,8 +39,8 @@ pub fn parse_reindex(args: &[String], home: &Path) -> Result<Command, UsageError
})
}
/// `bxctl memory log [N] | undo <id> | pending | accept <id> | reject <id>`, each with an optional
/// `--memory-socket PATH`.
/// `bxctl memory log [N] | undo <id> | pending | accept <id> | reject <id> | sync`, each with an
/// optional `--memory-socket PATH`.
pub fn parse_memory(args: &[String], home: &Path) -> Result<Command, UsageError> {
let (socket, rest) = socket_flag(args, home)?;
let rest: Vec<&str> = rest.iter().map(String::as_str).collect();
@@ -56,6 +56,7 @@ pub fn parse_memory(args: &[String], home: &Path) -> Result<Command, UsageError>
["undo", id] => Ask::Admin(MemoryCommand::Undo, id.to_string()),
["accept", id] => Ask::Admin(MemoryCommand::Accept, id.to_string()),
["reject", id] => Ask::Admin(MemoryCommand::Reject, id.to_string()),
["sync"] => Ask::Admin(MemoryCommand::Sync, String::new()),
_ => return Err(UsageError),
};
Ok(Command::Memory { socket, ask })
+108
View File
@@ -0,0 +1,108 @@
//! `bxctl memory sync` (M5c spec, section 5): parsed, sent to `memoryd` as the `sync` command,
//! its answer printed as it comes, and `memoryd` down reported as for every memory command. Do
//! not edit.
#[path = "support/temp.rs"]
mod temp;
use std::os::unix::net::UnixListener;
use std::path::{Path, PathBuf};
use std::sync::mpsc;
use std::time::Duration;
use bxcore::{
Envelope, MemoryAdmin, MemoryAdminReply, MemoryCommand, Message, read_frame, write_frame,
};
use bxctl::cli::{Command, USAGE, UsageError, parse};
use bxctl::memory::{Ask, run};
fn args(line: &str) -> Vec<String> {
line.split_whitespace().map(str::to_string).collect()
}
/// A fake `memoryd` answering one request with `text`; hands back what it was sent.
fn fake(socket: &Path, text: &str) -> mpsc::Receiver<Message> {
let listener = UnixListener::bind(socket).unwrap();
let (tx, rx) = mpsc::channel();
let answer = Message::MemoryAdminReply(MemoryAdminReply {
text: text.to_string(),
});
std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let env = read_frame(&mut stream).unwrap();
tx.send(env.msg).unwrap();
let reply = Envelope {
v: 1,
id: env.id,
r#final: true,
msg: answer,
};
write_frame(&mut stream, &reply).unwrap();
});
rx
}
#[test]
fn sync_parses() {
let home = PathBuf::from("/srv/bx");
assert_eq!(
parse(&args("memory sync"), &home),
Ok(Command::Memory {
socket: PathBuf::from("/srv/bx/run/owner-memory/admin.sock"),
ask: Ask::Admin(MemoryCommand::Sync, String::new()),
})
);
assert_eq!(
parse(&args("memory sync --memory-socket /r/m.sock"), &home),
Ok(Command::Memory {
socket: PathBuf::from("/r/m.sock"),
ask: Ask::Admin(MemoryCommand::Sync, String::new()),
})
);
assert_eq!(parse(&args("memory sync now"), &home), Err(UsageError));
assert!(
USAGE.contains("| reject <id> | sync [--memory-socket PATH]\n"),
"{USAGE}"
);
}
#[test]
fn the_command_is_named_sync_on_the_wire() {
assert_eq!(
serde_json::to_string(&MemoryCommand::Sync).unwrap(),
"\"sync\""
);
}
#[test]
fn the_answer_is_printed_as_it_comes() {
let dir = temp::TempDir::new("bx-mem-sync");
let socket = dir.join("admin.sock");
let report = "honcho sync: 3 decided: 1 sent, 1 waiting, 1 failed, 0 skipped\n\
failed, oldest first:\n \
2026-09-28T10:00:00.000Z turn 3 of session mm-a status (HTTP 422)";
let seen = fake(&socket, report);
let mut out = Vec::new();
let ask = Ask::Admin(MemoryCommand::Sync, String::new());
run(&socket, &ask, Duration::from_secs(5), &mut out).unwrap();
assert_eq!(String::from_utf8(out).unwrap(), format!("{report}\n"));
assert_eq!(
seen.recv_timeout(Duration::from_secs(5)).unwrap(),
Message::MemoryAdmin(MemoryAdmin {
command: MemoryCommand::Sync,
arg: String::new(),
})
);
}
#[test]
fn memoryd_down_is_reported_as_for_every_memory_command() {
let dir = temp::TempDir::new("bx-mem-sync-down");
let ask = Ask::Admin(MemoryCommand::Sync, String::new());
let socket = dir.join("nobody.sock");
let err = run(&socket, &ask, Duration::from_secs(1), &mut Vec::new()).unwrap_err();
assert!(
err.starts_with(&format!("cannot reach memoryd at {}: ", socket.display())),
"{err}"
);
}
+48 -8
View File
@@ -94,6 +94,15 @@ pub struct Counts {
pub skipped: usize,
}
/// A failed item, as its `Failed` record says.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Failure {
pub item: Item,
pub time: Timestamp,
pub status: Option<u16>,
pub kind: String,
}
#[derive(Debug)]
pub enum LedgerError {
Io(PathBuf, std::io::Error),
@@ -116,9 +125,9 @@ impl std::fmt::Display for LedgerError {
write!(f, "{}:{line}: {why}; {RUNBOOK}", path.display())
}
LedgerError::Encode(err) => write!(f, "a ledger record could not be encoded: {err}"),
LedgerError::AlreadyDecided(item) => write!(f, "{}: decided already", describe(item)),
LedgerError::AlreadyDecided(item) => write!(f, "{}: decided already", label(item)),
LedgerError::NotWaiting(item) => {
write!(f, "{}: not waiting to be sent", describe(item))
write!(f, "{}: not waiting to be sent", label(item))
}
}
}
@@ -132,7 +141,11 @@ enum State {
Skipped,
Waiting,
Sent(Vec<String>),
Failed,
Failed {
time: Timestamp,
status: Option<u16>,
kind: String,
},
}
#[derive(Debug, Clone)]
@@ -150,6 +163,8 @@ pub struct Ledger {
entries: HashMap<Item, Entry>,
/// Every sent item, in the order of its `Sent` record.
sends: Vec<Item>,
/// Every failed item, in the order of its `Failed` record.
fails: Vec<Item>,
repaired: bool,
}
@@ -163,6 +178,7 @@ impl Ledger {
order: Vec::new(),
entries: HashMap::new(),
sends: Vec::new(),
fails: Vec::new(),
repaired: false,
};
let Some(bytes) = read(path)? else {
@@ -267,6 +283,23 @@ impl Ledger {
live.pop()
}
/// Every failed item, oldest first (the order of the `Failed` records), with its record's
/// time, status and kind.
pub fn failures(&self) -> Vec<Failure> {
self.fails
.iter()
.filter_map(|item| match &self.entries.get(item)?.state {
State::Failed { time, status, kind } => Some(Failure {
item: item.clone(),
time: *time,
status: *status,
kind: kind.clone(),
}),
_ => None,
})
.collect()
}
/// How many items are in each state.
pub fn counts(&self) -> Counts {
let mut counts = Counts::default();
@@ -276,7 +309,7 @@ impl Ledger {
State::Skipped => counts.skipped += 1,
State::Waiting => counts.waiting += 1,
State::Sent(_) => counts.sent += 1,
State::Failed => counts.failed += 1,
State::Failed { .. } => counts.failed += 1,
}
}
counts
@@ -318,10 +351,16 @@ impl Ledger {
}
self.sends.push(item);
}
Record::Failed { item, .. } => {
Record::Failed {
time,
item,
status,
kind,
} => {
if let Some(entry) = self.entries.get_mut(&item) {
entry.state = State::Failed;
entry.state = State::Failed { time, status, kind };
}
self.fails.push(item);
}
}
}
@@ -375,8 +414,9 @@ fn cut(path: &Path, keep: usize) -> Result<(), LedgerError> {
file.sync_data().map_err(io)
}
/// An item as the error messages name it.
fn describe(item: &Item) -> String {
/// An item as the error messages and `bxctl memory sync` name it: `turn 3 of session mm-a`,
/// `part 0 of edit 2026-09-28-0001`.
pub fn label(item: &Item) -> String {
match item {
Item::Turn { session, turn } => format!("turn {turn} of session {}", session.as_str()),
Item::Edit { id, part } => format!("part {part} of edit {id}"),
+6
View File
@@ -8,6 +8,7 @@ use bxcore::{SessionId, Timestamp};
use crate::files::Store;
use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace};
use crate::honcho::report;
use crate::journal;
use crate::notes::{self, Kind};
@@ -167,6 +168,11 @@ impl Mirror {
}
}
/// `bxctl memory sync`'s report of the ledger, under the ledger's lock alone.
pub fn report(&self) -> String {
report::report(&self.ledger.lock().unwrap_or_else(|p| p.into_inner()))
}
/// 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. Its parts are
/// `Edit { id, part }` from 0. An edit already decided, or asking nothing, adds no line. A
+1
View File
@@ -7,6 +7,7 @@ pub mod edits;
pub mod ledger;
pub mod mirror;
pub mod naming;
pub mod report;
pub mod start;
pub mod state;
pub mod turns;
+33
View File
@@ -0,0 +1,33 @@
//! What `bxctl memory sync` prints (M5c spec, section 5): how many items the ledger holds in each
//! state, and every failed item, oldest first, by its label, never by its text.
use crate::honcho::ledger::{Ledger, label};
/// The answer without `[honcho]`: not an error.
pub const NOT_CONFIGURED: &str = "honcho sync: not configured (memoryd.toml has no [honcho])";
/// The report of `ledger`: one line of counts, then, when any item failed, a heading and one line
/// per failed item: its time, its label, its kind, and its HTTP status when it had one.
pub fn report(ledger: &Ledger) -> String {
let c = ledger.counts();
let mut out = format!(
"honcho sync: {} decided: {} sent, {} waiting, {} failed, {} skipped",
c.decided, c.sent, c.waiting, c.failed, c.skipped
);
let failures = ledger.failures();
if !failures.is_empty() {
out.push_str("\nfailed, oldest first:");
for f in failures {
out.push_str(&format!(
"\n {} {} {}",
f.time.to_rfc3339(),
label(&f.item),
f.kind
));
if let Some(status) = f.status {
out.push_str(&format!(" (HTTP {status})"));
}
}
}
out
}
+5
View File
@@ -18,6 +18,7 @@ use crate::edit::{self, Context};
use crate::files::Store;
use crate::honcho::ledger::{self, LedgerError};
use crate::honcho::mirror::Mirror;
use crate::honcho::report;
use crate::honcho::state;
use crate::honcho::turns::TurnSync;
use crate::honcho::worker::{self, Wake};
@@ -454,6 +455,10 @@ impl Memoryd {
MemoryCommand::Pending => review::list(&self.store),
MemoryCommand::Accept => review::accept(&self.store, arg, now),
MemoryCommand::Reject => review::reject(&self.store, arg, now),
MemoryCommand::Sync => Ok(match &self.mirror {
Some(mirror) => mirror.report(),
None => report::NOT_CONFIGURED.to_string(),
}),
};
if result.is_ok() && matches!(command, MemoryCommand::Undo | MemoryCommand::Accept) {
self.mirror(None);
+235
View File
@@ -0,0 +1,235 @@
//! What `bxctl memory sync` prints (M5c spec, section 5): the ledger's counts, and each failed
//! item, oldest first, by its label, its kind and its status, exactly; the same after a restart;
//! and `memoryd` answering it, with `[honcho]` and without. Do not edit.
#[path = "support/tmp.rs"]
mod tmp;
use std::os::unix::net::UnixStream;
use std::sync::Arc;
use std::time::Duration;
use bxcore::{
CallId, DataClass, Envelope, MemoryAdmin, MemoryCall, MemoryCommand, Message, SessionId,
Timestamp, ToolRequest, read_frame, write_frame,
};
use memoryd::config::Config;
use memoryd::honcho::ledger::{Action, Failure, Item, Ledger, Record, Workspace, label};
use memoryd::honcho::report::{NOT_CONFIGURED, report};
use memoryd::serve;
use tmp::TempDir;
fn at(minute: u32) -> Timestamp {
Timestamp::parse(&format!("2026-09-28T10:{minute:02}:00.000Z")).unwrap()
}
fn turn(n: u64) -> Item {
Item::Turn {
session: SessionId::new("mm-a").unwrap(),
turn: n,
}
}
fn edit(id: &str) -> Item {
Item::Edit {
id: id.to_string(),
part: 0,
}
}
fn to_trusted() -> Action {
Action::Turn {
workspace: Workspace::Trusted,
speaker: "kyle".to_string(),
}
}
/// A ledger with every state: 2 sent, 1 waiting, 2 failed (the later-decided one failing first),
/// 1 skipped.
fn mixed(dir: &TempDir) -> Ledger {
let mut l = Ledger::open(&dir.path().join("honcho.jsonl")).unwrap();
let decided = |item: Item, action: Action| Record::Decided {
time: at(0),
item,
action,
};
let records = vec![
decided(turn(1), to_trusted()),
decided(turn(2), to_trusted()),
decided(
edit("2026-09-28-0001"),
Action::Conclude {
workspace: Workspace::Trusted,
peer: "kyle".to_string(),
session: SessionId::new("mm-a").unwrap(),
text: "a private fact".to_string(),
},
),
decided(
turn(3),
Action::Skip {
why: "the turn has no single speaker".to_string(),
},
),
decided(turn(4), to_trusted()),
decided(turn(5), to_trusted()),
decided(turn(6), to_trusted()),
Record::Sent {
time: at(1),
item: turn(1),
ids: vec!["m1".to_string()],
},
Record::Failed {
time: at(2),
item: turn(4),
status: None,
kind: "log".to_string(),
},
Record::Failed {
time: at(3),
item: edit("2026-09-28-0001"),
status: Some(422),
kind: "status".to_string(),
},
Record::Sent {
time: at(4),
item: turn(2),
ids: vec!["m2".to_string()],
},
];
for r in records {
l.append(r).unwrap();
}
l
}
const MIXED: &str = "honcho sync: 7 decided: 2 sent, 2 waiting, 2 failed, 1 skipped
failed, oldest first:
2026-09-28T10:02:00.000Z turn 4 of session mm-a log
2026-09-28T10:03:00.000Z part 0 of edit 2026-09-28-0001 status (HTTP 422)";
#[test]
fn the_report_is_exactly_this() {
let dir = TempDir::new("sync-report");
let l = mixed(&dir);
assert_eq!(report(&l), MIXED);
assert!(!report(&l).contains("private"), "no text is printed");
let empty = Ledger::open(&dir.path().join("none.jsonl")).unwrap();
assert_eq!(
report(&empty),
"honcho sync: 0 decided: 0 sent, 0 waiting, 0 failed, 0 skipped"
);
assert_eq!(
NOT_CONFIGURED,
"honcho sync: not configured (memoryd.toml has no [honcho])"
);
}
#[test]
fn the_failures_are_oldest_first_and_survive_a_restart() {
let dir = TempDir::new("sync-failures");
let l = mixed(&dir);
let expected = vec![
Failure {
item: turn(4),
time: at(2),
status: None,
kind: "log".to_string(),
},
Failure {
item: edit("2026-09-28-0001"),
time: at(3),
status: Some(422),
kind: "status".to_string(),
},
];
assert_eq!(l.failures(), expected);
let again = Ledger::open(&dir.path().join("honcho.jsonl")).unwrap();
assert_eq!(again.failures(), expected);
assert_eq!(report(&again), MIXED);
assert_eq!(label(&turn(4)), "turn 4 of session mm-a");
assert_eq!(label(&edit("e")), "part 0 of edit e");
}
const SECTION: &str = "[honcho]
url = \"http://127.0.0.1:9\"
agent = \"boxmaker0\"
owner = \"kyle\"
trusted = \"boxmaker0-trusted\"
untrusted = \"boxmaker0-untrusted\"
fleet = \"fleet\"
";
fn ask(socket: &std::path::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: 5,
r#final: true,
msg,
};
write_frame(&mut stream, &env).unwrap();
read_frame(&mut stream).unwrap().msg
}
fn sync(config: &Config) -> String {
let msg = Message::MemoryAdmin(MemoryAdmin {
command: MemoryCommand::Sync,
arg: String::new(),
});
match ask(&config.admin_socket(), msg) {
Message::MemoryAdminReply(reply) => reply.text,
other => panic!("{other:?}"),
}
}
fn serve(dir: &TempDir, honcho: bool) -> Config {
let soul = dir.write("etc/soul.md", "You are Boxmaker.\n");
let text = format!(
"[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\n{}",
dir.path().display(),
soul.display(),
if honcho { SECTION } else { "" }
);
let config = Config::parse(&text).unwrap();
let log: serve::Log = Arc::new(|_: &str| {});
let running = serve::start(config.clone(), log).unwrap();
std::thread::spawn(move || running.run());
config
}
#[test]
fn memoryd_answers_with_the_report() {
let dir = TempDir::new("sync-serve");
let config = serve(&dir, true);
assert_eq!(
sync(&config),
"honcho sync: 0 decided: 0 sent, 0 waiting, 0 failed, 0 skipped"
);
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,
});
ask(&config.broker_socket(), call);
// Honcho is not reachable here: the conclusion waits.
assert_eq!(
sync(&config),
"honcho sync: 1 decided: 0 sent, 1 waiting, 0 failed, 0 skipped"
);
}
#[test]
fn without_honcho_it_is_not_configured() {
let dir = TempDir::new("sync-off");
let config = serve(&dir, false);
assert_eq!(sync(&config), NOT_CONFIGURED);
}