memoryd: sync begins when Honcho is turned on, and conclusions are matched by peer

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>

crates/memoryd/src/honcho/decide.rs

@@ -3,12 +3,13 @@
 //! and the session's log are given, and nothing is read or written here.
 
 use bxcore::sanitize::refuse;
-use bxcore::{DataClass, LogRecord, SessionStateReply};
+use bxcore::{DataClass, LogRecord, SessionStateReply, Timestamp};
 
 use crate::honcho::config::is_name;
 use crate::honcho::ledger::{Action, Workspace};
 
 /// Why a turn goes nowhere: the `why` of its `Action::Skip`.
+pub const BEFORE_SYNC: &str = "the turn is from before sync began";
 pub const SECRET_CLASS: &str = "the session is secret";
 pub const NO_SPEAKER: &str = "the turn has no single speaker";
 pub const BAD_SPEAKER: &str = "the speaker is not a Honcho peer name";
@@ -135,8 +136,9 @@ fn labelled_untrusted(records: &[LogRecord]) -> bool {
 }
 
 /// Where turn `turn` of the session whose log is `records` goes. `state` is `brokerd`'s answer
-/// for the session; `agent` is `[honcho] agent`. The checks run in this order, and the first that
-/// holds decides: the class `secret`; no speaker; a speaker that is not a peer name, or is the
+/// for the session; `agent` is `[honcho] agent`; `began` is when sync began (the ledger's
+/// `Began`). The checks run in this order, and the first that holds decides: the turn is from
+/// before `began`; the class `secret`; no speaker; a speaker that is not a peer name, or is the
 /// agent's; no reply; a secret in the user's text or the reply. Otherwise the turn goes to the
 /// untrusted workspace if `brokerd` or any tool result up to the turn's end says untrusted, else
 /// to the trusted one.
@@ -145,6 +147,7 @@ pub fn decide(
     records: &[LogRecord],
     turn: u64,
     agent: &str,
+    began: Option<Timestamp>,
 ) -> Result<Action, NoSuchTurn> {
     let missing = NoSuchTurn {
         turn,
@@ -159,6 +162,11 @@ pub fn decide(
         })
     };
 
+    if let (Some(began), Some(LogRecord::User { time, .. })) = (began, this.first())
+        && *time < began
+    {
+        return skip(BEFORE_SYNC);
+    }
     if state.class == DataClass::Secret {
         return skip(SECRET_CLASS);
     }

crates/memoryd/src/honcho/ledger.rs

@@ -50,7 +50,11 @@ pub enum Action {
         text: String,
     },
     /// Delete the conclusion that a sent `Conclude` with this text created.
-    Unconclude { workspace: Workspace, text: String },
+    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.
@@ -63,6 +67,9 @@ pub enum Action {
 #[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,
@@ -115,6 +122,10 @@ pub enum LedgerError {
     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 {
@@ -129,6 +140,8 @@ impl std::fmt::Display for LedgerError {
             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"),
         }
     }
 }
@@ -165,6 +178,10 @@ pub struct Ledger {
     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,
 }
 
@@ -179,6 +196,8 @@ impl Ledger {
             entries: HashMap::new(),
             sends: Vec::new(),
             fails: Vec::new(),
+            began: None,
+            records: 0,
             repaired: false,
         };
         let Some(bytes) = read(path)? else {
@@ -188,7 +207,13 @@ impl Ledger {
         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) {
+            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,
@@ -219,6 +244,11 @@ impl Ledger {
         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)
@@ -250,9 +280,14 @@ impl Ledger {
         }
     }
 
-    /// 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)> {
+    /// 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 {
@@ -264,17 +299,19 @@ impl Ledger {
             match &entry.action {
                 Action::Conclude {
                     workspace: w,
+                    peer: p,
                     text: t,
                     ..
-                } if *w == workspace && t == text => {
+                } 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 && t == text => {
+                } if *w == workspace && p == peer && t == text => {
                     live.pop();
                 }
                 _ => {}
@@ -319,6 +356,13 @@ impl Ledger {
     /// 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()))
@@ -336,7 +380,9 @@ impl Ledger {
 
     /// 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,

crates/memoryd/src/honcho/mirror.rs

@@ -96,6 +96,7 @@ fn identity_actions(
         .into_iter()
         .map(|text| Action::Unconclude {
             workspace: Workspace::Trusted,
+            peer: peer.to_string(),
             text,
         })
         .collect();

crates/memoryd/src/honcho/start.rs

@@ -3,12 +3,12 @@
 
 use std::sync::{Arc, Mutex};
 
-use bxcore::SessionId;
+use bxcore::{SessionId, Timestamp};
 
 use crate::config::Config;
 use crate::honcho::client::{Clients, Timeouts};
 use crate::honcho::config::HonchoConfig;
-use crate::honcho::ledger::Ledger;
+use crate::honcho::ledger::{Ledger, Record};
 use crate::honcho::mirror::Mirror;
 use crate::honcho::state;
 use crate::honcho::turns::{LEDGER, TurnSync};
@@ -25,6 +25,20 @@ pub struct Opened {
 }
 
 impl Opened {
+    /// The first time: sync begins now. The `Began` line is written and synced once `memory/`
+    /// exists and before any socket is bound, so every turn reported later is after it.
+    pub fn begin(&self) -> Result<(), StartError> {
+        let sync = self.sync.lock().unwrap_or_else(|p| p.into_inner());
+        let mut ledger = sync.ledger();
+        if ledger.began().is_none() {
+            let began = Record::Began {
+                time: Timestamp::now(),
+            };
+            ledger.append(began).map_err(StartError::Ledger)?;
+        }
+        Ok(())
+    }
+
     /// The memory writes' side of the ledger (task 12).
     pub fn mirror(&self) -> Mirror {
         let ledger = self

crates/memoryd/src/honcho/turns.rs

@@ -218,7 +218,8 @@ impl TurnSync {
             return Walk::Stop;
         };
         for turn in undecided {
-            let Ok(action) = decide(&state, &records, turn, &self.agent) else {
+            let began = self.ledger().began();
+            let Ok(action) = decide(&state, &records, turn, &self.agent, began) else {
                 continue;
             };
             let record = Record::Decided {

crates/memoryd/src/honcho/worker.rs

@@ -345,10 +345,17 @@ impl Worker {
                         text,
                     },
                 ),
-                (Item::Edit { .. }, Action::Unconclude { workspace, text }) => (
+                (
+                    Item::Edit { .. },
+                    Action::Unconclude {
+                        workspace,
+                        peer,
+                        text,
+                    },
+                ) => (
                     workspace,
                     Job::Unconclude {
-                        id: ledger.conclusion(workspace, &text).map(|(_, id)| id),
+                        id: ledger.conclusion(workspace, &peer, &text).map(|(_, id)| id),
                     },
                 ),
                 (Item::Edit { id, .. }, Action::Fleet { session, text }) => (

crates/memoryd/src/serve.rs

@@ -93,8 +93,8 @@ enum Which {
 }
 
 /// 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.
+/// under `memory/` (and, the first time, write when sync began), 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)?),
@@ -102,6 +102,9 @@ pub fn start(config: Config, log: Log) -> Result<Running, StartError> {
     };
     let store = Store::new(&config.memory_dir());
     store.init(&config.memory.soul_starter)?;
+    if let Some(opened) = &opened {
+        opened.begin()?;
+    }
     let index = Index::open(&config.index_path(), config.index.max_class)?;
     let (_counts, problems) = index.catch_up(&config.sessions_dir(), &store.path("notes.md"))?;
     for problem in problems {