The worker marked a turn failed on any error reading its log. A missing log, a damaged line or a turn not found still fail it; any other I/O error is now waited out with the outage's doubling wait, logged once when it starts and once when the log reads again (pointer session-log-damaged, whose runbook entry now covers it), shown by bxctl memory sync, and the turn stays waiting. read_records tells an unreadable file from a damaged line (read_log is unchanged). The message helpers move from worker.rs to honcho/messages.rs, re-exported, to keep worker.rs under 500 lines. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
191 lines
6.5 KiB
Rust
191 lines
6.5 KiB
Rust
//! What `bxctl memory sync` shows beyond the ledger (M5c spec, sections 10, 12 and 13; review of
|
|
//! M5c-1, finding 4): a workspace whose key was refused, an outage of Honcho and a silent
|
|
//! `brokerd`, each while it lasts, in exact lines; nothing more when all is well. The worker and
|
|
//! the walk keep the status; `memoryd` answers it.
|
|
|
|
#[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::os::unix::net::UnixStream;
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use bxcore::{
|
|
CallId, DataClass, Envelope, MemoryAdmin, MemoryCall, MemoryCommand, Message, SessionId,
|
|
Timestamp, ToolRequest, read_frame, write_frame,
|
|
};
|
|
use honcho::{FakeHoncho, Mode};
|
|
use logs::assistant;
|
|
use memoryd::config::Config;
|
|
use memoryd::honcho::ledger::{Ledger, Record, Workspace};
|
|
use memoryd::honcho::report::{report, with_status};
|
|
use memoryd::honcho::status::Status;
|
|
use memoryd::serve;
|
|
use tmp::TempDir;
|
|
use turns::spoken;
|
|
use worker::{FAST, Home};
|
|
|
|
fn at(minute: u32) -> Timestamp {
|
|
Timestamp::parse(&format!("2026-09-28T10:{minute:02}:00.000Z")).unwrap()
|
|
}
|
|
|
|
const ALL: &str = "honcho sync: 0 decided: 0 sent, 0 waiting, 0 failed, 0 skipped
|
|
brokerd unanswered since 2026-09-28T10:01:00.000Z: turns wait undecided
|
|
honcho unreachable since 2026-09-28T10:02:00.000Z: sends wait and are retried
|
|
session log unreadable since 2026-09-28T10:03:00.000Z: sends wait and are retried
|
|
key refused (HTTP 401): trusted workspace boxmaker0-trusted: its sends wait until memoryd restarts
|
|
key refused (HTTP 403): fleet workspace fleet: its sends wait until memoryd restarts";
|
|
|
|
#[test]
|
|
fn the_status_lines_are_exactly_these() {
|
|
let dir = TempDir::new("status-report");
|
|
let mut ledger = Ledger::open(&dir.path().join("honcho.jsonl")).unwrap();
|
|
ledger.append(Record::Began { time: at(0) }).unwrap();
|
|
assert_eq!(with_status(&ledger, &Status::default()), report(&ledger));
|
|
let mut status = Status {
|
|
broker_down_since: Some(at(1)),
|
|
outage_since: Some(at(2)),
|
|
log_unreadable_since: Some(at(3)),
|
|
..Status::default()
|
|
};
|
|
status
|
|
.refused
|
|
.insert(Workspace::Fleet, ("fleet".to_string(), 403));
|
|
status
|
|
.refused
|
|
.insert(Workspace::Trusted, ("boxmaker0-trusted".to_string(), 401));
|
|
assert_eq!(with_status(&ledger, &status), ALL);
|
|
}
|
|
|
|
fn turn(text: &str) -> Vec<bxcore::LogRecord> {
|
|
vec![spoken(text, Some("kyle")), assistant(2, Some("ok"), &[])]
|
|
}
|
|
|
|
fn status(home: &Home) -> Status {
|
|
let shared = home.sync.lock().unwrap().shared_status();
|
|
memoryd::honcho::status::read(&shared)
|
|
}
|
|
|
|
#[test]
|
|
fn the_worker_keeps_the_outage_and_the_refused_keys() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("status-worker");
|
|
home.log("mm-a", &[turn("a")]);
|
|
home.report("mm-a", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
fake.set(Mode::Status(503));
|
|
let before = Timestamp::now();
|
|
w.step();
|
|
let since = status(&home).outage_since.expect("an outage");
|
|
assert!(since >= before);
|
|
w.step();
|
|
assert_eq!(status(&home).outage_since, Some(since), "from its start");
|
|
fake.set(Mode::Fixtures);
|
|
w.step();
|
|
assert_eq!(status(&home), Status::default(), "the outage ended");
|
|
|
|
home.log("mm-b", &[turn("b")]);
|
|
home.report("mm-b", 1);
|
|
fake.set(Mode::Status(401));
|
|
w.step();
|
|
let refused = status(&home).refused;
|
|
assert_eq!(
|
|
refused.get(&Workspace::Trusted),
|
|
Some(&("boxmaker0-trusted".to_string(), 401))
|
|
);
|
|
assert_eq!(refused.len(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn the_walk_keeps_a_silent_brokerd() {
|
|
let home = Home::new("status-broker");
|
|
home.log("mm-a", &[turn("a")]);
|
|
*home.answer.lock().unwrap() = Err("brokerd is down".to_string());
|
|
home.report("mm-a", 1);
|
|
assert_eq!(
|
|
status(&home).broker_down_since,
|
|
Some(Timestamp::parse("2026-09-28T10:00:00.000Z").unwrap())
|
|
);
|
|
*home.answer.lock().unwrap() = Ok(turns::clean());
|
|
home.report("mm-a", 1);
|
|
assert_eq!(status(&home), Status::default());
|
|
}
|
|
|
|
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: 1,
|
|
r#final: true,
|
|
msg,
|
|
};
|
|
write_frame(&mut stream, &env).unwrap();
|
|
read_frame(&mut stream).unwrap().msg
|
|
}
|
|
|
|
#[test]
|
|
fn memoryd_shows_a_refused_key() {
|
|
let fake = FakeHoncho::start();
|
|
fake.set(Mode::Status(401));
|
|
let dir = TempDir::new("status-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 log: serve::Log = Arc::new(|_: &str| {});
|
|
let running = serve::start(config.clone(), log).unwrap();
|
|
std::thread::spawn(move || running.run());
|
|
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);
|
|
let sync = || match ask(
|
|
&config.admin_socket(),
|
|
Message::MemoryAdmin(MemoryAdmin {
|
|
command: MemoryCommand::Sync,
|
|
arg: String::new(),
|
|
}),
|
|
) {
|
|
Message::MemoryAdminReply(reply) => reply.text,
|
|
other => panic!("{other:?}"),
|
|
};
|
|
let shown = Mutex::new(String::new());
|
|
let deadline = Instant::now() + Duration::from_secs(10);
|
|
while Instant::now() < deadline {
|
|
*shown.lock().unwrap() = sync();
|
|
if shown.lock().unwrap().contains("key refused") {
|
|
break;
|
|
}
|
|
std::thread::sleep(Duration::from_millis(20));
|
|
}
|
|
assert_eq!(
|
|
*shown.lock().unwrap(),
|
|
"honcho sync: 1 decided: 0 sent, 1 waiting, 0 failed, 0 skipped\n\
|
|
key refused (HTTP 401): trusted workspace boxmaker0-trusted: its sends wait until memoryd restarts"
|
|
);
|
|
}
|