M5c-1 plan: task 09, turns decided on SessionUpdated after loopd has its answer

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

docs/plans/M5c1/files/crates/memoryd/src/honcho/state.rs

@@ -0,0 +1,36 @@
+//! Asking `brokerd` for a session's state on its `state.sock` (M5c spec, section 7): one question
+//! per connection, one final answer, as on every socket between the roles.
+
+use std::os::unix::net::UnixStream;
+use std::path::Path;
+use std::time::Duration;
+
+use bxcore::{
+    Envelope, Message, PROTOCOL_VERSION, SessionId, SessionState, SessionStateReply, read_frame,
+    write_frame,
+};
+
+/// How long `brokerd` may leave the connection without a byte, sending or answering.
+pub const WAIT: Duration = Duration::from_secs(5);
+
+/// The runbook entry for a `brokerd` that does not answer.
+pub const RUNBOOK: &str = "see docs/runbook.md#broker-state-unavailable";
+
+/// `brokerd`'s answer for `session`, asked on `socket`. `wait` is the read and the write timeout;
+/// the connect has none, and fails at once when nothing listens. Any error, an error frame
+/// included, is `Err` with one line that names the socket or `brokerd`.
+pub fn ask(
+    socket: &Path,
+    session: &SessionId,
+    wait: Duration,
+) -> Result<SessionStateReply, String> {
+    // `let line = |e: &dyn std::fmt::Display| format!("{}: {e}", socket.display());`. Then,
+    // each error through `.map_err(|e| line(&e))?`: `UnixStream::connect(socket)` (`mut stream`),
+    // `set_read_timeout(Some(wait))`, `set_write_timeout(Some(wait))`, and `write_frame` of an
+    // `Envelope { v: PROTOCOL_VERSION, id: 1, r#final: true, msg }` with `msg` the
+    // `Message::SessionState(SessionState { session: session.clone() })` (name it `question`).
+    // Then `match read_frame(&mut stream).map_err(|e| line(&e))?.msg`: `SessionStateReply(reply)`
+    // -> `Ok(reply)`; `Error(e)` -> `Err(format!("brokerd: {}", e.detail))`; anything else ->
+    // `Err("brokerd answered with the wrong message".to_string())`.
+    todo!()
+}

docs/plans/M5c1/files/crates/memoryd/src/honcho/turns.rs

