diff --git a/desktop/src-tauri/src/commands/desktop_lifecycle.rs b/desktop/src-tauri/src/commands/desktop_lifecycle.rs new file mode 100644 index 0000000000..a46ac7822a --- /dev/null +++ b/desktop/src-tauri/src/commands/desktop_lifecycle.rs @@ -0,0 +1,343 @@ +//! Trusted Desktop lifecycle adapter. Historical projection never launches. +use super::{ + desktop_profiles::scope, + desktop_stop::{local_id, owned_local}, +}; +use crate::{ + app_state::AppState, + managed_agents::{self, broker_launch, placement, retention::open_retention_db}, +}; +use buzz_core_pkg::{ + desktop_lifecycle::{Action, Outcome, Request, ResultMessage}, + desktop_stop::StopTarget, +}; +use nostr::{Event, JsonUtil}; +use rusqlite::{params, OptionalExtension}; +use tauri::{AppHandle, Manager}; + +#[tauri::command] +pub fn prepare_desktop_lifecycle( + app: AppHandle, + owner: String, + community: String, + desktop: String, + agent: String, + action: Action, + observed: Option, +) -> Result { + let state = app.state::(); + let scope = scope(&app, &state, &owner, &community)?; + let event = Request { + target: StopTarget { + v: 1, + community, + desktop, + agent, + }, + action, + observed, + } + .sign(&scope.owner_keys)?; + let conn = open_retention_db(&scope.db_path)?; + conn.execute_batch("CREATE TABLE IF NOT EXISTS desktop_lifecycle_outgoing (slot INTEGER PRIMARY KEY CHECK(slot=1),raw TEXT NOT NULL)").map_err(|e|e.to_string())?; + conn.execute("INSERT INTO desktop_lifecycle_outgoing VALUES(1,?1) ON CONFLICT(slot) DO UPDATE SET raw=excluded.raw",[event.as_json()]).map_err(|e|e.to_string())?; + Ok(event) +} + +/// Authenticated projection batches only; no Stop/Restart command replay. Stops +/// caused by superseded placement reuse the ordinary local lifecycle owner. +#[tauri::command] +pub async fn observe_desktop_placement( + app: AppHandle, + owner: String, + community: String, + events: Vec, + reconcile: bool, +) -> Result<(), String> { + if events.len() > 256 { + return Err("Too many placement events".into()); + } + tokio::task::spawn_blocking(move || { + let state = app.state::(); + let _transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|e| e.to_string())?; + let scope = scope(&app, &state, &owner, &community)?; + let mut conn = open_retention_db(&scope.db_path)?; + let desktop = local_id(&mut conn, &scope)?; + let mut agents = std::collections::BTreeSet::new(); + for event in events { + agents.insert(placement::observe( + &conn, + &event, + &scope.owner_keys, + &community, + )?); + } + if !reconcile { + return Ok(()); + } + placement::schema(&conn)?; + let mut query = conn + .prepare("SELECT DISTINCT agent FROM desktop_placement") + .map_err(|e| e.to_string())?; + for row in query + .query_map([], |r| r.get::<_, String>(0)) + .map_err(|e| e.to_string())? + { + agents.insert(row.map_err(|e| e.to_string())?); + } + for agent in agents { + if placement::blocked(&conn, &agent, &desktop)? + && owned_local(&app, &state, &owner, &agent)? + { + // Merely learning old Stop while no child exists performs no effect. + if generation(&app, &state, &agent, &community).map_or(true, |g| g.is_some()) { + managed_agents::stop_pair_locked(agent, community.clone(), app.clone())?; + } + } + } + Ok(()) + }) + .await + .map_err(|e| format!("Placement task failed: {e}"))? +} + +#[tauri::command] +pub fn read_desktop_placement( + app: AppHandle, + owner: String, + community: String, + agent: String, +) -> Result, String> { + let state = app.state::(); + let scope = scope(&app, &state, &owner, &community)?; + placement::latest_start(&open_retention_db(&scope.db_path)?, &agent) +} + +fn generation( + app: &AppHandle, + state: &AppState, + agent: &str, + community: &str, +) -> Result, String> { + let key = managed_agents::ManagedAgentRuntimeKey::new(agent, community)?; + let mut runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + if let Some(runtime) = runtimes.get_mut(&key) { + if runtime + .child + .try_wait() + .map_err(|e| e.to_string())? + .is_none() + { + return Ok(Some(runtime.start_nonce.clone())); + } + } + drop(runtimes); + let _store = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let records = managed_agents::load_managed_agents(app)?; + let legacy = records + .iter() + .find(|r| r.pubkey == agent) + .and_then(|r| r.runtime_pid); + let dir = managed_agents::managed_agents_base_dir(app)?.join("agent-pids"); + // A receipt may represent a surviving untracked child. Never turn a missing + // in-memory handle, unreadable receipt, or legacy live PID into Stopped. + if legacy.is_some_and(managed_agents::process_is_running) + || dir + .join(format!("{}.json", key.runtime_id())) + .try_exists() + .map_err(|e| e.to_string())? + || dir + .join(format!("{agent}.pid")) + .try_exists() + .map_err(|e| e.to_string())? + { + return Err("Local process state is untracked; use ordinary Desktop Stop".into()); + } + Ok(None) +} + +fn status( + app: &AppHandle, + conn: &rusqlite::Connection, + state: &AppState, + event: &Event, + request: &Request, +) -> Result { + conn.execute_batch("CREATE TABLE IF NOT EXISTS desktop_status_generation (id TEXT PRIMARY KEY,nonce TEXT NOT NULL,observed INTEGER NOT NULL)").map_err(|e|e.to_string())?; + let Some(nonce) = generation(app, state, &request.target.agent, &request.target.community)? + else { + return Ok(Outcome::Stopped); + }; + conn.execute( + "INSERT OR REPLACE INTO desktop_status_generation VALUES(?1,?2,?3)", + params![event.id.to_hex(), nonce, nostr::Timestamp::now().as_secs()], + ) + .map_err(|e| e.to_string())?; + conn.execute("DELETE FROM desktop_status_generation WHERE rowid NOT IN (SELECT rowid FROM desktop_status_generation ORDER BY rowid DESC LIMIT 256)",[]).map_err(|e|e.to_string())?; + Ok(Outcome::Running) +} +fn current_observation( + app: &AppHandle, + conn: &rusqlite::Connection, + state: &AppState, + request: &Request, +) -> Result { + let Some(id) = request.observed.as_deref() else { + return Ok(false); + }; + // A request cannot create this local record; only a real Status observation can. + conn.execute_batch("CREATE TABLE IF NOT EXISTS desktop_status_generation (id TEXT PRIMARY KEY,nonce TEXT NOT NULL,observed INTEGER NOT NULL)").map_err(|e|e.to_string())?; + let saved: Option<(String, u64)> = conn + .query_row( + "SELECT nonce,observed FROM desktop_status_generation WHERE id=?1", + [id], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .optional() + .map_err(|e| e.to_string())?; + Ok(match saved { + Some((nonce, stamp)) if nostr::Timestamp::now().as_secs().saturating_sub(stamp) <= 30 => { + generation(app, state, &request.target.agent, &request.target.community)?.as_deref() + == Some(nonce.as_str()) + } + _ => false, + }) +} + +#[tauri::command] +pub async fn receive_desktop_lifecycle( + app: AppHandle, + owner: String, + community: String, + event: Event, +) -> Result, String> { + tokio::task::spawn_blocking(move || { + let state = app.state::(); + let _transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|e| e.to_string())?; + let scope = scope(&app, &state, &owner, &community)?; + let request = Request::read(&event, &scope.owner_keys, &community)?; + let mut conn = open_retention_db(&scope.db_path)?; + let desktop = local_id(&mut conn, &scope)?; + let owned = owned_local(&app, &state, &owner, &request.target.agent)?; + placement::receive( + &mut conn, + &event, + &scope.owner_keys, + &community, + &desktop, + owned, + |conn, request| { + if request.action == Action::Status { + status(&app, conn, &state, &event, request) + } else { + execute(&app, &state, conn, &owner, request) + } + }, + ) + }) + .await + .map_err(|e| format!("Desktop lifecycle task failed: {e}"))? +} + +fn execute( + app: &AppHandle, + state: &AppState, + conn: &rusqlite::Connection, + owner: &str, + request: &Request, +) -> Result { + let target = &request.target; + if request.action == Action::Restart && !current_observation(app, conn, state, request)? { + return Ok(Outcome::Unknown); + } + if request.action == Action::Start + && generation(app, state, &target.agent, &target.community)?.is_some() + { + return Ok(Outcome::Running); + } + let record = { + let _store = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + managed_agents::load_managed_agents(app)? + .into_iter() + .find(|r| r.pubkey == target.agent) + .ok_or("Agent is not provisioned on this Desktop")? + }; + // Provision before destructive Restart Stop. No supported session: leave the + // existing child running and return the precise non-secret missing capability. + let broker = match broker_launch::provision( + broker_launch::LaunchScope { + owner, + community: &target.community, + agent: &target.agent, + }, + &record, + ) { + Ok(b) => b, + Err(outcome) => return Ok(outcome), + }; + if request.action == Action::Restart + && managed_agents::stop_pair_locked( + target.agent.clone(), + target.community.clone(), + app.clone(), + ) + .is_err() + { + return Ok(Outcome::Failed); + } + // Existing transition lock spans final intent check, ordinary launch and registration. + if placement::blocked(conn, &target.agent, &target.desktop)? { + return Ok(Outcome::Unknown); + } + match managed_agents::start_pair_locked( + target.agent.clone(), + target.community.clone(), + true, + None, + true, + Some(&broker), + app.clone(), + ) { + Ok(_) => Ok(Outcome::Running), + Err(_) => Ok(Outcome::Failed), + } +} + +#[tauri::command] +pub fn read_desktop_lifecycle_results( + app: AppHandle, + owner: String, + community: String, + request: Event, + events: Vec, +) -> Result { + let state = app.state::(); + let scope = scope(&app, &state, &owner, &community)?; + Request::read(&request, &scope.owner_keys, &community)?; + if events.len() > 16 { + return Err("Too many lifecycle results".into()); + } + let mut outcome = Outcome::Unknown; + for event in events { + let result = ResultMessage::read(&event, &scope.owner_keys, &request, &community)?; + if result.outcome != Outcome::Unknown { + outcome = result.outcome; + } + } + Ok(outcome) +} diff --git a/desktop/src-tauri/src/commands/desktop_stop.rs b/desktop/src-tauri/src/commands/desktop_stop.rs index 11ab7fe4e1..2c9623fb29 100644 --- a/desktop/src-tauri/src/commands/desktop_stop.rs +++ b/desktop/src-tauri/src/commands/desktop_stop.rs @@ -12,7 +12,7 @@ use nostr::{Event, JsonUtil, PublicKey}; use serde_json::{json, Value}; use tauri::{AppHandle, Manager}; -fn local_id( +pub(crate) fn local_id( conn: &mut rusqlite::Connection, scope: &managed_agents::retention::RetentionScope, ) -> Result { @@ -72,31 +72,29 @@ pub async fn receive_desktop_stop( let target = StopTarget::read(&event, &scope.owner_keys, &community)?; let mut conn = open_retention_db(&scope.db_path)?; let desktop = local_id(&mut conn, &scope)?; + managed_agents::placement::observe(&conn, &event, &scope.owner_keys, &community)?; if desktop != target.desktop { return Ok(None); } // Local possession alone is insufficient after an account switch: // verify the stored agent's owner delegation against the request author. - let owned = { - let _store = state - .managed_agents_store_lock - .lock() - .map_err(|e| e.to_string())?; - let records = managed_agents::load_managed_agents(&app)?; - records - .iter() - .find(|r| r.pubkey == target.agent) - .is_some_and(|r| { - r.backend == managed_agents::BackendKind::Local - && r.auth_tag - .as_deref() - .and_then(|tag| { - let key = PublicKey::from_hex(&target.agent).ok()?; - buzz_sdk_pkg::nip_oa::verify_auth_tag(tag, &key).ok() - }) - .is_some_and(|key| key.to_hex() == owner) - }) - }; + if let Some(raw) = remote_stop::saved_result(&conn, &event.id.to_hex())? { + let saved = Event::from_json(raw).map_err(|e| e.to_string())?; + StopResult::read(&saved, &scope.owner_keys, &event, &community)?; + return Ok(Some(saved)); + } + if managed_agents::placement::desired(&conn, &target.agent)? + .is_some_and(|(host, _)| host == desktop) + { + return StopResult { + target, + request: event.id.to_hex(), + outcome: StopOutcome::Unknown, + } + .sign(&scope.owner_keys) + .map(Some); + } + let owned = owned_local(&app, &state, &owner, &target.agent)?; remote_stop::receive( &mut conn, &event, @@ -143,3 +141,27 @@ pub fn read_desktop_stop_results( } Ok(json!(outcome)) } + +/// Local possession/profile alone never establishes owner authority. +pub(super) fn owned_local( + app: &AppHandle, + state: &AppState, + owner: &str, + agent: &str, +) -> Result { + let _store = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let records = managed_agents::load_managed_agents(app)?; + Ok(records.iter().find(|r| r.pubkey == agent).is_some_and(|r| { + r.backend == managed_agents::BackendKind::Local + && r.auth_tag + .as_deref() + .and_then(|tag| { + let key = PublicKey::from_hex(agent).ok()?; + buzz_sdk_pkg::nip_oa::verify_auth_tag(tag, &key).ok() + }) + .is_some_and(|key| key.to_hex() == owner) + })) +} diff --git a/desktop/src-tauri/src/commands/mod.rs b/desktop/src-tauri/src/commands/mod.rs index 5a5c2566ac..a0bbdd76c9 100644 --- a/desktop/src-tauri/src/commands/mod.rs +++ b/desktop/src-tauri/src/commands/mod.rs @@ -19,8 +19,9 @@ mod channel_window; mod channels; mod clipboard; mod desktop_capabilities; +mod desktop_lifecycle; mod desktop_profiles; -mod desktop_stop; +pub(crate) mod desktop_stop; mod dms; mod engrams; mod export_util; @@ -96,6 +97,7 @@ pub use channel_window::*; pub use channels::*; pub use clipboard::*; pub use desktop_capabilities::*; +pub use desktop_lifecycle::*; pub use desktop_profiles::*; pub use desktop_stop::*; pub use dms::*; diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 2eb43d1067..67b82af3dc 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -554,6 +554,11 @@ pub fn run() { get_identity, prepare_desktop_profile, prepare_desktop_stop, + prepare_desktop_lifecycle, + observe_desktop_placement, + read_desktop_placement, + receive_desktop_lifecycle, + read_desktop_lifecycle_results, receive_desktop_stop, read_desktop_stop_results, read_desktop_profiles, diff --git a/desktop/src-tauri/src/managed_agents/mod.rs b/desktop/src-tauri/src/managed_agents/mod.rs index 987e16a4f4..db259e80a7 100644 --- a/desktop/src-tauri/src/managed_agents/mod.rs +++ b/desktop/src-tauri/src/managed_agents/mod.rs @@ -28,6 +28,7 @@ pub(crate) mod parallelism; mod persona_avatars; pub(crate) mod persona_events; mod personas; +pub(crate) mod placement; #[cfg(windows)] mod process_lifecycle; pub(crate) mod readiness; diff --git a/desktop/src-tauri/src/managed_agents/remote_stop.rs b/desktop/src-tauri/src/managed_agents/remote_stop.rs index 93e3e9b5ad..55d0364900 100644 --- a/desktop/src-tauri/src/managed_agents/remote_stop.rs +++ b/desktop/src-tauri/src/managed_agents/remote_stop.rs @@ -167,6 +167,22 @@ pub(crate) fn check_launch( } let conn = connection(app, key, ¤t_owner)?; schema(&conn)?; + let scope = super::retention::RetentionScope { + db_path: scoped_retention_db_path( + &super::managed_agents_base_dir(app)?, + &key.relay_url, + ¤t_owner, + ), + relay_url: key.relay_url.clone(), + owner_keys: state.signing_keys()?, + }; + let mut local_conn = connection(app, key, ¤t_owner)?; + let host = crate::commands::desktop_stop::local_id(&mut local_conn, &scope)?; + if super::placement::blocked(&conn, &key.pubkey, &host)? + && (resume.is_none() || super::placement::has_start(&conn, &key.pubkey)?) + { + return Err("This Desktop is no longer the selected running destination. Use explicit Start on this Desktop.".into()); + } let row: Option<(String, bool)> = conn .query_row( "SELECT event_id, blocked FROM desktop_stop_fence WHERE agent=?1",