kyle/boxmaker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

memoryd: the Honcho worker sends conclusions, their deletions and fleet notes

Browse Source 
Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
This commit is contained in:
kyle
2026-09-28 13:11:51 -07:00
co-authored by ornith-1.5-35b-a3b
parent 4ebfc36c1a
commit fb05b26462
9 changed files with 790 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/memoryd/src/honcho/client.rs
+32
View File
@@ -127,6 +127,7 @@ pub struct Clients {
untrusted: Client,
fleet: Client,
agent: String,
owner: String,
ensured: HashSet<(Workspace, Ensured)>,
}
@@ -233,6 +234,7 @@ impl Clients {
)?,
fleet: client(&config.url, &config.fleet, fleet_key.as_ref(), timeouts)?,
agent: config.agent.clone(),
owner: config.owner.clone(),
ensured: HashSet::new(),
};
Ok((clients, warnings))
@@ -307,6 +309,36 @@ impl Clients {
Ok(())
}
/// Make sure memory session `name` (a Boxmaker session's `<id>_memory`) exists in
/// `workspace`, after the workspace, the agent's peer and the owner's. It holds conclusions,
/// no messages, so neither peer observes (M5c spec, section 4).
pub fn ensure_memory_session(
&mut self,
workspace: Workspace,
name: &str,
) -> Result<(), HonchoError> {
self.ensure_workspace(workspace)?;
let (agent, owner) = (self.agent.clone(), self.owner.clone());
self.ensure_peer(workspace, &agent)?;
self.ensure_peer(workspace, &owner)?;
let key = (workspace, Ensured::Session(name.to_string()));
if self.ensured.contains(&key) {
return Ok(());
}
let unobserved = || {
SessionPeerConfig::new()
.observe_me(false)
.observe_others(false)
};
let mut peers = HashMap::new();
peers.insert(agent, unobserved());
peers.insert(owner, unobserved());
let create = SessionCreate::new(name).with_peers(peers);
created(self.client(workspace).session(&create))?;
self.ensured.insert(key);
Ok(())
}
/// Delete conclusion `id` in `workspace`. Gone already (404) counts as done.
pub fn delete_conclusion(&self, workspace: Workspace, id: &str) -> Result<(), HonchoError> {
match self.client(workspace).delete_conclusion(id) {
crates/memoryd/src/honcho/edits.rs
+92
View File
@@ -0,0 +1,92 @@
//! 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())
}
/// 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);
clients.ensure_memory_session(workspace, &name)?;
let conclusion = ConclusionCreate::new(neutralise(text), peer, peer).with_session_id(name);
let batch = ConclusionBatchCreate::new(vec![conclusion]);
match clients.client(workspace).create_conclusions(&batch) {
Ok(made) => Ok(made.into_iter().map(|c| c.id).collect()),
Err(e) => Err(classify(&e)),
}
}
/// 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)?;
Ok(vec![id.to_string()])
}
None => Ok(Vec::new()),
}
}
/// 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> {
let mut metadata = Map::new();
metadata.insert("agent".to_string(), json!(agent));
metadata.insert("boxmaker_session".to_string(), json!(session.as_str()));
metadata.insert("edit".to_string(), json!(edit));
parts(&neutralise(text))
.into_iter()
.map(|part| MessageCreate::new(part, agent).with_metadata(metadata.clone()))
.collect()
}
/// 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(),
};
clients.ensure_session(Workspace::Fleet, &honcho)?;
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)),
}
}
crates/memoryd/src/honcho/mod.rs
+2
View File
@@ -3,9 +3,11 @@
pub mod client;
pub mod config;
pub mod decide;
pub mod edits;
pub mod ledger;
pub mod mirror;
pub mod naming;
pub mod start;
pub mod state;
pub mod turns;
pub mod worker;
crates/memoryd/src/honcho/start.rs
+85
View File
@@ -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}"))
}
crates/memoryd/src/honcho/worker.rs
+136 -111
View File
@@ -15,16 +15,15 @@ use bxcore::{SessionId, SessionStateReply, Timestamp};
use roncho::MessageCreate;
use serde_json::{Map, json};
use crate::config::Config;
use crate::honcho::client::{Clients, HonchoError, Timeouts, classify};
use crate::honcho::config::HonchoConfig;
use crate::honcho::client::{Clients, HonchoError, classify};
use crate::honcho::decide::{final_reply, turn_records, user_text};
use crate::honcho::edits;
use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace};
use crate::honcho::mirror::Mirror;
use crate::honcho::naming::{HonchoSession, honcho_session};
use crate::honcho::state;
use crate::honcho::turns::{LEDGER, TurnSync, log_path, read_log};
use crate::serve::{Log, StartError};
use crate::honcho::turns::{TurnSync, log_path, read_log};
use crate::serve::Log;
pub use crate::honcho::start::{Opened, open, spawn};
/// The most characters Honcho takes in one message (recorded: 25,000 is taken, 25,001 refused).
pub const MAX_PART: usize = 25_000;
@@ -103,14 +102,36 @@ enum Unsent {
Local(&'static str),
}
/// A turn to send: its item, workspace, speaker and Honcho session.
/// What to send for one waiting item (M5c spec, section 5).
enum Job {
/// A turn: its session and number, its speaker, and its Honcho session.
Turn {
session: SessionId,
turn: u64,
speaker: String,
honcho: Option<HonchoSession>,
},
/// A conclusion about `peer`, stamped with `session`'s memory session.
Conclude {
peer: String,
session: SessionId,
text: String,
},
/// Delete the conclusion with this id; `None` when no sent conclusion is live.
Unconclude { id: Option<String> },
/// An untrusted note: a fleet message from the agent.
Fleet {
session: SessionId,
edit: String,
text: String,
},
}
/// The oldest waiting item: what it is, where it goes, and what to send.
struct Next {
item: Item,
session: SessionId,
turn: u64,
workspace: Workspace,
speaker: String,
honcho: Option<HonchoSession>,
job: Job,
}
pub struct Worker {
@@ -173,6 +194,14 @@ pub fn batch(
messages
}
/// An item as the log lines name it: `turn 3 of session mm-a`, `part 0 of edit 2026-09-28-0001`.
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}"),
}
}
/// The ids of the messages Honcho made, in order.
fn ids(messages: &[roncho::Message]) -> Vec<String> {
messages.iter().map(|m| m.id.clone()).collect()
@@ -255,9 +284,8 @@ impl Worker {
Err(Unsent::Honcho(e)) => {
self.recovered();
(self.log)(&format!(
"memoryd: turn {} of session {} was not sent: {e}; it is marked failed; {SEND_REFUSED}",
next.turn,
next.session.as_str()
"memoryd: {} was not sent: {e}; it is marked failed; {SEND_REFUSED}",
label(&next.item)
));
self.record(Record::Failed {
time: Timestamp::now(),
@@ -273,9 +301,8 @@ impl Worker {
SEND_REFUSED
};
(self.log)(&format!(
"memoryd: turn {} of session {} cannot be sent ({kind}); it is marked failed; {pointer}",
next.turn,
next.session.as_str()
"memoryd: {} cannot be sent ({kind}); it is marked failed; {pointer}",
label(&next.item)
));
self.record(Record::Failed {
time: Timestamp::now(),
@@ -287,50 +314,117 @@ impl Worker {
}
}
/// The oldest waiting turn whose workspace's key was not refused. Other actions wait.
/// The oldest waiting item whose workspace's key was not refused. A retraction, which
/// nothing decides any more, waits.
fn next(&self) -> Option<Next> {
let ledger = self.ledger.lock().unwrap_or_else(|p| p.into_inner());
ledger.outbox().into_iter().find_map(|(item, action)| {
let Item::Turn { session, turn } = &item else {
return None;
};
let Action::Turn { workspace, speaker } = action else {
return None;
let (workspace, job) = match (&item, action) {
(Item::Turn { session, turn }, Action::Turn { workspace, speaker }) => (
workspace,
Job::Turn {
honcho: honcho_session(ledger.decisions(), session, *turn),
session: session.clone(),
turn: *turn,
speaker,
},
),
(
Item::Edit { .. },
Action::Conclude {
workspace,
peer,
session,
text,
},
) => (
workspace,
Job::Conclude {
peer,
session,
text,
},
),
(Item::Edit { .. }, Action::Unconclude { workspace, text }) => (
workspace,
Job::Unconclude {
id: ledger.conclusion(workspace, &text).map(|(_, id)| id),
},
),
(Item::Edit { id, .. }, Action::Fleet { session, text }) => (
Workspace::Fleet,
Job::Fleet {
session,
edit: id.clone(),
text,
},
),
_ => return None,
};
if self.refused.contains(&workspace) {
return None;
}
Some(Next {
honcho: honcho_session(ledger.decisions(), session, *turn),
session: session.clone(),
turn: *turn,
item: item.clone(),
workspace,
speaker,
job,
})
})
}
/// Read the turn again, make sure of its Honcho session, and add its messages as one batch.
/// Send one item, by what it is.
fn send(&mut self, next: &Next) -> Result<Vec<String>, Unsent> {
let (text, reply) = read_turn(&log_path(&self.sessions, &next.session), next.turn)
.ok_or(Unsent::Local("log"))?;
let honcho = next.honcho.as_ref().ok_or(Unsent::Local("log"))?;
let messages = batch(
&text,
&reply,
&next.speaker,
&self.agent,
&next.session,
next.turn,
);
let honcho = |e| Unsent::Honcho(e);
match &next.job {
Job::Turn {
session,
turn,
speaker,
honcho: name,
} => self.send_turn(next.workspace, session, *turn, speaker, name.as_ref()),
Job::Conclude {
peer,
session,
text,
} => edits::conclude(&mut self.clients, next.workspace, peer, session, text)
.map_err(honcho),
Job::Unconclude { id } => {
edits::unconclude(&self.clients, next.workspace, id.as_deref()).map_err(honcho)
}
Job::Fleet {
session,
edit,
text,
} => {
let messages = edits::fleet_batch(text, &self.agent, session, edit);
if messages.len() > MAX_BATCH {
return Err(Unsent::Local("too-long"));
}
edits::fleet(&mut self.clients, session, &messages).map_err(honcho)
}
}
}
/// Read the turn again, make sure of its Honcho session, and add its messages as one batch.
fn send_turn(
&mut self,
workspace: Workspace,
session: &SessionId,
turn: u64,
speaker: &str,
honcho: Option<&HonchoSession>,
) -> Result<Vec<String>, Unsent> {
let (text, reply) =
read_turn(&log_path(&self.sessions, session), turn).ok_or(Unsent::Local("log"))?;
let honcho = honcho.ok_or(Unsent::Local("log"))?;
let messages = batch(&text, &reply, speaker, &self.agent, session, turn);
if messages.len() > MAX_BATCH {
return Err(Unsent::Local("too-long"));
}
self.clients
.ensure_session(next.workspace, honcho)
.ensure_session(workspace, honcho)
.map_err(Unsent::Honcho)?;
let client = self.clients.client(next.workspace);
let client = self.clients.client(workspace);
match client.add_messages(&honcho.name, &messages) {
Ok(made) => Ok(ids(&made)),
Err(e) => Err(Unsent::Honcho(classify(&e))),
@@ -390,72 +484,3 @@ impl Worker {
}
}
}
/// 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}"))
}