@@ -0,0 +1,201 @@
+//! Deciding finished turns when `loopd` reports one (M5c spec, section 5). At each report every
+//! session with undecided turns is walked: its log is read, `brokerd` is asked once for its state,
+//! and each undecided turn is decided and appended to the ledger at once. A turn that cannot be
+//! decided now (`brokerd` silent, the log unreadable) waits for the next report of any session.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::path::{Path, PathBuf};
+
+use bxcore::{LogRecord, SessionId, SessionStateReply, Timestamp};
+
+use crate::honcho::decide::{decide, turn_records, turns};
+use crate::honcho::ledger::{self, Item, Ledger, Record};
+use crate::honcho::state;
+
+/// The ledger's file name, in `memory/`.
+pub const LEDGER: &str = "honcho.jsonl";
+
+/// The runbook entry for a session log with a damaged line.
+pub const LOG_RUNBOOK: &str = "see docs/runbook.md#session-log-damaged";
+
+/// `brokerd`'s answer for a session, or why there is none.
+pub type Ask<'a> = &'a dyn Fn(&SessionId) -> Result<SessionStateReply, String>;
+
+/// How one session's walk ended.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Walk {
+    /// Every turn up to the reported one is decided: the session leaves the map.
+    Done,
+    /// Some turn waits: the session stays in the map.
+    Waiting,
+    /// `brokerd` did not answer, or the ledger could not be written: the walk stops.
+    Stop,
+}
+
+/// The turns reported and not yet decided, and the ledger their decisions go to.
+pub struct TurnSync {
+    ledger: Ledger,
+    agent: String,
+    sessions: PathBuf,
+    /// Each session with undecided turns, by its id, and the highest turn `loopd` reported for
+    /// it. `SessionId` has no order, so the map is keyed by its text.
+    reported: BTreeMap<String, (SessionId, u64)>,
+    /// Whether the last question to `brokerd` went unanswered: an outage is logged once.
+    broker_down: bool,
+    /// Sessions whose log last had a damaged line: logged once each.
+    unreadable: BTreeSet<String>,
+}
+
+/// The path of `session`'s log under `sessions`: epoch 0, the only one `loopd` writes.
+pub fn log_path(sessions: &Path, session: &SessionId) -> PathBuf {
+    // `sessions.join(session.as_str()).join("0.jsonl")`.
+    todo!()
+}
+
+/// A session log's complete lines as records. A missing file has none. A last line without its
+/// newline is one `loopd` is still writing, and is left out. A complete line that does not parse
+/// is an error naming the file and the line, with the runbook pointer.
+pub fn read_log(path: &Path) -> Result<Vec<LogRecord>, String> {
+    // `std::fs::read(path)` into `text` (bytes): NotFound -> `return Ok(Vec::new())`; another
+    // error `e` -> `Err(format!("{}: {e}; {LOG_RUNBOOK}", path.display()))`. `whole` is one past
+    // the last `b'\n'` (`text.iter().rposition(..)`: `Some(last)` -> `last + 1`, `None` -> 0).
+    // Walk `text.get(..whole).unwrap_or_default().split_inclusive(|b| *b == b'\n').enumerate()`:
+    // each line through `serde_json::from_slice::<LogRecord>`, pushed; an error `e` -> `return
+    // Err(format!("{}:{}: {e}; {LOG_RUNBOOK}", path.display(), index + 1))`. `Ok(records)`.
+    todo!()
+}
+
+/// Whether a turn has ended with an answer: its last `Assistant` record asks for no tool.
+pub fn finished(turn: &[LogRecord]) -> bool {
+    // `let last = turn.iter().rev().find(|r| matches!(r, LogRecord::Assistant { .. }));` then
+    // `matches!(last, Some(LogRecord::Assistant { tool_calls, .. }) if tool_calls.is_empty())`.
+    todo!()
+}
+
+impl TurnSync {
+    /// `ledger` read through at start; `agent` is `[honcho] agent`; `sessions` the directory of
+    /// `loopd`'s session logs.
+    pub fn new(ledger: Ledger, agent: &str, sessions: &Path) -> TurnSync {
+        TurnSync {
+            ledger,
+            agent: agent.to_string(),
+            sessions: sessions.to_path_buf(),
+            reported: BTreeMap::new(),
+            broker_down: false,
+            unreadable: BTreeSet::new(),
+        }
+    }
+
+    pub fn ledger(&self) -> &Ledger {
+        &self.ledger
+    }
+
+    /// The sessions that still have undecided turns, with the highest turn reported for each.
+    pub fn waiting(&self) -> Vec<(SessionId, u64)> {
+        self.reported.values().cloned().collect()
+    }
+
+    /// `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.
+    pub fn updated(
+        &mut self,
+        session: &SessionId,
+        turn: u64,
+        ask: Ask<'_>,
+        now: Timestamp,
+        log: &dyn Fn(&str),
+    ) {
+        let entry = self
+            .reported
+            .entry(session.as_str().to_string())
+            .or_insert((session.clone(), 0));
+        entry.1 = entry.1.max(turn);
+        for (session, turn) in self.waiting() {
+            match self.walk(&session, turn, ask, now, log) {
+                Walk::Done => {
+                    self.reported.remove(session.as_str());
+                }
+                Walk::Waiting => {}
+                Walk::Stop => return,
+            }
+        }
+    }
+
+    /// Decide `session`'s undecided turns from 1 to `reported`, as far as its log goes. The log's
+    /// last turn waits until it has an answer: it may still be running.
+    fn walk(
+        &mut self,
+        session: &SessionId,
+        reported: u64,
+        ask: Ask<'_>,
+        now: Timestamp,
+        log: &dyn Fn(&str),
+    ) -> Walk {
+        let Some(records) = self.records(session, log) else {
+            return Walk::Waiting;
+        };
+        let (limit, held) = decidable(&records, reported);
+        let item = |turn: u64| Item::Turn {
+            session: session.clone(),
+            turn,
+        };
+        let undecided: Vec<u64> = (1..=limit)
+            .filter(|turn| !self.ledger.is_decided(&item(*turn)))
+            .collect();
+        if undecided.is_empty() {
+            return if held { Walk::Waiting } else { Walk::Done };
+        }
+        let Some(state) = self.state(session, ask, log) else {
+            return Walk::Stop;
+        };
+        for turn in undecided {
+            let Ok(action) = decide(&state, &records, turn, &self.agent) else {
+                continue;
+            };
+            let record = Record::Decided {
+                time: now,
+                item: item(turn),
+                action,
+            };
+            if let Err(e) = self.ledger.append(record) {
+                log(&format!("memoryd: {e}; {}", ledger::RUNBOOK));
+                return Walk::Stop;
+            }
+        }
+        if held { Walk::Waiting } else { Walk::Done }
+    }
+
+    /// `session`'s log, read. A log that does not read is logged once, until it reads again.
+    fn records(&mut self, session: &SessionId, log: &dyn Fn(&str)) -> Option<Vec<LogRecord>> {
+        // `match read_log(&log_path(&self.sessions, session))`: `Ok(records)` ->
+        // `self.unreadable.remove(session.as_str());` and `Some(records)`. `Err(e)` -> when
+        // `self.unreadable.insert(session.as_str().to_string())` is true (not logged yet),
+        // `log(&format!("memoryd: {e}"))`; then `None`.
+        todo!()
+    }
+
+    /// `brokerd`'s answer for `session`. The first failure after an answer is logged, and so is
+    /// the first answer after a failure.
+    fn state(
+        &mut self,
+        session: &SessionId,
+        ask: Ask<'_>,
+        log: &dyn Fn(&str),
+    ) -> Option<SessionStateReply> {
+        // `match ask(session)`. `Ok(state)`: if `self.broker_down`, set it `false` and log exactly
+        // `memoryd: brokerd answers on its state socket again; turns are decided`; `Some(state)`.
+        // `Err(e)`: if `!self.broker_down`, set it `true` and log
+        // `format!("memoryd: {e}; turns wait undecided; {}", state::RUNBOOK)`; then `None`.
+        todo!()
+    }
+}
+
+/// How far a log's turns may be decided for a report of turn `reported`: up to the smaller of
+/// `reported` and the log's turns, but for the log's last turn when it has no answer yet. The
+/// limit, and whether that last turn was held back.
+pub fn decidable(records: &[LogRecord], reported: u64) -> (u64, bool) {
+    // `let count = turns(records);` and `let limit = reported.min(count);`. When `limit > 0 &&
+    // limit == count && !turn_records(records, limit).is_some_and(finished)`: `(limit - 1, true)`;
+    // otherwise `(limit, false)`.
+    todo!()
+}