feat(desktop): persist ordered placement and lifecycle admission

Signed-off-by: Logan Johnson <loganj@squareup.com>

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<String, String> {
+    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<Option<Row>, 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<Option<(String, String)>, 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<bool, String> {
+    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<bool, String> {
+    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<Option<Event>, String> {
+    schema(conn)?;
+    let raw: Option<String> = 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<bool, String> {
+    Ok(row(conn, agent, "start")?.is_some())
+}
+pub(crate) fn latest_start(
+    conn: &Connection,
+    agent: &str,
+) -> Result<Option<(String, String)>, 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<bool, String> {
+    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<Outcome, String>,
+) -> Result<Option<Event>, 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))
+}

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);
+}