471 lines
16 KiB
Rust
471 lines
16 KiB
Rust
//! The Honcho ledger, `memory/honcho.jsonl`: one JSON line per event, appended and never rewritten
|
|
//! (M5c spec, section 5). A turn or a memory write is `decided` once, then `sent` or `failed` once.
|
|
//! What is decided and neither sent nor failed is the outbox; there is no other queue. `memoryd`
|
|
//! reads the ledger through at start, so nothing is decided or sent twice.
|
|
|
|
use std::collections::HashMap;
|
|
use std::fs::OpenOptions;
|
|
use std::io::Write;
|
|
use std::os::unix::fs::OpenOptionsExt;
|
|
use std::path::{Path, PathBuf};
|
|
|
|
use bxcore::{SessionId, Timestamp};
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
/// The runbook entry for a ledger that does not read through.
|
|
pub const RUNBOOK: &str = "see docs/runbook.md#honcho-ledger-damaged";
|
|
|
|
/// One of this agent's three workspaces.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
|
#[serde(rename_all = "snake_case")]
|
|
pub enum Workspace {
|
|
Trusted,
|
|
Untrusted,
|
|
Fleet,
|
|
}
|
|
|
|
/// What is sent: a finished turn of a session, or one part of a journalled memory edit.
|
|
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
|
#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
|
|
pub enum Item {
|
|
Turn { session: SessionId, turn: u64 },
|
|
Edit { id: String, part: u32 },
|
|
}
|
|
|
|
/// Where an item goes, decided once. A turn's text is read from the session log when it is sent,
|
|
/// never stored here.
|
|
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
|
#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
|
|
pub enum Action {
|
|
/// The turn's two messages, the speaker's and the agent's reply.
|
|
Turn {
|
|
workspace: Workspace,
|
|
speaker: String,
|
|
},
|
|
/// A conclusion about `peer`, stamped with the Honcho session of `session`.
|
|
Conclude {
|
|
workspace: Workspace,
|
|
peer: String,
|
|
session: SessionId,
|
|
text: String,
|
|
},
|
|
/// Delete the conclusion that a sent `Conclude` with this text created.
|
|
Unconclude {
|
|
workspace: Workspace,
|
|
peer: String,
|
|
text: String,
|
|
},
|
|
/// A message from the agent's peer in the fleet workspace.
|
|
Fleet { session: SessionId, text: String },
|
|
/// `retracted: <text>`, from the same peer in the same workspace.
|
|
Retract { session: SessionId, text: String },
|
|
/// Nothing is sent, and why.
|
|
Skip { why: String },
|
|
}
|
|
|
|
/// One line of the ledger.
|
|
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
|
#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
|
|
pub enum Record {
|
|
/// When sync began: always the ledger's first line, written once, by `memoryd` at its first
|
|
/// start with `[honcho]`. A turn from before it is not synced.
|
|
Began { time: Timestamp },
|
|
Decided {
|
|
time: Timestamp,
|
|
item: Item,
|
|
action: Action,
|
|
},
|
|
/// Honcho's ids for what the send created: a `Conclude` has exactly one.
|
|
Sent {
|
|
time: Timestamp,
|
|
item: Item,
|
|
ids: Vec<String>,
|
|
},
|
|
/// Honcho refused the send: its status, if it answered, and the kind of error.
|
|
Failed {
|
|
time: Timestamp,
|
|
item: Item,
|
|
status: Option<u16>,
|
|
kind: String,
|
|
},
|
|
}
|
|
|
|
/// How many items are in each state. `decided` counts every item, `Skip` included, so it is
|
|
/// always `sent + waiting + failed + skipped`.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
|
|
pub struct Counts {
|
|
pub decided: usize,
|
|
pub sent: usize,
|
|
pub waiting: usize,
|
|
pub failed: usize,
|
|
pub skipped: usize,
|
|
}
|
|
|
|
/// A failed item, as its `Failed` record says.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct Failure {
|
|
pub item: Item,
|
|
pub time: Timestamp,
|
|
pub status: Option<u16>,
|
|
pub kind: String,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub enum LedgerError {
|
|
Io(PathBuf, std::io::Error),
|
|
/// A complete line (numbered from 1) that does not parse, or breaks the ledger's rules.
|
|
Damaged {
|
|
path: PathBuf,
|
|
line: usize,
|
|
why: String,
|
|
},
|
|
Encode(serde_json::Error),
|
|
AlreadyDecided(Item),
|
|
NotWaiting(Item),
|
|
/// A `Began` after other records, or a second one.
|
|
BeganNotFirst,
|
|
/// A ledger whose first line is not its `Began`.
|
|
NotBegun,
|
|
}
|
|
|
|
impl std::fmt::Display for LedgerError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
LedgerError::Io(path, err) => write!(f, "{}: {err}", path.display()),
|
|
LedgerError::Damaged { path, line, why } => {
|
|
write!(f, "{}:{line}: {why}; {RUNBOOK}", path.display())
|
|
}
|
|
LedgerError::Encode(err) => write!(f, "a ledger record could not be encoded: {err}"),
|
|
LedgerError::AlreadyDecided(item) => write!(f, "{}: decided already", label(item)),
|
|
LedgerError::NotWaiting(item) => {
|
|
write!(f, "{}: not waiting to be sent", label(item))
|
|
}
|
|
LedgerError::BeganNotFirst => write!(f, "sync began, but not on the first line"),
|
|
LedgerError::NotBegun => write!(f, "the first line is not when sync began"),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl std::error::Error for LedgerError {}
|
|
|
|
/// Where a decided item stands.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
enum State {
|
|
Skipped,
|
|
Waiting,
|
|
Sent(Vec<String>),
|
|
Failed {
|
|
time: Timestamp,
|
|
status: Option<u16>,
|
|
kind: String,
|
|
},
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
struct Entry {
|
|
action: Action,
|
|
state: State,
|
|
}
|
|
|
|
/// The ledger, read through: the file's path and the state its records add up to.
|
|
#[derive(Debug)]
|
|
pub struct Ledger {
|
|
path: PathBuf,
|
|
/// Every decided item, in the order of its `Decided` record.
|
|
order: Vec<Item>,
|
|
entries: HashMap<Item, Entry>,
|
|
/// Every sent item, in the order of its `Sent` record.
|
|
sends: Vec<Item>,
|
|
/// Every failed item, in the order of its `Failed` record.
|
|
fails: Vec<Item>,
|
|
/// When sync began: the `Began` record's time.
|
|
began: Option<Timestamp>,
|
|
/// How many records the ledger holds.
|
|
records: usize,
|
|
repaired: bool,
|
|
}
|
|
|
|
impl Ledger {
|
|
/// Read the ledger at `path` through. A missing file is an empty ledger, and nothing is
|
|
/// created. A last line without its newline, cut off by a crash, is dropped and cut from the
|
|
/// file; any other line that does not read is `Damaged`, and the file is left as it is.
|
|
pub fn open(path: &Path) -> Result<Ledger, LedgerError> {
|
|
let mut ledger = Ledger {
|
|
path: path.to_path_buf(),
|
|
order: Vec::new(),
|
|
entries: HashMap::new(),
|
|
sends: Vec::new(),
|
|
fails: Vec::new(),
|
|
began: None,
|
|
records: 0,
|
|
repaired: false,
|
|
};
|
|
let Some(bytes) = read(path)? else {
|
|
return Ok(ledger);
|
|
};
|
|
let whole = complete(&bytes);
|
|
let lines = bytes.get(..whole).unwrap_or_default();
|
|
for (index, line) in lines.split_inclusive(|b| *b == b'\n').enumerate() {
|
|
let record = parse(path, index + 1, line)?;
|
|
let first_not_began = index == 0 && !matches!(record, Record::Began { .. });
|
|
let checked = if first_not_began {
|
|
Err(LedgerError::NotBegun)
|
|
} else {
|
|
ledger.check(&record)
|
|
};
|
|
if let Err(e) = checked {
|
|
return Err(LedgerError::Damaged {
|
|
path: path.to_path_buf(),
|
|
line: index + 1,
|
|
why: e.to_string(),
|
|
});
|
|
}
|
|
ledger.apply(record);
|
|
}
|
|
if whole < bytes.len() {
|
|
cut(path, whole)?;
|
|
ledger.repaired = true;
|
|
}
|
|
Ok(ledger)
|
|
}
|
|
|
|
/// Whether `open` dropped a cut-off last line.
|
|
pub fn repaired(&self) -> bool {
|
|
self.repaired
|
|
}
|
|
|
|
/// Append `record` to the file and sync it, and only then take it into the state. A record
|
|
/// the rules refuse, or a failed write, changes nothing in memory.
|
|
pub fn append(&mut self, record: Record) -> Result<(), LedgerError> {
|
|
self.check(&record)?;
|
|
let line = serde_json::to_string(&record).map_err(LedgerError::Encode)?;
|
|
self.write_line(&line)?;
|
|
self.apply(record);
|
|
Ok(())
|
|
}
|
|
|
|
/// When sync began, once the ledger has its `Began` record.
|
|
pub fn began(&self) -> Option<Timestamp> {
|
|
self.began
|
|
}
|
|
|
|
/// Whether `item` has its `Decided` record.
|
|
pub fn is_decided(&self, item: &Item) -> bool {
|
|
self.entries.contains_key(item)
|
|
}
|
|
|
|
/// Every decided item with its action, `Skip` included, in the order they were decided.
|
|
pub fn decisions(&self) -> impl Iterator<Item = (&Item, &Action)> + '_ {
|
|
self.order
|
|
.iter()
|
|
.filter_map(|item| Some((item, &self.entries.get(item)?.action)))
|
|
}
|
|
|
|
/// The items waiting to be sent, with their actions, in the order they were decided.
|
|
pub fn outbox(&self) -> Vec<(Item, Action)> {
|
|
self.order
|
|
.iter()
|
|
.filter_map(|item| {
|
|
let entry = self.entries.get(item)?;
|
|
(entry.state == State::Waiting).then(|| (item.clone(), entry.action.clone()))
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// The ids of `item`'s `Sent` record, if it has one.
|
|
pub fn sent_ids(&self, item: &Item) -> Option<&[String]> {
|
|
match &self.entries.get(item)?.state {
|
|
State::Sent(ids) => Some(ids.as_slice()),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
/// The latest sent `Conclude` of `text` about `peer` in `workspace` that no sent `Unconclude`
|
|
/// of the same has undone: its item and the conclusion's id.
|
|
pub fn conclusion(
|
|
&self,
|
|
workspace: Workspace,
|
|
peer: &str,
|
|
text: &str,
|
|
) -> Option<(Item, String)> {
|
|
let mut live: Vec<(Item, String)> = Vec::new();
|
|
for item in &self.sends {
|
|
let Some(entry) = self.entries.get(item) else {
|
|
continue;
|
|
};
|
|
let State::Sent(ids) = &entry.state else {
|
|
continue;
|
|
};
|
|
match &entry.action {
|
|
Action::Conclude {
|
|
workspace: w,
|
|
peer: p,
|
|
text: t,
|
|
..
|
|
} if *w == workspace && p == peer && t == text => {
|
|
if let Some(id) = ids.first() {
|
|
live.push((item.clone(), id.clone()));
|
|
}
|
|
}
|
|
Action::Unconclude {
|
|
workspace: w,
|
|
peer: p,
|
|
text: t,
|
|
} if *w == workspace && p == peer && t == text => {
|
|
live.pop();
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
live.pop()
|
|
}
|
|
|
|
/// Every failed item, oldest first (the order of the `Failed` records), with its record's
|
|
/// time, status and kind.
|
|
pub fn failures(&self) -> Vec<Failure> {
|
|
self.fails
|
|
.iter()
|
|
.filter_map(|item| match &self.entries.get(item)?.state {
|
|
State::Failed { time, status, kind } => Some(Failure {
|
|
item: item.clone(),
|
|
time: *time,
|
|
status: *status,
|
|
kind: kind.clone(),
|
|
}),
|
|
_ => None,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// How many items are in each state.
|
|
pub fn counts(&self) -> Counts {
|
|
let mut counts = Counts::default();
|
|
for entry in self.entries.values() {
|
|
counts.decided += 1;
|
|
match &entry.state {
|
|
State::Skipped => counts.skipped += 1,
|
|
State::Waiting => counts.waiting += 1,
|
|
State::Sent(_) => counts.sent += 1,
|
|
State::Failed { .. } => counts.failed += 1,
|
|
}
|
|
}
|
|
counts
|
|
}
|
|
|
|
/// The ledger's rules: an item is decided once; it is sent or failed once, and only while it
|
|
/// is waiting.
|
|
fn check(&self, record: &Record) -> Result<(), LedgerError> {
|
|
match record {
|
|
Record::Began { .. } => {
|
|
if self.records > 0 {
|
|
Err(LedgerError::BeganNotFirst)
|
|
} else {
|
|
Ok(())
|
|
}
|
|
}
|
|
Record::Decided { item, .. } => {
|
|
if self.entries.contains_key(item) {
|
|
Err(LedgerError::AlreadyDecided(item.clone()))
|
|
} else {
|
|
Ok(())
|
|
}
|
|
}
|
|
Record::Sent { item, .. } | Record::Failed { item, .. } => match self.entries.get(item)
|
|
{
|
|
Some(entry) if entry.state == State::Waiting => Ok(()),
|
|
_ => Err(LedgerError::NotWaiting(item.clone())),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Take a checked record into the state.
|
|
fn apply(&mut self, record: Record) {
|
|
self.records += 1;
|
|
match record {
|
|
Record::Began { time } => self.began = Some(time),
|
|
Record::Decided { item, action, .. } => {
|
|
let state = match action {
|
|
Action::Skip { .. } => State::Skipped,
|
|
_ => State::Waiting,
|
|
};
|
|
self.order.push(item.clone());
|
|
self.entries.insert(item, Entry { action, state });
|
|
}
|
|
Record::Sent { item, ids, .. } => {
|
|
if let Some(entry) = self.entries.get_mut(&item) {
|
|
entry.state = State::Sent(ids);
|
|
}
|
|
self.sends.push(item);
|
|
}
|
|
Record::Failed {
|
|
time,
|
|
item,
|
|
status,
|
|
kind,
|
|
} => {
|
|
if let Some(entry) = self.entries.get_mut(&item) {
|
|
entry.state = State::Failed { time, status, kind };
|
|
}
|
|
self.fails.push(item);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Append `line` and a newline to the file in one write, then sync the data.
|
|
fn write_line(&self, line: &str) -> Result<(), LedgerError> {
|
|
let io = |e| LedgerError::Io(self.path.clone(), e);
|
|
let mut file = OpenOptions::new()
|
|
.append(true)
|
|
.create(true)
|
|
.mode(0o600)
|
|
.open(&self.path)
|
|
.map_err(io)?;
|
|
file.write_all(format!("{line}\n").as_bytes()).map_err(io)?;
|
|
file.sync_data().map_err(io)
|
|
}
|
|
}
|
|
|
|
/// The file's bytes, or `None` when there is no file.
|
|
fn read(path: &Path) -> Result<Option<Vec<u8>>, LedgerError> {
|
|
match std::fs::read(path) {
|
|
Ok(bytes) => Ok(Some(bytes)),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
|
|
Err(e) => Err(LedgerError::Io(path.to_path_buf(), e)),
|
|
}
|
|
}
|
|
|
|
/// How many bytes the complete lines take: up to and including the last newline.
|
|
fn complete(bytes: &[u8]) -> usize {
|
|
match bytes.iter().rposition(|b| *b == b'\n') {
|
|
Some(last) => last + 1,
|
|
None => 0,
|
|
}
|
|
}
|
|
|
|
/// Line `number` (from 1), with its newline, as a record.
|
|
fn parse(path: &Path, number: usize, line: &[u8]) -> Result<Record, LedgerError> {
|
|
serde_json::from_slice(line).map_err(|e| LedgerError::Damaged {
|
|
path: path.to_path_buf(),
|
|
line: number,
|
|
why: e.to_string(),
|
|
})
|
|
}
|
|
|
|
/// Cut the file to its first `keep` bytes, and sync it.
|
|
fn cut(path: &Path, keep: usize) -> Result<(), LedgerError> {
|
|
let io = |e| LedgerError::Io(path.to_path_buf(), e);
|
|
let len = u64::try_from(keep).map_err(|e| io(std::io::Error::other(e)))?;
|
|
let file = OpenOptions::new().write(true).open(path).map_err(io)?;
|
|
file.set_len(len).map_err(io)?;
|
|
file.sync_data().map_err(io)
|
|
}
|
|
|
|
/// An item as the error messages and `bxctl memory sync` name it: `turn 3 of session mm-a`,
|
|
/// `part 0 of edit 2026-09-28-0001`.
|
|
pub 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}"),
|
|
}
|
|
}
|