memoryd: the Honcho ledger, decided, sent and failed, read through at start
Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
This commit is contained in:
ornith-1.5-35b-a3b
@@ -0,0 +1,377 @@
|
||||
//! 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, 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 {
|
||||
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,
|
||||
}
|
||||
|
||||
#[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),
|
||||
}
|
||||
|
||||
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", describe(item)),
|
||||
LedgerError::NotWaiting(item) => {
|
||||
write!(f, "{}: not waiting to be sent", describe(item))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for LedgerError {}
|
||||
|
||||
/// Where a decided item stands.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
enum State {
|
||||
Skipped,
|
||||
Waiting,
|
||||
Sent(Vec<String>),
|
||||
Failed,
|
||||
}
|
||||
|
||||
#[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>,
|
||||
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(),
|
||||
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)?;
|
||||
if let Err(e) = ledger.check(&record) {
|
||||
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(())
|
||||
}
|
||||
|
||||
/// Whether `item` has its `Decided` record.
|
||||
pub fn is_decided(&self, item: &Item) -> bool {
|
||||
self.entries.contains_key(item)
|
||||
}
|
||||
|
||||
/// 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` in `workspace` that no sent `Unconclude` has undone:
|
||||
/// its item and the conclusion's id.
|
||||
pub fn conclusion(&self, workspace: Workspace, 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,
|
||||
text: t,
|
||||
..
|
||||
} if *w == workspace && t == text => {
|
||||
if let Some(id) = ids.first() {
|
||||
live.push((item.clone(), id.clone()));
|
||||
}
|
||||
}
|
||||
Action::Unconclude {
|
||||
workspace: w,
|
||||
text: t,
|
||||
} if *w == workspace && t == text => {
|
||||
live.pop();
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
live.pop()
|
||||
}
|
||||
|
||||
/// 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::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) {
|
||||
match record {
|
||||
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 { item, .. } => {
|
||||
if let Some(entry) = self.entries.get_mut(&item) {
|
||||
entry.state = State::Failed;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 name it.
|
||||
fn describe(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}"),
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
//! Honcho, the memory this agent shares with the fleet, reached through `roncho` (M5c spec).
|
||||
|
||||
pub mod config;
|
||||
pub mod ledger;
|
||||
|
||||
Reference in New Issue
Block a user