M5c-1 plan: task 11, the Honcho worker that sends turns
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,403 @@
|
||||
//! The worker: one thread that sends what the ledger says is waiting, oldest first (M5c spec,
|
||||
//! section 5). A turn is read again from `loopd`'s log, neutralised, cut into parts Honcho takes,
|
||||
//! and added to its Honcho session as one batch; the ids Honcho gives go in the ledger. An outage
|
||||
//! is waited out with a doubling wait and logged when it starts and when it ends; a refused send is
|
||||
//! marked failed; a refused key stops that workspace until `memoryd` restarts. When nothing waits,
|
||||
//! the worker decides what `brokerd` could not answer for before, then sleeps until a turn ends.
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::path::Path;
|
||||
use std::sync::{Arc, Condvar, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use bxcore::sanitize::neutralise;
|
||||
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::decide::{final_reply, turn_records, user_text};
|
||||
use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace};
|
||||
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};
|
||||
|
||||
/// The most characters Honcho takes in one message (recorded: 25,000 is taken, 25,001 refused).
|
||||
pub const MAX_PART: usize = 25_000;
|
||||
/// The most messages the worker puts in one batch. A turn that needs more is marked failed.
|
||||
pub const MAX_BATCH: usize = 100;
|
||||
|
||||
pub const UNREACHABLE: &str = "see docs/runbook.md#honcho-unreachable";
|
||||
pub const SEND_REFUSED: &str = "see docs/runbook.md#honcho-send-refused";
|
||||
pub const KEY_REFUSED: &str = "see docs/runbook.md#honcho-key-refused";
|
||||
pub const LOG_DAMAGED: &str = "see docs/runbook.md#session-log-damaged";
|
||||
|
||||
/// The worker's waits. Tests shorten them.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct Tuning {
|
||||
/// The first wait in an outage; each next one doubles, up to `backoff_max`.
|
||||
pub backoff: Duration,
|
||||
pub backoff_max: Duration,
|
||||
/// How long the worker sleeps with nothing to send, unless a turn ends first.
|
||||
pub idle: Duration,
|
||||
}
|
||||
|
||||
impl Tuning {
|
||||
pub const DEFAULT: Tuning = Tuning {
|
||||
backoff: Duration::from_secs(5),
|
||||
backoff_max: Duration::from_secs(600),
|
||||
idle: Duration::from_secs(30),
|
||||
};
|
||||
}
|
||||
|
||||
/// Wakes the worker from its idle sleep when a turn ends.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct Wake {
|
||||
woken: Mutex<bool>,
|
||||
ready: Condvar,
|
||||
}
|
||||
|
||||
impl Wake {
|
||||
/// Wake the worker if it sleeps, or make its next sleep return at once.
|
||||
pub fn notify(&self) {
|
||||
// `let mut woken = self.woken.lock().unwrap_or_else(|p| p.into_inner());`, then
|
||||
// `*woken = true;` and `self.ready.notify_one();`.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Sleep until `notify`, or at most `timeout`.
|
||||
pub fn wait(&self, timeout: Duration) {
|
||||
// `let woken = self.woken.lock().unwrap_or_else(|p| p.into_inner());`. Then
|
||||
// `let (mut woken, _) = self.ready.wait_timeout_while(woken, timeout, |woken| !*woken)`
|
||||
// `.unwrap_or_else(|p| p.into_inner());`, and `*woken = false;` (the wake is used up).
|
||||
todo!()
|
||||
}
|
||||
}
|
||||
|
||||
/// What one step of the worker did, and so what the loop does next.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Step {
|
||||
/// A turn was sent, or marked failed: go on at once.
|
||||
Done,
|
||||
/// Honcho is unreachable: wait this long, then try the same turn again.
|
||||
Backoff(Duration),
|
||||
/// Nothing to send, and the retry decided nothing new: sleep.
|
||||
Idle,
|
||||
/// The ledger cannot be written: the worker stops.
|
||||
Stopped,
|
||||
}
|
||||
|
||||
/// `brokerd`'s answer for a session, for the retry.
|
||||
pub type AskBox = Box<dyn Fn(&SessionId) -> Result<SessionStateReply, String> + Send>;
|
||||
|
||||
/// Why a turn could not be sent.
|
||||
enum Unsent {
|
||||
Honcho(HonchoError),
|
||||
/// The turn cannot be read from the log any more, or is too long for one batch.
|
||||
Local(&'static str),
|
||||
}
|
||||
|
||||
/// A turn to send: its item, workspace, speaker and Honcho session.
|
||||
struct Next {
|
||||
item: Item,
|
||||
session: SessionId,
|
||||
turn: u64,
|
||||
workspace: Workspace,
|
||||
speaker: String,
|
||||
honcho: Option<HonchoSession>,
|
||||
}
|
||||
|
||||
pub struct Worker {
|
||||
sync: Arc<Mutex<TurnSync>>,
|
||||
clients: Clients,
|
||||
wake: Arc<Wake>,
|
||||
ask: AskBox,
|
||||
log: Log,
|
||||
tuning: Tuning,
|
||||
agent: String,
|
||||
/// In an outage: the next wait.
|
||||
outage: Option<Duration>,
|
||||
/// Workspaces whose key was refused: nothing more is sent to them until restart.
|
||||
refused: HashSet<Workspace>,
|
||||
stopped: bool,
|
||||
}
|
||||
|
||||
/// `text` in parts of at most `MAX_PART` characters, in order. Empty text has no part.
|
||||
pub fn parts(text: &str) -> Vec<String> {
|
||||
// `let chars: Vec<char> = text.chars().collect();` then
|
||||
// `chars.chunks(MAX_PART).map(|chunk| chunk.iter().collect()).collect()`.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Turn `turn`'s user text and final reply, read again from the log at `path`. `None` when the log
|
||||
/// does not read, or has no such turn, or the turn no reply.
|
||||
pub fn read_turn(path: &Path, turn: u64) -> Option<(String, String)> {
|
||||
// Each step with `?`: `read_log(path).ok()?` into `records`; then
|
||||
// `let records = turn_records(&records, turn)?;`; `let (text, _) = user_text(records);`;
|
||||
// `let reply = final_reply(records)?;`. Then `Some((text.to_string(), reply.to_string()))`.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// A turn's messages, in one batch: the speaker's text, then the agent's reply, each neutralised
|
||||
/// and in parts of at most `MAX_PART` characters, each part with the turn's metadata.
|
||||
pub fn batch(
|
||||
text: &str,
|
||||
reply: &str,
|
||||
speaker: &str,
|
||||
agent: &str,
|
||||
session: &SessionId,
|
||||
turn: u64,
|
||||
) -> Vec<MessageCreate> {
|
||||
// A `Map` `metadata` with `"agent"` -> `json!(agent)`, `"boxmaker_session"` ->
|
||||
// `json!(session.as_str())`, `"turn"` -> `json!(turn)` (in this order). Then for each
|
||||
// `(content, peer)` of `[(text, speaker), (reply, agent)]`, for each `part` of
|
||||
// `parts(&neutralise(content))`, push
|
||||
// `MessageCreate::new(part, peer).with_metadata(metadata.clone())`. Return the messages.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// The ids of the messages Honcho made, in order.
|
||||
fn ids(messages: &[roncho::Message]) -> Vec<String> {
|
||||
// `messages.iter().map(|m| m.id.clone()).collect()`.
|
||||
todo!()
|
||||
}
|
||||
|
||||
impl Worker {
|
||||
/// `agent` is `[honcho] agent`. `sync` and `wake` are shared with the socket threads.
|
||||
pub fn new(
|
||||
sync: Arc<Mutex<TurnSync>>,
|
||||
clients: Clients,
|
||||
wake: Arc<Wake>,
|
||||
ask: AskBox,
|
||||
log: Log,
|
||||
tuning: Tuning,
|
||||
agent: &str,
|
||||
) -> Worker {
|
||||
Worker {
|
||||
sync,
|
||||
clients,
|
||||
wake,
|
||||
ask,
|
||||
log,
|
||||
tuning,
|
||||
agent: agent.to_string(),
|
||||
outage: None,
|
||||
refused: HashSet::new(),
|
||||
stopped: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Step until the ledger cannot be written; wait as each step says.
|
||||
pub fn run(mut self) {
|
||||
loop {
|
||||
match self.step() {
|
||||
Step::Done => {}
|
||||
Step::Backoff(wait) => std::thread::sleep(wait),
|
||||
Step::Idle => self.wake.wait(self.tuning.idle),
|
||||
Step::Stopped => return,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Send the oldest waiting turn, or, with none, retry the undecided ones.
|
||||
pub fn step(&mut self) -> Step {
|
||||
if self.stopped {
|
||||
return Step::Stopped;
|
||||
}
|
||||
let Some(next) = self.next() else {
|
||||
self.retry();
|
||||
return match self.next() {
|
||||
Some(_) => Step::Done,
|
||||
None => Step::Idle,
|
||||
};
|
||||
};
|
||||
match self.send(&next) {
|
||||
Ok(sent) => {
|
||||
self.recovered();
|
||||
self.record(Record::Sent {
|
||||
time: Timestamp::now(),
|
||||
item: next.item,
|
||||
ids: sent,
|
||||
})
|
||||
}
|
||||
Err(Unsent::Honcho(e @ HonchoError::Unreachable { .. })) => self.unreachable(&e),
|
||||
Err(Unsent::Honcho(HonchoError::KeyRefused { status })) => {
|
||||
self.recovered();
|
||||
self.refused.insert(next.workspace);
|
||||
(self.log)(&format!(
|
||||
"memoryd: Honcho refused the key of workspace {} (HTTP {status}); nothing more is sent to it until memoryd restarts; {KEY_REFUSED}",
|
||||
self.clients.client(next.workspace).workspace_id()
|
||||
));
|
||||
Step::Done
|
||||
}
|
||||
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()
|
||||
));
|
||||
self.record(Record::Failed {
|
||||
time: Timestamp::now(),
|
||||
item: next.item,
|
||||
status: e.status(),
|
||||
kind: e.kind().to_string(),
|
||||
})
|
||||
}
|
||||
Err(Unsent::Local(kind)) => {
|
||||
let pointer = if kind == "log" {
|
||||
LOG_DAMAGED
|
||||
} else {
|
||||
SEND_REFUSED
|
||||
};
|
||||
(self.log)(&format!(
|
||||
"memoryd: turn {} of session {} cannot be sent ({kind}); it is marked failed; {pointer}",
|
||||
next.turn,
|
||||
next.session.as_str()
|
||||
));
|
||||
self.record(Record::Failed {
|
||||
time: Timestamp::now(),
|
||||
item: next.item,
|
||||
status: None,
|
||||
kind: kind.to_string(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The oldest waiting turn whose workspace's key was not refused. Other actions wait.
|
||||
fn next(&self) -> Option<Next> {
|
||||
// Lock `self.sync` (`unwrap_or_else(|p| p.into_inner())`), `let ledger =
|
||||
// sync.ledger();`, then `ledger.outbox().into_iter().find_map(|(item, action)| { .. })`:
|
||||
// `None` unless `item` is `Item::Turn { session, turn }` (by `&item`) and `action` is
|
||||
// `Action::Turn { workspace, speaker }` (each by `let .. else`), and `None` if
|
||||
// `self.refused.contains(&workspace)`; else `Some(Next { honcho:
|
||||
// honcho_session(ledger.decisions(), session, *turn), session: session.clone(), turn:
|
||||
// *turn, item: item.clone(), workspace, speaker })`.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Read the turn again, make sure of its Honcho session, and add its messages as one batch.
|
||||
fn send(&mut self, next: &Next) -> Result<Vec<String>, Unsent> {
|
||||
let sessions = self
|
||||
.sync
|
||||
.lock()
|
||||
.unwrap_or_else(|p| p.into_inner())
|
||||
.sessions()
|
||||
.to_path_buf();
|
||||
let (text, reply) = read_turn(&log_path(&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,
|
||||
);
|
||||
if messages.len() > MAX_BATCH {
|
||||
return Err(Unsent::Local("too-long"));
|
||||
}
|
||||
self.clients
|
||||
.ensure_session(next.workspace, honcho)
|
||||
.map_err(Unsent::Honcho)?;
|
||||
let client = self.clients.client(next.workspace);
|
||||
match client.add_messages(&honcho.name, &messages) {
|
||||
Ok(made) => Ok(ids(&made)),
|
||||
Err(e) => Err(Unsent::Honcho(classify(&e))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Honcho did not answer: log the outage's start once, and wait longer each time.
|
||||
fn unreachable(&mut self, error: &HonchoError) -> Step {
|
||||
// `let wait = match self.outage`: `Some(wait)` -> `wait`; `None` -> log
|
||||
// `format!("memoryd: {error}; sends wait and are retried; {UNREACHABLE}")`, then
|
||||
// `self.tuning.backoff`. Then
|
||||
// `self.outage = Some(wait.saturating_mul(2).min(self.tuning.backoff_max));` and
|
||||
// `Step::Backoff(wait)`.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Honcho answered: if that ends an outage, log it once, and decide what waited meanwhile.
|
||||
fn recovered(&mut self) {
|
||||
// If `self.outage.take().is_some()`: log exactly
|
||||
// `memoryd: Honcho answers again; waiting sends go on`, then `self.retry();`.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Task 09's walk, for the turns `brokerd` could not answer for.
|
||||
fn retry(&mut self) {
|
||||
// Lock `self.sync` as in `next` (as `mut sync`), then
|
||||
// `sync.retry(&*self.ask, Timestamp::now(), &*self.log);`.
|
||||
todo!()
|
||||
}
|
||||
|
||||
/// Append the worker's record. A ledger that cannot be written stops the worker: the send
|
||||
/// would be repeated at every step.
|
||||
fn record(&mut self, record: Record) -> Step {
|
||||
// `let result =
|
||||
// self.sync.lock().unwrap_or_else(|p| p.into_inner()).append(record);`. `Ok(())` ->
|
||||
// `Step::Done`. `Err(e)` -> log `format!("memoryd: {e}; the Honcho worker stops until
|
||||
// memoryd restarts; {}", ledger::RUNBOOK)`, then `self.stopped = true;`, `Step::Stopped`.
|
||||
todo!()
|
||||
}
|
||||
}
|
||||
|
||||
/// 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,
|
||||
}
|
||||
|
||||
/// 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(),
|
||||
})
|
||||
}
|
||||
|
||||
/// 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}"))
|
||||
}
|
||||
Reference in New Issue
Block a user