Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
27 KiB
M5c-1 task 11: memoryd's Honcho worker sends decided turns
Branch: m5c1 (run git switch m5c1; git status --short must be empty, otherwise stop)
Commit subject: memoryd: the Honcho worker sends decided turns, waits out outages, retries undecided turns
Log row: the Task cell is M5c1/11-memoryd-honcho-worker.
Goal
Task 09 decides each finished turn into the ledger; nothing sends it yet. This task adds the
worker that does (M5c spec, section 5): one thread in memoryd, started with [honcho] only.
- Start (
serve::start,[honcho]only), before anything under the home is made and before any socket is bound: the ledger is read through (as in task 09), the sessions it knows wait again (TurnSync::resume, below), and the three clients are built (task 10) with their keys. A key that cannot be read stopsmemoryd: exit 1,bxcore::secrets' own lines with the pointersecret-unavailable, no socket bound. The key files' plaintext warnings are logged. After the sockets are bound the worker's thread starts. - Sending, oldest first,
Action::Turnitems only: the other actions (conclusions, fleet messages, retractions) wait in the outbox untouched, for task 13. For each turn the worker reads its records fromloopd's log again, takes the speaker's text and the final reply (task 08'suser_textandfinal_reply, made public here), neutralises both (bxcore::sanitize::neutralise), cuts each into parts of at most 25,000 characters, makes sure of its Honcho session (honcho_sessionandensure_session, task 10) and adds every part of both as one batch, the speaker's first, then the agent's. The ids Honcho gives go in aSentrecord. A turn the log no longer holds (a damaged log, a missing turn or reply) is markedFailedwith kindlogand no status; one that needs more than 100 messages, with kindtoo-long. - Honcho unreachable (a connection error, a timeout, a 5xx): wait, then try the same turn again: 5 s, 10 s, 20 s and so on up to 10 minutes. The outage is logged once when it starts and once when it ends (the first answer of any kind), not at every try.
- Refused (another 4xx, or a body that cannot be read): the turn is marked
Failedwith the error's kind and status, logged once, and the next turn goes on. - Key refused (401, 403): nothing more is sent to that workspace until
memorydrestarts, logged once; its turns stay waiting, and the other workspaces go on. - Nothing to send: the worker walks task 09's waiting sessions (
TurnSync::retry: the "worker's retry" of spec 5, step 3), and so it does right after an outage ends. Then it sleeps 30 s, or until the nextSessionUpdatedwakes it. - A ledger that cannot be written stops the worker until
memorydrestarts (logged once): a send whoseSentline could not be written would otherwise be sent again at every step.
What a restart knows. The ledger says what was sent: nothing is sent twice (unless a crash
falls between a send and its Sent line). The sessions task 09 was waiting on are rebuilt from
the ledger: each session with a decided turn waits again at the turns its log holds, and the
next walk decides what is left (never a turn still running: task 09's decidable). A session
with no decided turn is not scanned: it is picked up at its next SessionUpdated. So a session
first reported while brokerd was down, and never again before a restart, waits until its next
turn. (Scanning every session would decide old history; task 15 adds the cutoff for turns from
before Honcho was turned on.)
memoryd has no shutdown: it runs until its process ends, and the worker with it. Its waits are
plain sleeps; only the idle sleep is woken early, by a SessionUpdated.
Interfaces
In crates/memoryd/src/honcho/worker.rs (the module memoryd::honcho::worker):
pub const MAX_PART: usize = 25_000;
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";
pub struct Tuning { pub backoff: Duration, pub backoff_max: Duration, pub idle: Duration }
// Tuning::DEFAULT = 5 s, 600 s, 30 s
pub struct Wake { /* private */ } // Default; notify(&self); wait(&self, timeout)
pub enum Step { Done, Backoff(Duration), Idle, Stopped }
pub type AskBox = Box<dyn Fn(&SessionId) -> Result<SessionStateReply, String> + Send>;
pub struct Worker { /* private */ }
impl Worker {
pub fn new(sync: Arc<Mutex<TurnSync>>, clients: Clients, wake: Arc<Wake>, ask: AskBox,
log: Log, tuning: Tuning, agent: &str) -> Worker;
pub fn run(self);
pub fn step(&mut self) -> Step;
}
pub fn parts(text: &str) -> Vec<String>;
pub fn read_turn(path: &Path, turn: u64) -> Option<(String, String)>;
pub fn batch(text: &str, reply: &str, speaker: &str, agent: &str, session: &SessionId,
turn: u64) -> Vec<MessageCreate>;
pub struct Opened { pub sync: Arc<Mutex<TurnSync>>, /* private */ }
pub fn open(config: &Config, honcho: &HonchoConfig, log: &Log) -> Result<Opened, StartError>;
pub fn spawn(opened: Opened, config: &Config, wake: Arc<Wake>, log: Log) -> Result<(), String>;
run is step in a loop: Backoff(d) sleeps d, Idle sleeps on the Wake for
tuning.idle, Stopped returns. The tests call step directly, with short waits, and see each
Backoff it returns.
In crates/memoryd/src/honcho/turns.rs, TurnSync gains sessions(&self) -> &Path,
resume(&mut self), append(&mut self, Record) -> Result<(), LedgerError> and
retry(&mut self, ask, now, log) (the walk updated did, now also callable alone). In
serve.rs, StartError gains Secret(String), and Memoryd holds the TurnSync in an Arc
shared with the worker, and a Wake it notifies after each SessionUpdated's decisions.
The batch
Fixture 08-messages-add-batch.json is the shape: POST /v3/workspaces/<workspace>/sessions/<honcho session>/messages with
{"messages": [
{"content": "<part>", "peer_id": "<speaker>", "metadata": {"agent": "<agent>", "boxmaker_session": "<id>", "turn": 1}},
{"content": "<part>", "peer_id": "<agent>", "metadata": {"agent": "<agent>", "boxmaker_session": "<id>", "turn": 1}}
]}
answered 201 with the messages made, whose ids go in the Sent record in order. Fixtures 09
and 10: a message of 25,000 characters is taken, one of 25,001 refused (422). Characters, not
bytes: parts cuts by char. A batch holds 1 to 100 messages, and a message's content may be
empty: Honcho's src/schemas/api.py (main) has MessageBatchCreate.messages: Field(..., min_length=1, max_length=100) and MessageCreate.content with min_length=0 (the same cap of
100 holds for a batch of conclusions). So MAX_BATCH is 100, and a turn that needs more (over
2,500,000 characters) is marked failed without a request.
The lines the worker logs
| When | The line |
|---|---|
| an outage starts | memoryd: <error>; sends wait and are retried; see docs/runbook.md#honcho-unreachable |
| it ends | memoryd: Honcho answers again; waiting sends go on |
| a send is refused | memoryd: turn <n> of session <id> was not sent: <error>; it is marked failed; see docs/runbook.md#honcho-send-refused |
| a turn cannot be read | memoryd: turn <n> of session <id> cannot be sent (log); it is marked failed; see docs/runbook.md#session-log-damaged |
| a turn is too long | the same with (too-long) and see docs/runbook.md#honcho-send-refused |
| a key is refused | memoryd: Honcho refused the key of workspace <name> (HTTP <status>); nothing more is sent to it until memoryd restarts; see docs/runbook.md#honcho-key-refused |
| the ledger cannot be written | memoryd: <ledger error>; the Honcho worker stops until memoryd restarts; see docs/runbook.md#honcho-ledger-damaged |
<error> is a HonchoError's Display (task 10): its kind and status, never a body. No line
holds a turn's text or a key.
Every place that changes
This is the complete list.
| Place | What happens there |
|---|---|
crates/memoryd/src/honcho/decide.rs |
user_text and final_reply made public (edits D1, D2) |
crates/memoryd/src/honcho/mod.rs |
pub mod worker; (edit H1) |
crates/memoryd/src/honcho/turns.rs |
sessions, resume, append, retry (edits T1 to T3) |
crates/memoryd/src/honcho/worker.rs |
new: a skeleton you copy, then fill 11 bodies |
crates/memoryd/src/serve.rs |
imports, Memoryd, StartError::Secret, start, decide_turns (edits S1 to S8) |
crates/memoryd/src/main.rs |
a key's error printed with its own pointer (edit M1) |
crates/memoryd/tests/honcho_serve.rs |
a new version, given (task 09's test with a skipped turn in its cut-off ledger, since a worker now runs beside it) |
docs/runbook.md |
three entries appended |
| everything else | no change |
Files
- Copy, from
docs/plans/M5c1/files/(the same path under it):- new tests:
crates/memoryd/tests/honcho_worker.rs, andcrates/memoryd/tests/honcho_worker_more.rsfromhoncho_worker_more.rs-task11(this task's version; task 13 gives a later one) - new test support:
crates/memoryd/tests/support/worker.rs - a new version of
crates/memoryd/tests/honcho_serve.rs(it replaces the file in the tree;honcho_serve.rs-task09beside it is the older one: do not copy that) - the skeleton
crates/memoryd/src/honcho/worker.rs
- new tests:
- Append, from
docs/plans/M5c1/files/, to the end ofdocs/runbook.md, each with one blank line before it, in this order:runbook-honcho-unreachable.md,runbook-honcho-send-refused.md,runbook-honcho-key-refused.md(step 7 gives the commands). - Modify: the files of edits D1 to M1,
docs/implementer-log.md. - Do not touch: any other source file, any other test or support file, the fixtures, any
Cargo.toml, and in the skeleton anything but the 11todo!()bodies.
Facts you need
- What the skeleton gives whole: the types,
Worker::new,run,step(the table of what each outcome does),send(read, batch, check the size, make sure of the session, add),openandspawn. Readstepandsendbefore you fill the helpers they call. - The shared state.
TurnSync(task 09) holds the ledger and the waiting sessions. The socket threads take its lock to decide; the worker takes it to find the next turn, to append its records and to walk, and never holds it while it talks to Honcho or reads a log. Condvar.Wakeis aMutex<bool>and aCondvar:notifysets the flag andnotify_ones;waitwaits withwait_timeout_while(guard, timeout, |woken| !*woken), which returns at once if the flag is already set (aSessionUpdatedthat came while the worker was busy is not lost) and then clears it.Duration::saturating_mul(2)doubles a wait without overflow;.min(max)caps it.- The given tests.
honcho_worker.rs(12 tests): the limits andparts; a turn sent once as one batch, exactly, then idle; parts over 25,000 characters; both texts neutralised (a tag fromTAGS); a new speaker's turn tomm-a-2; an outage (backoffs 1, 2, 4, 8, 8, 8 ms with the tests'FASTwaits), its two lines, the sends in order after it, a new outage from the first wait again; a 422 marked failed and the next sent; a 401 stopping one workspace only; a 403 ending an outage; a turn gone from the log, or without its reply, marked failed; a turn too long for a batch.honcho_worker_more.rs(10 tests): the default waits; undecided turns decided when idle and after an outage; a restart sending nothing again and resuming only the sessions the ledger knows; a running turn never decided; a conclusion left waiting; a ledger that cannot be written stopping the worker; no log line with a turn's text;memorydsending a reported turn at once (woken, not after 30 s); a key that cannot be read stopping the start and the process.support/worker.rs: a home, aTurnSync,brokerdas a closure, collected log lines, and a worker against the fake Honcho with short waits.
- The calls the bodies use:
Call Signature, and what it gives Mutex::lockm.lock().unwrap_or_else(|p| p.into_inner()): the guard, even after a panic elsewhereCondvar::notify_one,wait_timeout_whilecv.wait_timeout_while(guard, timeout, |v| cond) -> LockResult<(MutexGuard, WaitTimeoutResult)>str::chars,slice::chunkstext.chars().collect::<Vec<char>>();chars.chunks(n): slices of at mostnneutralisebxcore::sanitize::neutralise(text: &str) -> Cow<'_, str>MessageCreateroncho::MessageCreate::new(content, peer_id).with_metadata(Map<String, Value>)Ledger::outbox,decisionstasks 07 and 10: Vec<(Item, Action)>in decision order; an iterator of(&Item, &Action)honcho_sessiontask 10: honcho_session(ledger.decisions(), session, turn) -> Option<HonchoSession>read_log,turn_records,user_text,final_replytasks 09 and 08 TurnSync::retry,appendthis task's edits T2 and T3
The edits, exactly
Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file.
crates/memoryd/src/honcho/decide.rs
D1. In decide.rs, the line
fn user_text(turn: &[LogRecord]) -> (&str, Option<&str>) {
becomes
pub fn user_text(turn: &[LogRecord]) -> (&str, Option<&str>) {
D2. The line
fn final_reply(turn: &[LogRecord]) -> Option<&str> {
becomes
pub fn final_reply(turn: &[LogRecord]) -> Option<&str> {
crates/memoryd/src/honcho/mod.rs
H1. The line
pub mod turns;
becomes
pub mod turns;
pub mod worker;
crates/memoryd/src/honcho/turns.rs
T1. In impl TurnSync, the line
/// The sessions that still have undecided turns, with the highest turn reported for each.
becomes
/// The directory of `loopd`'s session logs.
pub fn sessions(&self) -> &Path {
&self.sessions
}
/// The sessions that still have undecided turns, with the highest turn reported for each.
T2. The lines
/// `loopd` reported that turn `turn` of `session` ended. Remember it, then walk every session
/// with undecided turns, in the order of their ids, until one stops the walk.
become
/// After a restart: each session with a decided turn in the ledger waits again, at the turns
/// its log holds, so that a turn left undecided before the restart is decided at the next
/// walk (`decidable` holds back one still running). A session with no decided turn is left
/// to its next report: sessions from before Honcho was turned on are never walked.
pub fn resume(&mut self) {
let mut sessions: Vec<SessionId> = Vec::new();
for (item, _) in self.ledger.decisions() {
if let Item::Turn { session, .. } = item
&& !sessions.contains(session)
{
sessions.push(session.clone());
}
}
for session in sessions {
if let Ok(records) = read_log(&log_path(&self.sessions, &session)) {
self.reported.insert(
session.as_str().to_string(),
(session.clone(), turns(&records)),
);
}
}
}
/// Append `record` to the ledger (the worker's `Sent` and `Failed`).
pub fn append(&mut self, record: Record) -> Result<(), ledger::LedgerError> {
self.ledger.append(record)
}
/// `loopd` reported that turn `turn` of `session` ended. Remember it, then walk.
T3. In updated, the lines
entry.1 = entry.1.max(turn);
for (session, turn) in self.waiting() {
become
entry.1 = entry.1.max(turn);
self.retry(ask, now, log);
}
/// Walk every session with undecided turns, in the order of their ids, until one stops the
/// walk: at a report, and from the worker when it has nothing to send (M5c spec, section 5).
pub fn retry(&mut self, ask: Ask<'_>, now: Timestamp, log: &dyn Fn(&str)) {
for (session, turn) in self.waiting() {
crates/memoryd/src/serve.rs
S1. The lines
use crate::honcho::ledger::{self, Ledger, LedgerError};
use crate::honcho::state;
use crate::honcho::turns::{self, TurnSync};
become
use crate::honcho::ledger::{self, LedgerError};
use crate::honcho::state;
use crate::honcho::turns::TurnSync;
use crate::honcho::worker::{self, Wake};
S2. In pub struct Memoryd, the lines
honcho: Option<Mutex<TurnSync>>,
log: Log,
}
become
honcho: Option<Arc<Mutex<TurnSync>>>,
/// Wakes the Honcho worker when a turn has been decided.
wake: Arc<Wake>,
log: Log,
}
S3. In pub enum StartError, the lines
Failed(String),
Ledger(LedgerError),
}
become
Failed(String),
Ledger(LedgerError),
/// A Honcho key that cannot be read: `bxcore::secrets`' line, with its own pointer.
Secret(String),
}
S4. In impl std::fmt::Display for StartError, the line
StartError::Failed(why) => write!(f, "{why}"),
becomes
StartError::Failed(why) | StartError::Secret(why) => write!(f, "{why}"),
S5. The lines
/// With `[honcho]`, read the ledger through first. Then create what is missing under `memory/`,
/// open the index and catch it up, and bind the three sockets. The error names what failed.
pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
let honcho = match config.honcho_config()? {
Some(honcho) => {
let path = config.memory_dir().join(turns::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()
)
.as_str(),
);
}
Some(Mutex::new(TurnSync::new(
ledger,
&honcho.agent,
&config.sessions_dir(),
)))
}
None => None,
};
become
/// With `[honcho]`, read the ledger through and the keys first. Then create what is missing
/// under `memory/`, open the index and catch it up, bind the three sockets, and start the Honcho
/// worker. The error names what failed.
pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
let opened = match config.honcho_config()? {
Some(honcho) => Some(worker::open(&config, &honcho, &log)?),
None => None,
};
S6. In start, the lines
(admin_listener, Which::Admin),
];
become
(admin_listener, Which::Admin),
];
let wake = Arc::new(Wake::default());
let honcho = match opened {
Some(opened) => {
let sync = Arc::clone(&opened.sync);
worker::spawn(opened, &config, Arc::clone(&wake), Arc::clone(&log))?;
Some(sync)
}
None => None,
};
S7. In start, where Memoryd is built, the lines
honcho,
log,
become
honcho,
wake,
log,
S8. In decide_turns, the lines
Timestamp::now(),
&*self.log,
);
}
become
Timestamp::now(),
&*self.log,
);
drop(sync);
self.wake.notify();
}
crates/memoryd/src/main.rs
M1. The lines
// The ledger's line names its own pointer, `honcho-ledger-damaged`.
Err(e @ serve::StartError::Ledger(_)) => {
become
// The ledger's and a key's lines name their own pointers.
Err(e @ (serve::StartError::Ledger(_) | serve::StartError::Secret(_))) => {
Steps
-
1. Copy the given tests. From the repository root, four commands:
cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker.rs crates/memoryd/tests/ cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_worker_more.rs-task11 crates/memoryd/tests/honcho_worker_more.rs cp docs/plans/M5c1/files/crates/memoryd/tests/honcho_serve.rs crates/memoryd/tests/ cp docs/plans/M5c1/files/crates/memoryd/tests/support/worker.rs crates/memoryd/tests/support/ -
2. See them fail.
cargo build --tests --keep-going -p memoryd, with a shell timeout of 600000 ms: two test targets do not compile.honcho_worker, 3 errors:E0432(unresolved import `memoryd::honcho::worker`) in the test file and insupport/worker.rs, andE0599(no method named `resume` found for struct `TurnSync`).honcho_worker_more, 4 errors: the same three, andE0599(no variant, associated function, or constant named `Secret` found for enum `StartError`). Every other target compiles, andcargo test -p memoryd --test honcho_servepasses its 6 tests already. -
3. The module. Make edits D1, D2, H1, T1 to T3, S1 to S8 and M1, then copy the skeleton:
cp docs/plans/M5c1/files/crates/memoryd/src/honcho/worker.rs crates/memoryd/src/honcho/cargo check -p memoryd: no error, and 21 warnings, all from the unfilled bodies: sixunused importlines (bxcore::sanitize::neutralise;Mapandjson;final_reply,turn_recordsanduser_text;Actionandself;honcho_session;read_log), thirteenunused variable(textandturntwice each, andagent,error,messages,path,record,reply,session,speaker,timeoutonce each),fields `woken` and `ready` are never readandfields `ask` and `outage` are never read. -
4. Fill the eleven bodies of
worker.rs, one at a time, in the order of the file, each from its comment, and runcargo check -p memorydafter each:notify,wait,parts,read_turn,batch,ids,next,unreachable,recovered,retry,record. When a body is filled, delete its instruction comment (the//lines inside the body) and keep the function's doc comment (///) as it is. Change nothing else. After the last,cargo check -p memorydprints no warning. -
5. The tests. With a shell timeout of 600000 ms, and every test passes (12, 10, 6, and the existing ones unchanged):
cargo test -p memoryd --test honcho_worker --test honcho_worker_more --test honcho_serve \ --test honcho_turns --test honcho_client --test honcho_start --test serve -
6. Check. Three commands:
grep -n 'thread::spawn\|Builder::new()' crates/memoryd/src/honcho/worker.rsprints one line, inspawn: the worker is the one new thread.grep -rn 'runbook.md#honcho-unreachable\|runbook.md#honcho-send-refused\|runbook.md#honcho-key-refused' crates/memoryd/srcprints three lines, the constants inworker.rs.git diff --stat -- crates/bxcore crates/loopd crates/brokerd crates/gatewayd crates/bxctl Cargo.toml Cargo.lock crates/memoryd/Cargo.tomlprints nothing.
-
7. The runbook. From the repository root, three commands:
{ echo; cat docs/plans/M5c1/files/runbook-honcho-unreachable.md; } >> docs/runbook.md { echo; cat docs/plans/M5c1/files/runbook-honcho-send-refused.md; } >> docs/runbook.md { echo; cat docs/plans/M5c1/files/runbook-honcho-key-refused.md; } >> docs/runbook.mdThen
grep -c '^## honcho-' docs/runbook.mdprints5. -
8. Run the gate.
cargo fmt --all, thenmake gate, with a shell timeout of 600000 ms. Expected last line:gate: ok, with 1300 tests: 1278 before this task, plus 22 new ones (12 inhoncho_worker.rs, 10 inhoncho_worker_more.rs). -
9. Log and commit. Add your row to
docs/implementer-log.md; its Task cell isM5c1/11-memoryd-honcho-worker.git status --shortlistsdecide.rs,honcho/mod.rs,turns.rs,serve.rs,main.rs,tests/honcho_serve.rs,docs/runbook.mdand your log as modified, andworker.rs, the two new test files andtests/support/worker.rsas new. Stage withgit add crates/memoryd docs/runbook.md docs/implementer-log.md Cargo.lock, then commit asAGENTS.mdsays.Cargo.lockdoes not change in this task; thegit addis harmless.
Done when
make gateprintsgate: okwith 1300 tests.git diff HEAD~1 --statlists only the files of step 9.
Stop and report if
- A "before" block of D1 to M1 is not in its file exactly as shown, or is there more than once.
- Step 2 or 3 shows a different result: other test targets, errors of another kind, or other warnings.
cargo checkreports an error after a body you filled or an edit you made, and the comment or the edit does not tell you how to fix it. Quote the error, and stop.- A comment in the skeleton does not fit the code under it, or disagrees with this task's tables. Quote both and stop.
- After step 4 a test fails although each body follows its comment and each edit is as shown. Do not change a test, a support file or the skeleton's given code: stop, and give the failing tests' names and their messages.
- A test seems to need a real Honcho, a new fixture or a longer wait. It does not; stop and say why you think so.
- The fix seems to need a change anywhere else:
client.rs,ledger.rs, another crate, a second thread. It does not; stop and say why you think so. - A test outside the new files fails in the gate: check once whether it fails without your change, then stop and report both results.
- The gate passes with a count other than 1300.
What the given tests pin
Each of these changes to the finished code makes at least one given test fail (36 tried; one
cannot change behaviour): a part of 25,001, or parts cut by bytes; no neutralising; the agent
first, or the reply from the speaker; the wrong turn in the metadata; no batch limit; no session
made first; the messages sent to the Boxmaker id instead of the Honcho session; the ids dropped;
the wait not doubled or not capped; the outage logged at every try, its end not logged, no
retry after it, or a refused key not ending it; a refused key not remembered, its workspace not
skipped, or its turns failed; a refused send not recorded; a local failure not recorded; the
newest turn first; no retry when idle, or idle after a retry that decided something; a ledger
that cannot be written ignored, or stopped ignored; the wake doing nothing; a turn without its
reply sent anyway; every session in sessions/ resumed, or none; the retry treated as a report;
no worker started, or not woken; a key's error taken for a start failure, or printed with the
start pointer.
After this task
- Task 13 sends the other actions (conclusions, their deletion, fleet messages and retractions) from the same worker.
- Task 15 adds the cutoff: turns from before Honcho was turned on are skipped.