From 2c70a68dce7d07c3f0fd0bd6d3b04339dc76b22d Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 4 Sep 2026 17:46:29 -0400 Subject: [PATCH] feat(desktop): persist ordered placement and lifecycle admission Signed-off-by: Logan Johnson --- .../src-tauri/src/managed_agents/placement.rs | 217 +++++++++++++ .../src/managed_agents/placement/tests.rs | 284 ++++++++++++++++++ 2 files changed, 501 insertions(+) create mode 100644 desktop/src-tauri/src/managed_agents/placement.rs create mode 100644 desktop/src-tauri/src/managed_agents/placement/tests.rs diff --git a/desktop/src-tauri/src/managed_agents/placement.rs b/desktop/src-tauri/src/managed_agents/placement.rs new file mode 100644 index 00000000000..13efe7a8b19 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/placement.rs @@ -0,0 +1,217 @@ +//! Compact intent, separate from one-shot execution. No history replay. +use buzz_core_pkg::{ + desktop_lifecycle::{Action, Outcome, Request, ResultMessage}, + desktop_stop::StopTarget, + kind::KIND_DESKTOP_STOP, +}; +use nostr::{Event, JsonUtil, Keys}; +use rusqlite::{params, Connection, OptionalExtension}; + +pub(crate) fn schema(conn: &Connection) -> Result<(), String> { + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS desktop_placement ( + agent TEXT NOT NULL, slot TEXT NOT NULL, host TEXT NOT NULL, stamp INTEGER NOT NULL, + id TEXT NOT NULL, PRIMARY KEY(agent,slot)); + CREATE TABLE IF NOT EXISTS desktop_lifecycle_admission ( + agent TEXT NOT NULL, host TEXT NOT NULL, action TEXT NOT NULL, stamp INTEGER NOT NULL, + id TEXT NOT NULL, PRIMARY KEY(agent,host,action)); + CREATE TABLE IF NOT EXISTS desktop_lifecycle_results ( + id TEXT PRIMARY KEY, raw TEXT NOT NULL);", + ) + .map_err(|e| e.to_string()) +} + +/// Start's shared max and each host's Stop max are sufficient statistics. +/// Stopping newest Start never falls back to an earlier host. +pub(crate) fn observe( + conn: &Connection, + event: &Event, + keys: &Keys, + community: &str, +) -> Result { + let (target, slot) = if event.kind.as_u16() as u32 == KIND_DESKTOP_STOP { + let target = StopTarget::read(event, keys, community)?; + let slot = format!("stop:{}", target.desktop); + (target, slot) + } else { + let request = Request::read(event, keys, community)?; + if request.action != Action::Start { + return Ok(request.target.agent); + } + (request.target, "start".into()) + }; + schema(conn)?; + conn.execute( + "INSERT INTO desktop_placement VALUES (?1,?2,?3,?4,?5) + ON CONFLICT(agent,slot) DO UPDATE SET host=excluded.host,stamp=excluded.stamp,id=excluded.id + WHERE excluded.stamp > desktop_placement.stamp OR + (excluded.stamp = desktop_placement.stamp AND excluded.id < desktop_placement.id)", + params![ + target.agent, + slot, + target.desktop, + event.created_at.as_secs(), + event.id.to_hex() + ], + ) + .map_err(|e| e.to_string())?; + Ok(target.agent) +} + +type Row = (String, u64, String); +fn row(conn: &Connection, agent: &str, slot: &str) -> Result, String> { + schema(conn)?; + conn.query_row( + "SELECT host,stamp,id FROM desktop_placement WHERE agent=?1 AND slot=?2", + params![agent, slot], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .optional() + .map_err(|e| e.to_string()) +} +fn newer(a: &Row, b: &Row) -> bool { + a.1 > b.1 || (a.1 == b.1 && a.2 < b.2) +} + +/// Some Start remains desired, or none. Intent does not establish process state. +pub(crate) fn desired(conn: &Connection, agent: &str) -> Result, String> { + let Some(start) = row(conn, agent, "start")? else { + return Ok(None); + }; + if row(conn, agent, &format!("stop:{}", start.0))?.is_some_and(|stop| newer(&stop, &start)) { + return Ok(None); + } + Ok(Some((start.0, start.2))) +} + +/// Unknown preserves existing local behavior; known supersession blocks every spawn. +pub(crate) fn blocked(conn: &Connection, agent: &str, host: &str) -> Result { + let start = row(conn, agent, "start")?; + let stop = row(conn, agent, &format!("stop:{host}"))?; + Ok(match (start, stop) { + (None, Some(_)) => true, + (None, None) => false, + (Some(start), Some(stop)) if newer(&stop, &start) => true, + (Some(start), _) => start.0 != host, + }) +} + +/// Durable high-water marks are never evicted with diagnostic/result history. +pub(crate) fn admit(conn: &Connection, event: &Event, request: &Request) -> Result { + schema(conn)?; + let action = match request.action { + Action::Start => "start", + Action::Restart => "restart", + Action::Status => "status", + }; + let changed = conn.execute("INSERT INTO desktop_lifecycle_admission VALUES (?1,?2,?3,?4,?5) + ON CONFLICT(agent,host,action) DO UPDATE SET stamp=excluded.stamp,id=excluded.id + WHERE excluded.stamp > desktop_lifecycle_admission.stamp OR + (excluded.stamp = desktop_lifecycle_admission.stamp AND excluded.id < desktop_lifecycle_admission.id)", + params![request.target.agent,request.target.desktop,action,event.created_at.as_secs(),event.id.to_hex()]).map_err(|e| e.to_string())?; + Ok(changed == 1) +} +pub(crate) fn saved(conn: &Connection, id: &str) -> Result, String> { + schema(conn)?; + let raw: Option = conn + .query_row( + "SELECT raw FROM desktop_lifecycle_results WHERE id=?1", + [id], + |r| r.get(0), + ) + .optional() + .map_err(|e| e.to_string())?; + raw.map(|s| Event::from_json(s).map_err(|e| e.to_string())) + .transpose() +} +pub(crate) fn save(conn: &mut Connection, id: &str, event: &Event) -> Result<(), String> { + schema(conn)?; + let tx = conn.transaction().map_err(|e| e.to_string())?; + tx.execute( + "INSERT OR IGNORE INTO desktop_lifecycle_results VALUES (?1,?2)", + params![id, event.as_json()], + ) + .map_err(|e| e.to_string())?; + tx.execute("DELETE FROM desktop_lifecycle_results WHERE rowid NOT IN (SELECT rowid FROM desktop_lifecycle_results ORDER BY rowid DESC LIMIT 256)", []).map_err(|e|e.to_string())?; + tx.commit().map_err(|e| e.to_string()) +} + +#[cfg(test)] +mod tests; + +pub(crate) fn has_start(conn: &Connection, agent: &str) -> Result { + Ok(row(conn, agent, "start")?.is_some()) +} +pub(crate) fn latest_start( + conn: &Connection, + agent: &str, +) -> Result, String> { + Ok(row(conn, agent, "start")?.map(|(host, _, id)| (host, id))) +} + +/// A newer local Stop invalidates stale Restart even after a subsequent Start. +pub(crate) fn stale_restart( + conn: &Connection, + event: &Event, + request: &Request, +) -> Result { + let command = ( + request.target.desktop.clone(), + event.created_at.as_secs(), + event.id.to_hex(), + ); + Ok(row( + conn, + &request.target.agent, + &format!("stop:{}", request.target.desktop), + )? + .is_some_and(|stop| newer(&stop, &command))) +} + +/// Authenticate and consume before effects; crashes and evicted results never +/// turn an exact retry into a fresh launch or Restart. +pub(crate) fn receive( + conn: &mut Connection, + event: &Event, + keys: &Keys, + community: &str, + desktop: &str, + owned: bool, + effect: impl FnOnce(&Connection, &Request) -> Result, +) -> Result, String> { + let request = Request::read(event, keys, community)?; + observe(conn, event, keys, community)?; + if request.target.desktop != desktop { + return Ok(None); + } + if let Some(saved) = saved(conn, &event.id.to_hex())? { + ResultMessage::read(&saved, keys, event, community)?; + return Ok(Some(saved)); + } + let outcome = if !owned { + Outcome::Failed + } else if !admit(conn, event, &request)? + || (request.action != Action::Status + && (blocked(conn, &request.target.agent, desktop)? + || (request.action == Action::Restart && stale_restart(conn, event, &request)?) + || (request.action == Action::Start + && desired(conn, &request.target.agent)?.map(|(_, id)| id) + != Some(event.id.to_hex())))) + { + Outcome::Unknown + } else { + effect(conn, &request).unwrap_or(if request.action == Action::Status { + Outcome::Unknown + } else { + Outcome::Failed + }) + }; + let result = ResultMessage { + request, + id: event.id.to_hex(), + outcome, + } + .sign(keys)?; + save(conn, &event.id.to_hex(), &result)?; + Ok(Some(result)) +} diff --git a/desktop/src-tauri/src/managed_agents/placement/tests.rs b/desktop/src-tauri/src/managed_agents/placement/tests.rs new file mode 100644 index 00000000000..acf98915e58 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/placement/tests.rs @@ -0,0 +1,284 @@ +use super::*; +use buzz_core_pkg::desktop_lifecycle::{Outcome, ResultMessage}; +use nostr::Timestamp; +fn event(keys: &Keys, host: &str, start: bool, stamp: u64) -> Event { + let target = StopTarget { + v: 1, + community: "wss://one.example".into(), + desktop: host.repeat(32), + agent: keys.public_key().to_hex(), + }; + let event = if start { + Request { + target, + action: Action::Start, + observed: None, + } + .sign(keys) + .unwrap() + } else { + target.sign(keys).unwrap() + }; + nostr::EventBuilder::new(event.kind, event.content) + .tags(event.tags) + .custom_created_at(Timestamp::from(stamp)) + .sign_with_keys(keys) + .unwrap() +} +#[test] +fn opposite_arrival_and_scoped_stops_converge_without_resurrection() { + let keys = Keys::generate(); + let agent = keys.public_key().to_hex(); + let events = [ + event(&keys, "a", true, 1), + event(&keys, "b", true, 2), + event(&keys, "a", false, 3), + ]; + for order in [vec![0, 1, 2], vec![2, 0, 1], vec![1, 2, 0]] { + let conn = Connection::open_in_memory().unwrap(); + for i in order { + observe(&conn, &events[i], &keys, "wss://one.example").unwrap(); + } + assert_eq!( + desired(&conn, &agent).unwrap(), + Some(("b".repeat(32), events[1].id.to_hex())) + ); + assert!(blocked(&conn, &agent, &"a".repeat(32)).unwrap()); + assert!(!blocked(&conn, &agent, &"b".repeat(32)).unwrap()); + observe( + &conn, + &event(&keys, "b", false, 4), + &keys, + "wss://one.example", + ) + .unwrap(); + assert_eq!(desired(&conn, &agent).unwrap(), None); + assert!(blocked(&conn, &agent, &"b".repeat(32)).unwrap()); + observe(&conn, &events[0], &keys, "wss://one.example").unwrap(); + assert_eq!(desired(&conn, &agent).unwrap(), None); + } +} +#[test] +fn same_second_lower_id_and_future_timestamp_are_authority() { + let keys = Keys::generate(); + let conn = Connection::open_in_memory().unwrap(); + let a = event(&keys, "a", true, 1000); + let b = event(&keys, "b", true, 1000); + for e in [&a, &b, &a] { + observe(&conn, e, &keys, "wss://one.example").unwrap(); + } + let winner = if a.id < b.id { &a } else { &b }; + assert_eq!( + desired(&conn, &keys.public_key().to_hex()) + .unwrap() + .unwrap() + .1, + winner.id.to_hex() + ); + observe( + &conn, + &event(&keys, "c", true, 999), + &keys, + "wss://one.example", + ) + .unwrap(); + assert_eq!( + desired(&conn, &keys.public_key().to_hex()) + .unwrap() + .unwrap() + .1, + winner.id.to_hex() + ); +} +#[test] +fn consumption_survives_restart_and_result_eviction() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("journal.db"); + let keys = Keys::generate(); + let event = event(&keys, "a", true, 1); + let request = Request::read(&event, &keys, "wss://one.example").unwrap(); + let mut conn = Connection::open(&path).unwrap(); + assert!(admit(&conn, &event, &request).unwrap()); + for i in 0..258 { + let result = ResultMessage { + request: request.clone(), + id: event.id.to_hex(), + outcome: Outcome::Unknown, + } + .sign(&keys) + .unwrap(); + save(&mut conn, &i.to_string(), &result).unwrap(); + } + drop(conn); + let conn = Connection::open(&path).unwrap(); + assert!(!admit(&conn, &event, &request).unwrap()); + assert!(saved(&conn, "0").unwrap().is_none()); + assert!(admit(&conn, &super::tests::event(&keys, "a", true, 2), &request).unwrap()); +} + +#[test] +fn receiver_consumes_before_effect_and_never_repeats_restart() { + let keys = Keys::generate(); + let mut conn = Connection::open_in_memory().unwrap(); + let start = event(&keys, "a", true, 10); + let start_request = Request::read(&start, &keys, "wss://one.example").unwrap(); + observe(&conn, &start, &keys, "wss://one.example").unwrap(); + let restart = Request { + action: Action::Restart, + observed: Some("f".repeat(64)), + ..start_request + } + .sign(&keys) + .unwrap(); + let mut effects = 0; + let result = receive( + &mut conn, + &restart, + &keys, + "wss://one.example", + &"a".repeat(32), + true, + |conn, request| { + assert!( + !admit(conn, &restart, request).unwrap(), + "effect must see durable consumption" + ); + effects += 1; + Ok(Outcome::Running) + }, + ) + .unwrap() + .unwrap(); + let retry = receive( + &mut conn, + &restart, + &keys, + "wss://one.example", + &"a".repeat(32), + true, + |_, _| panic!("duplicate effect"), + ) + .unwrap() + .unwrap(); + assert_eq!(result, retry); + assert_eq!(effects, 1); + conn.execute("DELETE FROM desktop_lifecycle_results", []) + .unwrap(); + let unknown = receive( + &mut conn, + &restart, + &keys, + "wss://one.example", + &"a".repeat(32), + true, + |_, _| panic!("evicted effect"), + ) + .unwrap() + .unwrap(); + assert_eq!( + ResultMessage::read(&unknown, &keys, &restart, "wss://one.example") + .unwrap() + .outcome, + Outcome::Unknown + ); +} + +#[test] +fn receiver_rejects_wrong_owner_route_and_superseded_start() { + let keys = Keys::generate(); + let mut conn = Connection::open_in_memory().unwrap(); + let first = event(&keys, "a", true, 1); + let newer = event(&keys, "a", true, 2); + observe(&conn, &newer, &keys, "wss://one.example").unwrap(); + assert!(receive( + &mut conn, + &first, + &Keys::generate(), + "wss://one.example", + &"a".repeat(32), + true, + |_, _| panic!("owner") + ) + .is_err()); + assert!(receive( + &mut conn, + &first, + &keys, + "wss://one.example", + &"b".repeat(32), + true, + |_, _| panic!("route") + ) + .unwrap() + .is_none()); + let result = receive( + &mut conn, + &first, + &keys, + "wss://one.example", + &"a".repeat(32), + true, + |_, _| panic!("stale Start"), + ) + .unwrap() + .unwrap(); + assert_eq!( + ResultMessage::read(&result, &keys, &first, "wss://one.example") + .unwrap() + .outcome, + Outcome::Unknown + ); + let denied = receive( + &mut conn, + &newer, + &keys, + "wss://one.example", + &"a".repeat(32), + false, + |_, _| panic!("unowned"), + ) + .unwrap() + .unwrap(); + assert_eq!( + ResultMessage::read(&denied, &keys, &newer, "wss://one.example") + .unwrap() + .outcome, + Outcome::Failed + ); +} + +#[test] +fn receiver_failure_is_saved_without_reinvoking_launch() { + let keys = Keys::generate(); + let mut conn = Connection::open_in_memory().unwrap(); + let start = event(&keys, "a", true, 1); + let result = receive( + &mut conn, + &start, + &keys, + "wss://one.example", + &"a".repeat(32), + true, + |_, _| Err("native error with private path".into()), + ) + .unwrap() + .unwrap(); + assert_eq!( + ResultMessage::read(&result, &keys, &start, "wss://one.example") + .unwrap() + .outcome, + Outcome::Failed + ); + let retry = receive( + &mut conn, + &start, + &keys, + "wss://one.example", + &"a".repeat(32), + true, + |_, _| panic!("retry"), + ) + .unwrap() + .unwrap(); + assert_eq!(retry, result); +}